构建一个完整的 WebRTC 通信系统 (架构篇)

构建一个完整的 WebRTC 通信系统 (架构篇)

本文是 WebRTC 系列专栏的第二十九篇,也是本系列的收官之作。我们将从架构角度,系统性地讲解如何构建一个端到端的 WebRTC 通信系统。

目录

1. 系统架构概述

1.1 整体架构

完整 WebRTC 系统架构:

+——————+
| 负载均衡器 |
+——–+———+
|
+——————–+——————–+
| | |
+——-v——-+ +——-v——-+ +——-v——-+
| 信令服务器 | | API 服务器 | | Web 服务器 |
| (WebSocket) | | (REST) | | (静态资源) |
+——-+——-+ +——-+——-+ +—————+
| |
| +——-v——-+
| | 数据库 |
| | (用户/房间) |
| +—————+
|
+——-v——-+
| 消息队列 |
| (Redis) |
+——-+——-+
|
+——-v——-+ +—————+
| SFU 集群 | | TURN/STUN |
| (mediasoup) |<–>| (coturn) |
+—————+ +—————+
^
|
+——-+——-+
| 客户端 |
| Web/iOS/Android|
+—————+

1.2 组件职责

1.3 数据流

数据流示意:

1. 用户认证流
客户端 -> API 服务器 -> 数据库 -> JWT Token

2. 信令流
客户端 -> 信令服务器 -> Redis -> 其他信令服务器 -> 客户端

3. 媒体流
客户端 -> TURN(可选) -> SFU -> TURN(可选) -> 客户端

4. 控制流
信令服务器 -> SFU (创建房间/传输)

2. 信令服务器

2.1 架构设计

// signaling-server/src/index.js
const express = require('express');
const http = require('http');
const WebSocket = require('ws');
const Redis = require('ioredis');
const jwt = require('jsonwebtoken');

const app = express();
const server = http.createServer(app);
const wss = new WebSocket.Server({ server });

// Redis 用于集群通信
const redis = new Redis(process.env.REDIS_URL);
const redisSub = new Redis(process.env.REDIS_URL);

// 本地连接管理
const connections = new Map();

// 订阅 Redis 消息
redisSub.subscribe('signaling');
redisSub.on('message', (channel, message) => {
const data = JSON.parse(message);
handleRedisMessage(data);
});

// WebSocket 连接处理
wss.on('connection', async (ws, req) => {
// 验证 JWT
const token = req.url.split('token=')[1];
try {
const decoded = jwt.verify(token, process.env.JWT_SECRET);
ws.userId = decoded.userId;
connections.set(ws.userId, ws);

ws.on('message', (data) => handleMessage(ws, data));
ws.on('close', () => handleDisconnect(ws));
} catch (error) {
ws.close(4001, 'Unauthorized');
}
});

// 消息处理
async function handleMessage(ws, data) {
const message = JSON.parse(data);

switch (message.type) {
case 'join-room':
await handleJoinRoom(ws, message);
break;
case 'leave-room':
await handleLeaveRoom(ws, message);
break;
case 'offer':
case 'answer':
case 'candidate':
await forwardToUser(message.targetId, message);
break;
}
}

// 跨服务器转发
async function forwardToUser(userId, message) {
const localWs = connections.get(userId);

if (localWs) {
localWs.send(JSON.stringify(message));
} else {
// 通过 Redis 转发到其他服务器
redis.publish('signaling', JSON.stringify({
targetUserId: userId,
message
}));
}
}

server.listen(8080);

2.2 房间管理

// signaling-server/src/RoomManager.js
class RoomManager {
constructor(redis) {
this.redis = redis;
}

async createRoom(roomId, options = {}) {
const room = {
id: roomId,
createdAt: Date.now(),
maxParticipants: options.maxParticipants || 10,
sfuEndpoint: null,
participants: []
};

await this.redis.hset('rooms', roomId, JSON.stringify(room));
return room;
}

async joinRoom(roomId, userId) {
let room = await this.getRoom(roomId);

if (!room) {
room = await this.createRoom(roomId);
}

if (room.participants.length >= room.maxParticipants) {
throw new Error('Room is full');
}

room.participants.push(userId);
await this.redis.hset('rooms', roomId, JSON.stringify(room));

// 分配 SFU
if (!room.sfuEndpoint) {
room.sfuEndpoint = await this.allocateSfu(roomId);
}

return room;
}

async leaveRoom(roomId, userId) {
const room = await this.getRoom(roomId);
if (!room) return;

room.participants = room.participants.filter(id => id !== userId);

if (room.participants.length === 0) {
await this.redis.hdel('rooms', roomId);
} else {
await this.redis.hset('rooms', roomId, JSON.stringify(room));
}
}

async getRoom(roomId) {
const data = await this.redis.hget('rooms', roomId);
return data ? JSON.parse(data) : null;
}

async allocateSfu(roomId) {
// 从 SFU 集群中选择负载最低的
const sfuList = await this.redis.smembers('sfu-nodes');
let bestSfu = null;
let minLoad = Infinity;

for (const sfu of sfuList) {
const load = await this.redis.get(`sfu-load:${sfu}`);
if (parseInt(load) < minLoad) {
minLoad = parseInt(load);
bestSfu = sfu;
}
}

return bestSfu;
}
}

module.exports = RoomManager;

3. TURN/STUN 服务

3.1 coturn 配置

# /etc/turnserver.conf

# 基本配置
listening-port=3478
tls-listening-port=5349
listening-ip=0.0.0.0
relay-ip=<公网IP>
external-ip=<公网IP>

# 认证
lt-cred-mech
use-auth-secret
static-auth-secret=<长随机字符串>

# TLS 证书
cert=/etc/letsencrypt/live/turn.example.com/fullchain.pem
pkey=/etc/letsencrypt/live/turn.example.com/privkey.pem

# 日志
log-file=/var/log/turnserver.log
verbose

# 安全
no-multicast-peers
denied-peer-ip=10.0.0.0-10.255.255.255
denied-peer-ip=192.168.0.0-192.168.255.255
denied-peer-ip=172.16.0.0-172.31.255.255

# 性能
total-quota=100
bps-capacity=0
max-bps=3000000

3.2 动态凭证生成

// api-server/src/turnCredentials.js
const crypto = require('crypto');

class TurnCredentialGenerator {
constructor(secret, ttl = 86400) {
this.secret = secret;
this.ttl = ttl;
}

generate(userId) {
const timestamp = Math.floor(Date.now() / 1000) + this.ttl;
const username = `${timestamp}:${userId}`;

const hmac = crypto.createHmac('sha1', this.secret);
hmac.update(username);
const credential = hmac.digest('base64');

return {
urls: [
'stun:turn.example.com:3478',
'turn:turn.example.com:3478',
'turns:turn.example.com:5349'
],
username,
credential,
ttl: this.ttl
};
}
}

// API 端点
app.get('/api/turn-credentials', authenticate, (req, res) => {
const generator = new TurnCredentialGenerator(process.env.TURN_SECRET);
const credentials = generator.generate(req.userId);
res.json(credentials);
});

4. SFU 媒体服务器

4.1 mediasoup 服务

// sfu-server/src/index.js
const mediasoup = require('mediasoup');
const express = require('express');
const https = require('https');
const WebSocket = require('ws');

const app = express();
const server = https.createServer(sslOptions, app);
const wss = new WebSocket.Server({ server });

// mediasoup 配置
const config = {
worker: {
rtcMinPort: 10000,
rtcMaxPort: 10100,
logLevel: 'warn'
},
router: {
mediaCodecs: [
{
kind: 'audio',
mimeType: 'audio/opus',
clockRate: 48000,
channels: 2
},
{
kind: 'video',
mimeType: 'video/VP8',
clockRate: 90000
},
{
kind: 'video',
mimeType: 'video/VP9',
clockRate: 90000
}
]
},
webRtcTransport: {
listenIps: [
{ ip: '0.0.0.0', announcedIp: process.env.PUBLIC_IP }
],
initialAvailableOutgoingBitrate: 1000000,
maxIncomingBitrate: 1500000
}
};

// Worker 池
const workers = [];
let nextWorkerIndex = 0;

async function createWorkers() {
const numWorkers = require('os').cpus().length;

for (let i = 0; i < numWorkers; i++) {
const worker = await mediasoup.createWorker(config.worker);
workers.push(worker);

worker.on('died', () => {
console.error(`Worker ${i} died`);
process.exit(1);
});
}
}

function getNextWorker() {
const worker = workers[nextWorkerIndex];
nextWorkerIndex = (nextWorkerIndex + 1) % workers.length;
return worker;
}

// 房间管理
const rooms = new Map();

class Room {
constructor(id, router) {
this.id = id;
this.router = router;
this.peers = new Map();
}

async addPeer(peerId) {
const peer = {
id: peerId,
transports: new Map(),
producers: new Map(),
consumers: new Map()
};
this.peers.set(peerId, peer);
return peer;
}

async createWebRtcTransport(peerId) {
const transport = await this.router.createWebRtcTransport(
config.webRtcTransport
);

const peer = this.peers.get(peerId);
peer.transports.set(transport.id, transport);

return {
id: transport.id,
iceParameters: transport.iceParameters,
iceCandidates: transport.iceCandidates,
dtlsParameters: transport.dtlsParameters
};
}

async produce(peerId, transportId, kind, rtpParameters) {
const peer = this.peers.get(peerId);
const transport = peer.transports.get(transportId);

const producer = await transport.produce({ kind, rtpParameters });
peer.producers.set(producer.id, producer);

return producer;
}

async consume(peerId, producerId, rtpCapabilities) {
const peer = this.peers.get(peerId);

if (!this.router.canConsume({ producerId, rtpCapabilities })) {
throw new Error('Cannot consume');
}

const transport = Array.from(peer.transports.values())
.find(t => t.appData.consuming);

const consumer = await transport.consume({
producerId,
rtpCapabilities,
paused: true
});

peer.consumers.set(consumer.id, consumer);

return {
id: consumer.id,
producerId,
kind: consumer.kind,
rtpParameters: consumer.rtpParameters
};
}
}

// 创建房间
async function getOrCreateRoom(roomId) {
let room = rooms.get(roomId);

if (!room) {
const worker = getNextWorker();
const router = await worker.createRouter({
mediaCodecs: config.router.mediaCodecs
});

room = new Room(roomId, router);
rooms.set(roomId, room);
}

return room;
}

// 启动
createWorkers().then(() => {
server.listen(4443, () => {
console.log('SFU server running on port 4443');
});
});

5. Web 客户端

5.1 客户端架构

// web-client/src/WebRTCClient.js
class WebRTCClient {
constructor(config) {
this.config = config;
this.signaling = null;
this.device = null;
this.sendTransport = null;
this.recvTransport = null;
this.producers = new Map();
this.consumers = new Map();
}

async connect(token) {
// 获取 TURN 凭证
const turnCredentials = await this.fetchTurnCredentials(token);

// 连接信令服务器
this.signaling = new SignalingClient(
this.config.signalingUrl,
token
);
await this.signaling.connect();

// 设置事件处理
this.setupSignalingHandlers();
}

async joinRoom(roomId) {
// 加入房间
const { rtpCapabilities, sfuEndpoint } = await this.signaling.joinRoom(roomId);

// 加载 mediasoup Device
this.device = new mediasoupClient.Device();
await this.device.load({ routerRtpCapabilities: rtpCapabilities });

// 创建传输
await this.createTransports(sfuEndpoint);

// 获取已有的 producers
const producers = await this.signaling.getProducers();
for (const producer of producers) {
await this.consume(producer.producerId);
}
}

async publish(stream) {
for (const track of stream.getTracks()) {
const producer = await this.sendTransport.produce({
track,
encodings: track.kind === 'video' ? [
{ maxBitrate: 100000, scaleResolutionDownBy: 4 },
{ maxBitrate: 300000, scaleResolutionDownBy: 2 },
{ maxBitrate: 900000, scaleResolutionDownBy: 1 }
] : undefined
});

this.producers.set(producer.id, producer);
}
}

async consume(producerId) {
const consumerInfo = await this.signaling.consume({
producerId,
rtpCapabilities: this.device.rtpCapabilities
});

const consumer = await this.recvTransport.consume(consumerInfo);
this.consumers.set(consumer.id, consumer);

await this.signaling.resumeConsumer(consumer.id);

return consumer.track;
}

setupSignalingHandlers() {
this.signaling.on('new-producer', async ({ producerId }) => {
const track = await this.consume(producerId);
this.emit('track', track);
});

this.signaling.on('producer-closed', ({ producerId }) => {
// 处理 producer 关闭
});
}
}

5.2 UI 组件

// web-client/src/components/VideoRoom.jsx
import React, { useEffect, useRef, useState } from 'react';
import { WebRTCClient } from '../WebRTCClient';

function VideoRoom({ roomId, token }) {
const [participants, setParticipants] = useState([]);
const localVideoRef = useRef(null);
const clientRef = useRef(null);

useEffect(() => {
const client = new WebRTCClient(config);
clientRef.current = client;

async function init() {
await client.connect(token);

// 获取本地媒体
const stream = await navigator.mediaDevices.getUserMedia({
video: true,
audio: true
});
localVideoRef.current.srcObject = stream;

// 加入房间
await client.joinRoom(roomId);

// 发布本地流
await client.publish(stream);

// 监听新轨道
client.on('track', (track, peerId) => {
setParticipants(prev => […prev, { peerId, track }]);
});
}

init();

return () => {
client.disconnect();
};
}, [roomId, token]);

return (
<div className="video-room">
<div className="local-video">
<video ref={localVideoRef} autoPlay muted playsInline />
</div>
<div className="remote-videos">
{participants.map(p => (
<RemoteVideo key={p.peerId} track={p.track} />
))}
</div>
<div className="controls">
<button onClick={() => toggleAudio()}>Mute</button>
<button onClick={() => toggleVideo()}>Camera</button>
<button onClick={() => leaveRoom()}>Leave</button>
</div>
</div>
);
}

function RemoteVideo({ track }) {
const videoRef = useRef(null);

useEffect(() => {
if (videoRef.current && track) {
const stream = new MediaStream([track]);
videoRef.current.srcObject = stream;
}
}, [track]);

return <video ref={videoRef} autoPlay playsInline />;
}

6. 移动端客户端

6.1 Android 集成

// android/app/src/main/java/com/example/webrtc/WebRTCManager.kt
class WebRTCManager(
private val context: Context,
private val config: Config
) {
private var peerConnectionFactory: PeerConnectionFactory? = null
private var signaling: SignalingClient? = null
private val transports = mutableMapOf<String, Transport>()

fun initialize() {
val options = PeerConnectionFactory.InitializationOptions
.builder(context)
.createInitializationOptions()
PeerConnectionFactory.initialize(options)

peerConnectionFactory = PeerConnectionFactory.builder()
.setVideoEncoderFactory(DefaultVideoEncoderFactory(
EglBase.create().eglBaseContext, true, true
))
.setVideoDecoderFactory(DefaultVideoDecoderFactory(
EglBase.create().eglBaseContext
))
.createPeerConnectionFactory()
}

suspend fun connect(token: String) {
signaling = SignalingClient(config.signalingUrl, token)
signaling?.connect()
}

suspend fun joinRoom(roomId: String): RoomInfo {
return signaling?.joinRoom(roomId) ?: throw Exception("Not connected")
}

suspend fun publish(videoTrack: VideoTrack, audioTrack: AudioTrack) {
// 创建 Producer
}

suspend fun subscribe(producerId: String): MediaStreamTrack {
// 创建 Consumer
}
}

6.2 iOS 集成

// ios/WebRTCManager.swift
class WebRTCManager {
private var peerConnectionFactory: RTCPeerConnectionFactory!
private var signaling: SignalingClient?

func initialize() {
RTCInitializeSSL()

let encoderFactory = RTCDefaultVideoEncoderFactory()
let decoderFactory = RTCDefaultVideoDecoderFactory()

peerConnectionFactory = RTCPeerConnectionFactory(
encoderFactory: encoderFactory,
decoderFactory: decoderFactory
)
}

func connect(token: String) async throws {
signaling = SignalingClient(url: config.signalingUrl, token: token)
try await signaling?.connect()
}

func joinRoom(roomId: String) async throws -> RoomInfo {
guard let signaling = signaling else {
throw WebRTCError.notConnected
}
return try await signaling.joinRoom(roomId)
}
}

7. 业务逻辑层

7.1 用户认证

// api-server/src/auth.js
const jwt = require('jsonwebtoken');
const bcrypt = require('bcrypt');

class AuthService {
constructor(db) {
this.db = db;
}

async register(email, password, name) {
const hashedPassword = await bcrypt.hash(password, 10);

const user = await this.db.user.create({
data: {
email,
password: hashedPassword,
name
}
});

return this.generateToken(user);
}

async login(email, password) {
const user = await this.db.user.findUnique({
where: { email }
});

if (!user || !await bcrypt.compare(password, user.password)) {
throw new Error('Invalid credentials');
}

return this.generateToken(user);
}

generateToken(user) {
return jwt.sign(
{ userId: user.id, email: user.email },
process.env.JWT_SECRET,
{ expiresIn: '7d' }
);
}

verifyToken(token) {
return jwt.verify(token, process.env.JWT_SECRET);
}
}

7.2 房间权限

// api-server/src/roomPermissions.js
class RoomPermissionService {
constructor(db) {
this.db = db;
}

async canJoinRoom(userId, roomId) {
const room = await this.db.room.findUnique({
where: { id: roomId },
include: { members: true }
});

if (!room) return true; // 房间不存在,可以创建

if (room.isPrivate) {
return room.members.some(m => m.userId === userId);
}

return room.members.length < room.maxParticipants;
}

async canPublish(userId, roomId) {
const membership = await this.db.roomMember.findUnique({
where: {
roomId_userId: { roomId, userId }
}
});

return membership?.role === 'host' || membership?.role === 'speaker';
}

async setRole(roomId, userId, role) {
await this.db.roomMember.update({
where: {
roomId_userId: { roomId, userId }
},
data: { role }
});
}
}

8. 部署与运维

8.1 Docker Compose

# docker-compose.yml
version: '3.8'

services:
signaling:
build: ./signaling–server
ports:
– "8080:8080"
environment:
– REDIS_URL=redis://redis:6379
– JWT_SECRET=${JWT_SECRET}
depends_on:
– redis
deploy:
replicas: 2

api:
build: ./api–server
ports:
– "3000:3000"
environment:
– DATABASE_URL=${DATABASE_URL}
– JWT_SECRET=${JWT_SECRET}
– TURN_SECRET=${TURN_SECRET}
depends_on:
– postgres

sfu:
build: ./sfu–server
ports:
– "4443:4443"
– "10000-10100:10000-10100/udp"
environment:
– PUBLIC_IP=${PUBLIC_IP}
– REDIS_URL=redis://redis:6379
deploy:
replicas: 2

turn:
image: coturn/coturn
ports:
– "3478:3478/udp"
– "3478:3478/tcp"
– "5349:5349/tcp"
– "49152-65535:49152-65535/udp"
volumes:
– ./turnserver.conf:/etc/turnserver.conf
command: –c /etc/turnserver.conf

redis:
image: redis:7–alpine
ports:
– "6379:6379"

postgres:
image: postgres:15–alpine
environment:
– POSTGRES_DB=webrtc
– POSTGRES_USER=${DB_USER}
– POSTGRES_PASSWORD=${DB_PASSWORD}
volumes:
– postgres_data:/var/lib/postgresql/data

nginx:
image: nginx:alpine
ports:
– "80:80"
– "443:443"
volumes:
– ./nginx.conf:/etc/nginx/nginx.conf
– ./certs:/etc/nginx/certs
depends_on:
– signaling
– api

volumes:
postgres_data:

8.2 监控告警

// monitoring/metrics.js
const prometheus = require('prom-client');

// 定义指标
const activeConnections = new prometheus.Gauge({
name: 'webrtc_active_connections',
help: 'Number of active WebRTC connections'
});

const roomCount = new prometheus.Gauge({
name: 'webrtc_room_count',
help: 'Number of active rooms'
});

const mediaBitrate = new prometheus.Histogram({
name: 'webrtc_media_bitrate',
help: 'Media bitrate distribution',
buckets: [100000, 500000, 1000000, 2000000, 5000000]
});

const packetLoss = new prometheus.Histogram({
name: 'webrtc_packet_loss',
help: 'Packet loss rate distribution',
buckets: [0.01, 0.02, 0.05, 0.1, 0.2]
});

// 导出指标端点
app.get('/metrics', async (req, res) => {
res.set('Content-Type', prometheus.register.contentType);
res.end(await prometheus.register.metrics());
});

9. 总结

9.1 系统组件总结

9.2 扩展建议

系统扩展方向:

1. 功能扩展
– 屏幕共享
– 录制回放
– 实时字幕
– 虚拟背景

2. 性能扩展
– SFU 集群
– 地理分布部署
– CDN 加速

3. 运维扩展
– 自动扩缩容
– 故障自愈
– 全链路监控

9.3 系列总结

恭喜你完成了 WebRTC 技术专栏的全部 29 篇文章!

本系列涵盖了:

• Part 1-2: 基础概念与信令 (11 篇)
• Part 3: 媒体传输 (6 篇)
• Part 4: 编码与处理 (3 篇)
• Part 5: 工程实践 (5 篇)
• Part 6: 高级主题 (4 篇)

希望这个系列能帮助你全面掌握 WebRTC 技术,构建出色的实时通信应用!

参考资料