【山大会议】多人视频通话 WebRTC 工具类搭建

前言

山大会议 基于 WebRTC 技术实现多人同时在线的视频会议功能。但是 WebRTC 技术是一项针对 P2P 实现的实时通讯技术，这意味着我们无法直接使用 WebRTC 实现多人的视频会议，因此，在对 WebRTC 技术有一定程度的熟悉后，我将 WebRTC 技术封装为了一组能够支持多人在线的视频会议工具类。

系统架构

目前，要使用 WebRTC 实现支持多人的视频聊天功能，主流的架构有三种：

• Mesh
• MCU
• SFU

Mesh 架构

Mesh 架构对流量和带宽的要求极大，它本质上就是在每一个与会者之间建立起完全图网络，每个用户之间互相进行 P2P 通信。这种架构的好处是实现起来比较基础，且服务器负载较小。但是由于连接的流较多，因此对客户端的资源占用也非常大。

MCU 架构

MCU 架构是一种重后端服务器的架构，它将编码、转码、解码、混合的任务都交给了后端的 MCU 服务器。其优点是所需要的带宽少，每个用户与服务器只需要建立一条双向流即可，但是对服务器压力极高，服务器成本也会增高。

SFU 架构

SFU 是一种折中的架构，它允许用户只需上传一个流至服务器即可，由服务器对流进行转发。SFU架构看似和MCU一样都有一个中心化的服务器，但是SFU的服务器只负责转发媒体或者存储媒体；不直接做编码、转码、解码、混合这些算力要求较高的工作；SFU服务器接到RTP包后直接转发，因此SFU架构服务端压力相对较小。

考虑到服务器成本等一系列问题，我们最终选择采用 SFU 架构进行实现。

具体代码

RTC.ts

// RTC.ts
import {
     EventEmitter } from 'events';
import {
     receiverCodecs, senderCodecs } from 'Utils/Constraints';

const ices = 'stun:stun.stunprotocol.org:3478'; // INFO: 一个免费的 STUN 服务器

export interface RTCSender {
    
	pc: RTCPeerConnection;
	offerSent: boolean;
}
export interface RTCReceiver {
    
	offerSent: boolean;
	pc: RTCPeerConnection;
	id: number;
	stream?: MediaStream;
}

export default class RTC extends EventEmitter {
    
	_sender!: RTCSender;
	_receivers!: Map<number, RTCReceiver>;

	constructor(sendOnly: boolean) {
    
		super();
		if (!sendOnly) this._receivers = new Map();
	}

	getSender() {
    
		return this._sender;
	}

	getReceivers(pubId: number) {
    
		return this._receivers.get(pubId);
	}

	createSender(pubId: number, stream: MediaStream): RTCSender {
    
		let sender = {
    
			offerSent: false,
			pc: new RTCPeerConnection({
    
				iceServers: [{
     urls: ices }],
			}),
		};
		for (const track of stream.getTracks()) {
    
			sender.pc.addTrack(track);
		}

		if (localStorage.getItem('gpuAcceleration') !== 'false')
			sender.pc
				.getTransceivers()
				.find((t) => t.sender.track?.kind === 'video')
				?.setCodecPreferences(senderCodecs);

		this.emit('localstream', pubId, stream);
		this._sender = sender;
		return sender;
	}

	createReceiver(pubId: number): RTCReceiver {
    
		const _receiver = this._receivers.get(pubId);
		// INFO: 阻止重复建立接收器
		if (_receiver) return _receiver;
		try {
    
			const pc = new RTCPeerConnection({
    
				iceServers: [{
     urls: ices }],
			});

			pc.onicecandidate = (e) => {
    
				// console.log(`receiver.pc.onicecandidate => ${e.candidate}`);
			};

			// 添加收发器

			pc.addTransceiver('audio', {
     direction: 'recvonly' });
			pc.addTransceiver('video', {
     direction: 'recvonly' });

			pc.ontrack = (e) => {
    
				if (localStorage.getItem('gpuAcceleration') !== 'false')
					pc.getTransceivers()
						.find((t) => t.receiver.track.kind === 'video')
						?.setCodecPreferences(receiverCodecs);

				// console.log(`ontrack`);
				const receiver = this._receivers.get(pubId) as RTCReceiver;
				if (!receiver.stream) {
    
					receiver.stream = new MediaStream();
					// console.log(`receiver.pc.onaddtrack => ${receiver.stream.id}`);
					this.emit('addtrack', pubId, receiver.stream);
				}
				receiver.stream.addTrack(e.track);
			};

			let receiver = {
    
				offerSent: false,
				pc: pc,
				id: pubId,
				stream: undefined,
			};

			// console.log(`createReceiver::id => ${pubId}`);
			this._receivers.set(pubId, receiver);
			return receiver;
		} catch (e) {
    
			// console.log(e);
			throw e;
		}
	}

	closeReceiver(pubId: number) {
    
		const receiver = this._receivers.get(pubId);
		if (receiver) {
    
			this.emit('removestream', pubId, receiver.stream);
			receiver.pc.close();
			this._receivers.delete(pubId);
		}
	}
}

SFU.ts

// SFU.ts
import {
     EventEmitter } from 'events';
import {
     globalMessage } from 'Utils/GlobalMessage/GlobalMessage';
import RTC, {
     RTCSender } from './RTC';

export default class SFU extends EventEmitter {
    
	_rtc: RTC;
	userId: number;
	userName: string;
	meetingId: number;
	socket: WebSocket;
	sender!: RTCSender;
	sfuIp: string;
	sendOnly: boolean;

	constructor(sfuIp: string, userId: number, userName: string, meetingId: string) {
    
		super();

		// this.sendOnly = false;
		this.sendOnly = userId < 0;

		this._rtc = new RTC(this.sendOnly);
		this.userId = userId;
		this.userName = userName;
		this.meetingId = Number(meetingId);

		// const sfuUrl = 'ws://localhost:3000/ws';
		// const sfuUrl = 'ws://webrtc.aiolia.top:3000/ws';
		// const sfuUrl = 'ws://121.40.95.78:3000/ws';
		// TOFIX: 巩义的代码有问题，会返回 127.0.0.1
		this.sfuIp = sfuIp === '127.0.0.1:3000' ? '121.40.95.78:3000' : sfuIp;
		console.log(this.sfuIp);
		const sfuUrl = `ws://${
      this.sfuIp}/ws`;

		this.socket = new WebSocket(sfuUrl);

		this.socket.onopen = () => {
    
			// console.log('WebSocket连接成功...');
			this._onRoomConnect();
		};

		this.socket.onmessage = (e) => {
    
			const parseMessage = JSON.parse(e.data);
			// if (parseMessage && parseMessage.type !== 'heartPackage') console.log(parseMessage);
			switch (parseMessage.type) {
    
				case 'newUser':
					this.onNewMemberJoin(parseMessage);
					break;
				case 'joinSuccess':
					// console.log(parseMessage);
					this.onJoinSuccess(parseMessage);
					break;
				case 'publishSuccess':
					// 这里是接到有人推流的信息
					this.onPublish(parseMessage);
					break;
				case 'userLeave':
					// 这里是有人停止推流
					if (!this.sendOnly) this.onUnpublish(parseMessage);
					break;
				case 'subscribeSuccess':
					// 这里是加入会议后接到已推流的消息进行订阅
					this.onSubscribe(parseMessage);
					break;
				case 'chatSuccess':
					this.emit('onChatMessage', parseMessage.data);
					break;
				case 'heartPackage':
					// 心跳包
					// console.log('heartPackage:::');
					break;
				case 'requestError':
					globalMessage.error(`服务器错误: ${
      parseMessage.data}`);
					break;
				default:
					console.error('未知消息', parseMessage);
			}
		};

		this.socket.onerror = (e) => {
    
			// console.log('onerror::');
			console.warn(e);
			this.emit('error');
		};

		this.socket.onclose = (e) => {
    
			// console.log('onclose::');
			console.warn(e);
		};
	}

	_onRoomConnect = () => {
    
		// console.log('onRoomConnect');

		this._rtc.on('localstream', (id, stream) => {
    
			this.emit('addLocalStream', id, stream);
		});

		this._rtc.on('addtrack', (id, stream) => {
    
			if (id < 0 && id !== -this.userId) {
    
				this.emit('addScreenShare', id, stream);
			} else {
    
				this.emit('addRemoteStream', id, stream);
			}
		});

		this.emit('connect');
	};

	join() {
    
		// console.log(`Join to [${this.meetingId}] as [${this.userName}:${this.userId}]`);
		let message = {
    
			type: 'join',
			data: {
    
				userName: this.userName,
				userId: this.userId,
				meetingId: this.meetingId,
			},
		};
		this.send(message);
	}

	// 新成员入会
	onNewMemberJoin(message: any) {
    
		this.emit('onNewMemberJoin', message.data.newUserInfo);
	}

	// 成功加入会议
	onJoinSuccess(message: any) {
    
		this.emit('onJoinSuccess', message.data.allUserInfos);
		if (this.sendOnly) return;
		for (const pubId of message.data.pubIds) {
    
			console.log(`${
      this.userId} 准备接收 ${
      pubId}`);
			this._onRtcCreateReceiver(pubId);
		}
	}

	send(data: any) {
    
		this.socket.send(JSON.stringify(data));
	}

	publish(stream: MediaStream) {
    
		this._createSender(this.userId, stream);
	}

	_createSender(pubId: number, stream: MediaStream) {
    
		try {
    
			// 创建一个sender
			let sender = this._rtc.createSender(pubId, stream);
			this.sender = sender;
			// 监听IceCandidate回调
			sender.pc.onicecandidate = async (e) => {
    
				if (!sender.offerSent) {
    
					const offer = sender.pc.localDescription;
					sender.offerSent = true;
					this.publishToServer(offer, pubId);
				}
			};
			// 创建Offer
			sender.pc
				.createOffer({
    
					offerToReceiveVideo: false,
					offerToReceiveAudio: false,
				})
				.then((desc) => {
    
					sender.pc.setLocalDescription(desc);
				});
		} catch (error) {
    
			// console.log('onCreateSender error =>' + error);
		}
	}

	publishToServer(offer: RTCSessionDescription | null, pubId: number) {
    
		let message = {
    
			type: 'publish',
			data: {
    
				jsep: offer,
				pubId,
				userId: this.userId,
				meetingId: this.meetingId,
			},
		};
		// console.log('===publish===');
		// console.log(message);
		this.send(message);
	}

	onPublish(message: any) {
    
		const pubId = message['data']['pubId'];

		// 服务器返回的Answer信息 如A ---> Offer---> SFU---> Answer ---> A
		if (this.sender && pubId === this.userId) {
    
			// console.log('onPublish:::自已发布的Id:::' + message['data']['pubId']);
			this.sender.pc.setRemoteDescription(message['data']['jsep']);
			return;
		}

		if (this.userId > 0 && pubId !== this.userId && pubId !== -this.userId) {
    
			// 服务器返回其他人发布的信息 如 A ---> Pub ---> SFU ---> B
			// console.log('onPublish:::其他人发布的Id:::' + pubId);
			// 使用发布者的userId创建Receiver
			this._onRtcCreateReceiver(pubId);
		}
	}

	onUnpublish(message: any) {
    
		// console.log('退出用户:' + message['data']['leaverId']);
		const leaverId = message['data']['leaverId'];

		this._rtc.closeReceiver(leaverId);
		if (leaverId > 0) {
    
			this.emit('removeRemoteStream', leaverId);
		} else {
    
			this.emit('removeScreenShare', leaverId);
		}
	}

	_onRtcCreateReceiver(pubId: number) {
    
		try {
    
			let receiver = this._rtc.createReceiver(pubId);

			receiver.pc.onicecandidate = () => {
    
				if (!receiver.offerSent) {
    
					const offer = receiver.pc.localDescription;
					receiver.offerSent = true;
					this.subscribeFromServer(offer, pubId);
				}
			};
			// 创建Offer
			receiver.pc.createOffer().then((desc) => {
    
				receiver.pc.setLocalDescription(desc);
			});
		} catch (error) {
    
			// console.log('onRtcCreateReceiver error =>' + error);
		}
	}

	subscribeFromServer(offer: RTCSessionDescription | null, pubId: number) {
    
		let message = {
    
			type: 'subscribe',
			data: {
    
				jsep: offer,
				pubId,
				userId: this.userId,
				meetingId: this.meetingId,
			},
		};

		// console.log('===subscribe===');
		// console.log(message);
		this.send(message);
	}

	onSubscribe(message: any) {
    
		// 使用发布者的Id获取Receiver
		const receiver = this._rtc.getReceivers(message['data']['pubId']);
		if (receiver) {
    
			// console.log('服务器应答Id:' + message['data']['pubId']);
			if (receiver.pc.remoteDescription) {
    
				console.warn('已建立远程连接！');
			} else {
    
				receiver.pc.setRemoteDescription(message['data']['jsep']);
			}
		} else {
    
			// console.log('receiver == null');
		}
	}
}