基于 Tokio 的实时通信服务设计：WebRTC 信令、媒体流转发与房间状态管理

一、WebRTC 服务端的两个关键角色

WebRTC 是一个 P2P 协议，浏览器之间可以直接传输音视频数据。但实际部署中，P2P 连接在很多场景下失败（对称 NAT 穿透率约 92%），此时需要 TURN 中继服务器转发媒体流。此外，P2P 无法支持多人会议（N 个端到端连接的 Mesh 拓扑在 N > 4 时带宽爆炸），需要 SFU（Selective Forwarding Unit）服务器统一管理。

WebRTC 服务端实际上承担两个独立角色：

信令服务器（Signaling Server）：帮助两个 Peer 交换 SDP（会话描述）和 ICE（交互连接建立）候选信息。在房间管理中跟踪参与者加入/离开，通过 WebSocket 推送状态变更。信令服务器的核心是消息路由和房间状态管理——对实时性要求高（延迟 < 100ms），但数据量小（每条 SDP 约 1-5KB）。

媒体服务器（Media Server）：对于 SFU 模式，接收每个发送端的媒体流，根据订阅关系选择性转发给接收端。对于 TURN 模式，在 Peer 之间中继未穿透 NAT 的媒体流。媒体服务器的核心是低延迟包转发（延迟 < 50ms），但对 CPU 的编解码能力要求高。

Tokio 同时适合这两个角色——信令服务器是典型的 I/O 密集场景，媒体服务器可以通过 tokio::net::UdpSocket 异步处理 RTP 包。

二、WebRTC 信令与房间管理架构

信令流程（简化）：

1. Client A 连接 WebSocket，发送 JoinRoom { room_id }
2. Room Manager 注册 A，推送 ParticipantJoined 给 B
3. A 创建 Offer SDP → 信令服务器转发给 B
4. B 创建 Answer SDP → 信令服务器转发给 A
5. 双方交换 ICE Candidates → 建立 P2P 或回退 TURN

房间状态管理需要处理并发——多个参与者在毫秒级时间窗口内加入、离开房间。使用 tokio::sync::RwLock 保护房间状态：读操作（查询参与者列表）使用读锁，写操作（加入/离开）使用写锁。

三、信令服务器与房间管理实现

use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use tokio::sync::{RwLock, mpsc};
use axum::{
    extract::ws::{WebSocket, Message, WebSocketUpgrade},
    extract::State,
    response::IntoResponse,
    routing::get,
    Router,
};
use serde::{Deserialize, Serialize};
use futures::{SinkExt, StreamExt};

/// 信令消息类型
#[derive(Serialize, Deserialize, Debug, Clone)]
#[serde(tag = "type")]
pub enum SignalingMessage {
    /// 加入房间
    JoinRoom { room_id: String, participant_id: String },
    /// 离开房间
    LeaveRoom { room_id: String },
    /// SDP Offer
    Offer { to: String, sdp: String },
    /// SDP Answer
    Answer { to: String, sdp: String },
    /// ICE Candidate
    IceCandidate {
        to: String,
        candidate: String,
        sdp_mid: String,
        sdp_m_line_index: u16,
    },
    /// 房间状态通知
    ParticipantJoined {
        participant_id: String,
    },
    ParticipantLeft {
        participant_id: String,
    },
    RoomState {
        participants: Vec<String>,
    },
}

/// 参与者 —— WebSocket 连接 + 元信息
pub struct Participant {
    id: String,
    /// 消息发送通道 —— 发送给该参与者的信令消息
    tx: mpsc::UnboundedSender<SignalingMessage>,
}

/// 房间
pub struct Room {
    id: String,
    /// 参与者映射
    participants: HashMap<String, Participant>,
}

/// 房间管理器 —— 全局共享状态
pub struct RoomManager {
    /// room_id → Room
    rooms: RwLock<HashMap<String, Room>>,
    /// participant_id → room_id (反向索引，用于快速查找)
    participant_rooms: RwLock<HashMap<String, String>>,
}

impl RoomManager {
    pub fn new() -> Self {
        Self {
            rooms: RwLock::new(HashMap::new()),
            participant_rooms: RwLock::new(HashMap::new()),
        }
    }

    /// 参与者加入房间
    pub async fn join_room(
        &self,
        room_id: &str,
        participant: Participant,
    ) -> Vec<Arc<mpsc::UnboundedSender<SignalingMessage>>> {
        let mut rooms = self.rooms.write().await;
        let mut participant_rooms = self.participant_rooms.write().await;
        
        let room = rooms.entry(room_id.to_string())
            .or_insert_with(|| Room {
                id: room_id.to_string(),
                participants: HashMap::new(),
            });
        
        let participant_id = participant.id.clone();
        
        // 通知房间内已有参与者：新成员加入
        let join_notification = SignalingMessage::ParticipantJoined {
            participant_id: participant_id.clone(),
        };
        
        let mut existing_txs = Vec::new();
        for (_, p) in &room.participants {
            let _ = p.tx.send(join_notification.clone());
            existing_txs.push(Arc::new(p.tx.clone()));
        }
        
        // 注册新参与者
        room.participants.insert(participant_id.clone(), participant);
        participant_rooms.insert(participant_id, room_id.to_string());
        
        existing_txs
    }

    /// 参与者离开房间
    pub async fn leave_room(&self, participant_id: &str) {
        let room_id = {
            let participant_rooms = self.participant_rooms.read().await;
            participant_rooms.get(participant_id).cloned()
        };
        
        if let Some(room_id) = room_id {
            let mut rooms = self.rooms.write().await;
            let mut participant_rooms = self.participant_rooms.write().await;
            
            if let Some(room) = rooms.get_mut(&room_id) {
                room.participants.remove(participant_id);
                
                // 通知房间内剩余参与者
                let leave_notification = SignalingMessage::ParticipantLeft {
                    participant_id: participant_id.to_string(),
                };
                for (_, p) in &room.participants {
                    let _ = p.tx.send(leave_notification.clone());
                }
                
                // 如果房间为空，清理房间
                if room.participants.is_empty() {
                    rooms.remove(&room_id);
                }
            }
            
            participant_rooms.remove(participant_id);
        }
    }

    /// 获取房间参与者列表
    pub async fn get_participants(&self, room_id: &str) -> Vec<String> {
        let rooms = self.rooms.read().await;
        rooms.get(room_id)
            .map(|r| r.participants.keys().cloned().collect())
            .unwrap_or_default()
    }

    /// 将消息转发给房间内的特定参与者
    pub async fn send_to_participant(
        &self,
        room_id: &str,
        target_id: &str,
        message: SignalingMessage,
    ) -> Result<(), SignalingError> {
        let rooms = self.rooms.read().await;
        let room = rooms.get(room_id)
            .ok_or(SignalingError::RoomNotFound)?;
        
        let participant = room.participants.get(target_id)
            .ok_or(SignalingError::ParticipantNotFound)?;
        
        participant.tx.send(message)
            .map_err(|_| SignalingError::SendFailed)
    }
}

/// 每个 WebSocket 连接的处理逻辑
async fn handle_websocket(
    ws: WebSocket,
    room_manager: Arc<RoomManager>,
    participant_id: String,
) {
    let (mut ws_tx, mut ws_rx) = ws.split();
    
    // 为当前连接创建消息通道
    let (tx, mut rx) = mpsc::unbounded_channel::<SignalingMessage>();
    
    // 后台任务：从通道接收消息 → 通过 WebSocket 发送给客户端
    let send_task = tokio::spawn(async move {
        while let Some(msg) = rx.recv().await {
            let json = serde_json::to_string(&msg).unwrap();
            if ws_tx.send(Message::Text(json.into())).await.is_err() {
                break; // 客户端断开
            }
        }
    });
    
    let mut current_room: Option<String> = None;
    
    // 处理来自客户端的消息
    while let Some(Ok(msg)) = ws_rx.next().await {
        let text = match msg {
            Message::Text(t) => t.to_string(),
            Message::Close(_) => break,
            _ => continue,
        };
        
        let signaling_msg: SignalingMessage = match serde_json::from_str(&text) {
            Ok(m) => m,
            Err(_) => continue, // 忽略无效 JSON
        };
        
        match signaling_msg.clone() {
            SignalingMessage::JoinRoom { room_id, .. } => {
                let participant = Participant {
                    id: participant_id.clone(),
                    tx: tx.clone(),
                };
                room_manager.join_room(&room_id, participant).await;
                current_room = Some(room_id);
            }
            SignalingMessage::Offer { to, .. } |
            SignalingMessage::Answer { to, .. } |
            SignalingMessage::IceCandidate { to, .. } => {
                if let Some(room_id) = &current_room {
                    // 转发给目标参与者
                    let _ = room_manager.send_to_participant(
                        room_id, &to, signaling_msg,
                    ).await;
                }
            }
            _ => {}
        }
    }
    
    // 清理：离开房间 + 取消后台任务
    if let Some(room_id) = current_room {
        room_manager.leave_room(&participant_id).await;
    }
    send_task.abort();
}

/// 构建 axum Router
pub fn build_router(room_manager: Arc<RoomManager>) -> Router {
    Router::new()
        .route("/ws/:participant_id", get(
            |ws_upgrade: WebSocketUpgrade,
             axum::extract::Path(participant_id): axum::extract::Path<String>,
             State(state): State<Arc<RoomManager>>| async move {
                ws_upgrade.on_upgrade(move |ws| {
                    handle_websocket(ws, state, participant_id)
                })
            }
        ))
        .with_state(room_manager)
}

#[derive(Debug)]
pub enum SignalingError {
    RoomNotFound,
    ParticipantNotFound,
    SendFailed,
}

关键设计决策：

• UnboundedSender 用于信令消息推送：信令消息小（< 5KB）、频率低（< 10/s），使用无界通道简化错误处理。缺点是如果 WebSocket 客户端消费慢，消息可能堆积在内存中——但信令场景下不成为问题。
• 双重索引（room → participants + participant → room）：支持两个方向的高效查找——O(1) 查找参与者所在房间，O(1) 查找房间的参与者列表。空间换时间。
• 房间为空时自动清理：避免内存泄漏——长时间运行后空房间占用的 HashMap 条目被清理。
• ws_rx.next() 循环处理消息：每个 WebSocket 连接有独立的处理 Task，利用 Tokio 的异步 I/O 并发处理数千个连接。

四、实时通信服务的适用边界与权衡

适用场景：

• 会议系统、远程协助、实时协作编辑等需要信令服务器的场景。
• 参与者 < 100 的中型会议——WebSocket 连接数和消息转发复杂度可控。
• Rust/axum 技术栈，希望信令服务与业务服务统一部署。

不适用场景：

• 超大型会议（> 1000 参与者）。此时 Mesh 信令的消息爆炸（N × N 条 Offer/Answer），需要改为 SFU 架构。
• 无需媒体转发的场景——纯 WebSocket 的信令服务可以非常简单，不需要引入 WebRTC 的复杂性。
• 必须兼容已有 WebRTC 库的项目——如需要与 mediasoup、janus-gateway 等成熟媒体服务器集成。

主要权衡：

1. 信令协议：JSON over WebSocket 是最通用的方案，但二进制协议（Protobuf/MessagePack）可以减少 50-70% 的 SDP 传输体积。
2. 房间状态的持久化：当前实现将所有状态存储在内存中。进程重启后房间信息丢失——需要从外部数据源（Redis/数据库）恢复。
3. TURN 中继的成本：TURN 服务器的带宽和 CPU 成本高（媒体流比特率 × 参与者数）。应该作为最后手段——优先尝试 STUN 直连和 SFU 转发。

五、总结

1. WebRTC 服务端包含信令服务器（SDP/ICE 交换）和媒体服务器（TURN/SFU 流转发）两个独立角色。
2. Tokio 的异步 WebSocket 处理千万级并发连接，是信令服务器的天然适合技术栈。
3. 房间管理的双重索引（room → participants + participant → room）实现 O(1) 双向查找。
4. 信令消息通道使用 UnboundedSender 简化了错误处理，适合低频率、小体积的信令场景。
5. Mesh 拓扑在 N > 4 时带宽膨胀，应切换到 SFU 模式——信令层不变，媒体层升级。