赞
踩
WebRTC是一项允许网页浏览器进行实时音视频通信的技术标准。旨在实现在浏览器之间直接进行点对点的音频、视频以及数据共享,无需安装任何插件或额外软件。
ICE全称Interactive Connectivity Establishment ,是一种交互式连接框架,他允许两个设备进行p2p通讯。
在进行p2p通讯过程中,存在诸多问题,就比如两个客户端(下文均以A,B为代称)他们如何发现对方,通讯信息是否会被防火墙拦截等等问题。为了解决p2p通讯过程中存在的阻碍,ICE采用了STUN或者TURN服务解决这些问题,而STUN,TURN也就是我们接下来需要阐述的东西
STUN全称Session Traversal Utilities for NAT,是一种帮助客户端A,B相互定位到对方的协议,同时他也能够定位路由器中在信息转发过程中的阻碍因素。
在正常运行情况下,想要通讯的客户端会优先向在互联网上部署的STUN服务询问自己的地址,然后STUN会将客户端的公网IP + 端口返回给客户端,同时告诉客户端能否在路由器的NAT下被访问。
NAT全称 Network Address Translation,是一种IP映射技术,它能够将公网地址映射到局域网地址,这样只要有一个公网IP,就能够让所有NAT下的设备与互联网通讯
如下图所示,黄圈内为内网设备,拥有内网IP,红圈则为路由器,拥有公网IP。如果内网设备想要和公网通讯,则需要利用NAT技术实现端口映射。如192.168.1.100,将82端口和47.100.12.12的97端口映射。外网设备可以将数据发送给47.100.12.12:97,然后路由器在路由给192.168.1.100:82,实现内外网访问。
NAT技术在很大程度上缓解了公网IP数量不够的情况,但NAT在实际工作中依然存在一些问题。有的路由器对于连接互联网上的设备有着非常严格的限制,这意味着STUN找到公网IP,依然存在不能建立连接的可能,这就需要TURN服务解决STUN可能存在的无法实现p2p通讯的问题。
TURN全称Traversal Using Relays around NAT ,在客户端A,B之间无法通过STUN建立p2p连接时,可以采用 TURN中继服务器进行数据转发。虽然TURN能够很好的解决因路由器的限制导致的STUN服务建立P2P连接失败的问题,但TURN服务会给服务器本身增加压力。客户端的数据沟通所产生的带宽需要由TURN所在的服务器承担,这在一定程度上增加了服务器的压力。
SDP全称Session Description Protocol ,是一种用于描述媒体信息的协议,准确来说SDP是一种数据格式,规定了媒体通讯时应该以何种形式描述媒体信息。
一般来说,媒体信息包含 以下几部分
用通俗的话来说,在建立P2P连接时,连接的客户端需要相互认识,既然需要认识对方,那就少不了自我介绍,而SDP就规定了双方该如何介绍。按照SDP规定的数据格式,双方很快就能知晓对方诸如对方是否支持我传递的数据(编解码器)等信息
SDP数据示例
sdp=v=0
o=- 3348483619658186367 2 IN IP4 127.0.0.1
s=-
t=0 0
a=group:BUNDLE 0 1
a=extmap-allow-mixed
a=msid-semantic: WMS 5862ee9a-8b26-4657-ba5c-3bc69c3c7fff
...
web RTC通讯可以分为两大模块
每个客户端可能都有一种或者几种自己会的语言,为了让通讯双方的客户端能够正常进行音视频通话,需要找到一种双方都会的语言,这个过程就是媒体协商。通过SDP描述媒体信息,客户端双方通过交换SDP数据来进行媒体协商。
当媒体协商完成后,需要进行网络协商,实现P2P沟通。网络协商所作的事情就是让客户端能够发现对方,并进行直接通讯。为此客户端需要交换网络信息,这也被成为ICE candidate。
当完成上述两个步骤后,就能够实现通讯。
值得注意的是,不论是媒体协商还是网络协商,都涉及到信息的交换,比如SDP,ICE candidate。而独立的客户端本身是不具备交换信息的功能,因此需要中间服务实现信息的交换转发,这样的服务器通常被称为信令服务器(signaling service)或者信令通道(signal channel)。本文后续将使用WebSocket技术,用Java实现信令服务器。
tip: 笔者用的是BordMix绘制的流程图,没有会员,因此导出的时候会保留水印
整个WebRTC通讯流程如上如所示。
当用户A尝试着通过WebRTC调用另外一个用户的时候,会在本地产生一个特殊的描述信息——offer,offer这个描述信息我们sdp协议规定的数据格式描述。当用户A发送offer后,对端用户B接受并创建一个回应性质的描述信息——answer,answer同样也是用sdp数据描述,这整个过程就是媒体协商。
当媒体协商完成,双方交换完sdp后,需要网络协商。用户A会通过监听事件向STUN/TURN服务发送ICE Request,STUN/TURN服务接收到请求后会告诉客户端A他的网络信息——ICE candidate,然后客户端A需要通过信令服务器将自己的candidate信息转发给客户端B。客户端B按照同样的步骤,获取自己的candidate信息并转发给A进行相互协商。
当上述步骤都完成后,即可进行音视频通讯。
RTCPeerConnection
接口表示本地计算机与远程对等方之间的 WebRTC 连接。 它提供了连接到远程对等体、维护和监视连接以及在不再需要连接时关闭连接的方法。
RTCPeerConnection
类中封装了大量关于WebRTC相关的接口,offer创建,answer创建,candidate创建等等功能都被封装到该类当中
const pc = new RTCPeerConnection( // { // bundlePolicy: "max-bundle", // rtcpMuxPolicy: "require", // iceTransportPolicy:"all",//relay 或者 // iceServers: [ // { // "urls": [ // "turn:192.168.1.100:3478?transport=udp", // "turn:192.168.1.100:3478:port?transport=tcp" // ], // "username": "fgbg", // "credential": "123456" // }, // { // "urls": [ // "stun:192.168.1.100:3478" // ] // } // ] // } );
括号中的对象即为RTCPeerConnection创建时的配置内容,本文所有的通讯内容均在本地实现,因此选择无参构造创建对象
- bundlePolicy 一般用maxbundle,都绑定到同一个传输通道
- banlanced:音频与视频轨使用各自的传输通道
- maxcompat:每个轨使用自己的传输通道
- maxbundle:都绑定到同一个传输通道
- iceTransportPolicy 指定ICE的传输策略
- relay:只使用中继候选者
- all:可以使用任何类型的候选者
- iceServers 配置ice服务器,一般都是搭建coturn服务,本文不涉及相关搭建内容,详见下方连接所示文章
创建offer
const offer = await pc.createOffer()
创建answer
const answer = await pc.createAnswer()
保存本地描述信息
pc.setLocalDescription(sessionDescription)
保存offer或者answer
保存远端描述信息
pc.setRemoteDescription(sessionDescription)
保存远端转发而来的offer或者answer
保存candidate信息
pc.addIceCandidate(candidate)
获取视频流轨道集
pc.addTrack(track, stream)
该方法让RTCPeerConnection控制本地的视频流
更加详细的写法如下
// 获取本地视频流
const localStream = await navigator.mediaDevices.getUserMedia({
audio: true,
video: true
});
// 将本地视频流添加到peerConnection中
localStream.getTracks().forEach(track => {
pc.value.addTrack(track, localStream.value);
});
ontrack
pc.value.ontrack = (event: RTCTrackEvent) => {
// 添加远程的stream
remoteVideo.srcObject = event.streams[0];
remoteStream = event.streams[0];
}
onicecandidate
rtcPeerConnection.onicecandidate = (event) => {
if (event.candidate !== null) {
sendCandidateToRemotePeer(event.candidate);
} else {
/* there are no more candidates coming during this negotiation */
}
};
前端部分采用vue3 + vite + ts(【前后端的那些事】系列文章共用同一个搭建好的环境),后端springboot + WebSocket
简单介绍什么是WebSocket。平时我们的Http通讯,数据是单向流通的且只能由浏览器发起。但WebSocket是双向的,就像跨海大桥,允许前端通知后端,后端通知前端。WebSocket使得后端拥有主动发送消息的能力。
前端创建websocket
// 创建一个新的WebSocket对象,传入WebSocket服务器的URL var socket = new WebSocket('ws://your-websocket-server-url'); // 当WebSocket连接打开时触发 socket.onopen = function(event) { console.log('Connection open!'); // 可以在这里发送数据给服务器 socket.send('Hello Server!'); }; // 当接收到消息时触发 socket.onmessage = function(event) { console.log('Received: ' + event.data); // 向后端websocket发送信息 socket.send("message"); }; // 当WebSocket连接关闭时触发 socket.onclose = function(event) { console.log('Connection closed'); };
后端创建websocket
@ServerEndpoint("/ws/{userId}") public class WebSocket { @OnOpen public void onOpen(Session session, @PathParam("userId") String userId) { // 当WebSocket连接打开时触发 } @OnClose public void onClose(@PathParam("userId") String userId) { // 当接收到消息时触发 } @OnMessage public void onMessage(String message, @PathParam("userId") String userId) { // 当WebSocket连接关闭时触发 } }
上方代码分别表示前后端最简单的WebSocket的创建方式,由上方代码可知,前后端通过websocket通信,传递的都是字符串信息。单纯的字符串所包含的信息过少,因此我们通过序列化对象,交换json数据进行前后端信息传递。
笔者规定,Message类为前后端数据沟通的类,其定义如下
@Data public class Message { private String userId; private String remoteUserId; private Object data; private String roomId; private String cmd; @Override public String toString() { return "Message{" + "userId='" + userId + '\'' + ", remoteUserId='" + remoteUserId + '\'' + ", roomId='" + roomId + '\'' + ", cmd='" + cmd + '\'' + '}'; } }
tip
- roomId: 我们以腾讯会议为例,每一场会议都有一个会议号,我们可以将会议号看作房间号,每个通讯的客户端都需要加入房间,这样才能进行通讯. 为了简化逻辑,本文只有一个房间号——“1”
- cmd: 每一条信息都可能对应着不同的处理逻辑。比如加入房间需要处理加入房间的逻辑,转发offer对应转发的逻辑
tip: 笔者用的是BordMix绘制的流程图,没有会员,因此导出的时候会保留水印
红色字,表示的是客户端方法名称
绿色字,表示的是当前处理逻辑的名称
黄色背景黑色字,表示的是服务端方法名称
前端
/src/views/webrtc.vue
<script setup lang='ts'> import { onMounted } from "vue"; import { ref } from "vue"; const localVideo = ref<HTMLVideoElement>(); const localStream = ref<MediaStream>(); const remoteVideo = ref<HTMLVideoElement>(); const remoteStream = ref<MediaStream>(); const pc = ref<RTCPeerConnection>(); const userId = ref<string>(Math.random().toString(36).substr(2)); const remoteUserId = ref<string>(); const ws = new WebSocket("ws://localhost:1000/ws2/" + userId.value); onMounted(() => { localVideo.value = document.querySelector("#localVideo"); remoteVideo.value = document.querySelector("#remoteVideo"); }) ws.onopen = (ev: Event) => { console.log("连接成功 userId = " + userId.value); } ws.onmessage = (ev: MessageEvent) => { } ws.onclose = (ev) => { console.log("连接关闭 userId = " + userId.value); } // 加入房间 const join = () => { console.log("join..."); // todo: } </script> <template> <el-button @click="join">加入</el-button> <div id="videos"> <video id="localVideo" autoplay muted playsinline>本地窗口</video> <video id="remoteVideo" autoplay playsinline>远端窗口</video> </div> </template> <style lang='scss' scoped> </style>
/src/api/webrtc.ts
// cmd export const SIGNAL_TYPE_JOIN = "join"; export const SIGNAL_TYPE_RESP_JOIN = "resp-join"; // 告知加入者对方是谁 export const SIGNAL_TYPE_LEAVE = "leave"; export const SIGNAL_TYPE_NEW_PEER = "new-peer"; export const SIGNAL_TYPE_PEER_LEAVE = "peer-leave"; export const SIGNAL_TYPE_OFFER = "offer"; export const SIGNAL_TYPE_ANSWER = "answer"; export const SIGNAL_TYPE_CANDIDATE = "candidate"; export class Message { userId: string; roomId: string; remoteUserId: string; data: any; cmd: string; constructor() { // 每位用户只允许加入房间号为"1"的房间 this.roomId = "1"; } }
后端
/model/Message.java
import lombok.Data; @Data public class Message { private String userId; private String remoteUserId; private Object data; private String roomId; private String cmd; @Override public String toString() { return "Message{" + "userId='" + userId + '\'' + ", remoteUserId='" + remoteUserId + '\'' + ", roomId='" + roomId + '\'' + ", cmd='" + cmd + '\'' + '}'; } }
/model/Client.java
import lombok.Data;
import javax.websocket.Session;
@Data
public class Client {
private String userId;
private String roomId;
private Session session;
}
/model/Constant.java
public interface Constant {
String SIGNAL_TYPE_JOIN = "join";
String SIGNAL_TYPE_RESP_JOIN = "resp-join";
String SIGNAL_TYPE_LEAVE = "leave";
String SIGNAL_TYPE_NEW_PEER = "new-peer";
String SIGNAL_TYPE_PEER_LEAVE = "peer-leave";
String SIGNAL_TYPE_OFFER = "offer";
String SIGNAL_TYPE_ANSWER = "answer";
String SIGNAL_TYPE_CANDIDATE = "candidate";
}
/config/WebSocketConfig.java
import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.web.socket.server.standard.ServerEndpointExporter; @Configuration public class WebSocketConfig { /** * 注入ServerEndpointExporter, * 这个bean会自动注册使用了@ServerEndpoint注解声明的Websocket endpoint */ @Bean public ServerEndpointExporter serverEndpointExporter() { return new ServerEndpointExporter(); } }
/websocket/WebSocket.java
import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import com.fgbg.webrtc.model.Client; import com.fgbg.webrtc.model.Message; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import javax.websocket.OnClose; import javax.websocket.OnMessage; import javax.websocket.OnOpen; import javax.websocket.Session; import javax.websocket.server.PathParam; import javax.websocket.server.ServerEndpoint; import java.util.HashSet; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import static com.fgbg.webrtc.model.Constant.*; @Component @Slf4j @ServerEndpoint("/ws2/{userId}") public class WebSocket2 { //与某个客户端的连接会话,需要通过它来给客户端发送数据 private Client client; // 存储用户 private static Map<String, Client> clientMap = new ConcurrentHashMap<>(); // 存储房间 private static Map<String, Set<String>> roomMap = new ConcurrentHashMap<>(); // 为了简化逻辑, 只有一个房间->1号房间 static { roomMap.put("1", new HashSet<String>()); } private ObjectMapper objectMapper = new ObjectMapper(); @OnOpen public void onOpen(Session session, @PathParam(value="userId")String userId) { log.info("userId = " + userId + " 加入房间1"); Client client = new Client(); client.setRoomId("1"); client.setSession(session); client.setUserId(userId); this.client = client; clientMap.put(userId, client); } @OnClose public void onClose() { String userId = client.getUserId(); clientMap.remove(userId); roomMap.get("1").remove(userId); log.info("userId = " + userId + " 退出房间1"); } @OnMessage public void onMessage(String message) { } /** * 根据远端用户id, 转发信息 */ private void sendMsgByUserId(Message msg, String remoteId) throws JsonProcessingException { Client client = clientMap.get(remoteId); client.getSession().getAsyncRemote().sendText(objectMapper.writeValueAsString(msg)); } }
按照从左到右的顺序,编写下图红框标注的功能
join
// 加入房间 const join = async () => { console.log("join..."); // 获取本地视频流 localStream.value = await navigator.mediaDevices.getUserMedia({ audio: true, video: true }); localVideo.value.srcObject = localStream.value; // 创建peerConnection pc.value = new RTCPeerConnection(); // 将本地流的控制权交给pc localStream.value.getTracks().forEach(track => { pc.value.addTrack(track, localStream.value) }); // todo: pc.onicecandidate && pc.ontrack // 加入房间(Message创建默认设置roomId = "1") const message = new Message(); message.userId = userId.value; message.cmd = SIGNAL_TYPE_JOIN; ws.send(JSON.stringify(message)); }
tip: todo部分涉及到candidate交换,这部分后续再进行编写
handleJoin
让我们将视角切换回Java后端。前端通过ws.send()
方法将数据传递给后端,后端将通过onMessage()
方法接收
我们需要让onMessage()
方法判断后端收到的数据需要按照何种逻辑处理。如果判断cmd = SIGNAL_TYPE_JOIN(“join”),那么处理加入房间逻辑,也就是handleJoin需要实现的逻辑,其需要实现的逻辑如下
@OnMessage public void onMessage(String message) throws JsonProcessingException { Message data = objectMapper.readValue(message, Message.class); if (data.getCmd().equals(SIGNAL_TYPE_JOIN)) { handleJoin(data); } } /** * 处理加入房间逻辑 * @param message 前端发送给后端的数据 */ private void handleJoin(Message message) throws JsonProcessingException { String userId = message.getUserId(); String roomId = message.getRoomId(); // 保存用户加入的房间 clientMap.get(userId).setRoomId(roomId); // 在对应房间中加入用户 roomMap.get(roomId).add(userId); int size = roomMap.get(roomId).size(); if (size == 1) { // 人数为1, 什么都不做 return; }else if (size == 2) { String remoteUserId = null; // 一个两人的房间, 只要id不是自己, 那就是remoteUserId for (String id : roomMap.get(roomId)) { if (!id.equals(userId)) { remoteUserId = id; break; } } // new-peer Message newPeerMsg = new Message(); newPeerMsg.setCmd(SIGNAL_TYPE_NEW_PEER); newPeerMsg.setRoomId(roomId); /** * 当前逻辑是由第二个用户出发的, remoteUserId是站在第二个用户的视角上获取的 * new-peer逻辑是返回给第一个用户的, 从第一个用户的视角来看, remoteUserId是他 * 的userId */ newPeerMsg.setUserId(remoteUserId); newPeerMsg.setRemoteUserId(userId); // 发送消息 (new-peer是发送给第一个用户的(用户A), 在用户B的视角下,remoteUserId是A的userId) sendMsgByUserId(newPeerMsg, remoteUserId); // resp-join Message respJoinMsg = new Message(); respJoinMsg.setCmd(SIGNAL_TYPE_RESP_JOIN); respJoinMsg.setRoomId(roomId); respJoinMsg.setUserId(userId); respJoinMsg.setRemoteUserId(remoteUserId); // 发送消息 sendMsgByUserId(respJoinMsg, userId); }else if (size > 2) { log.error("人数超过2人, 拒绝加入"); } }
测试
客户端A
客户端B
服务端
按照序号,编写下图红框标注的功能
handleRemoteNewPeer
handleRemoteNewPeer方法是通过**后端(信令服务器)**通过websocket转发而来,前端则是由ws.onmessage
接受。
onmessage方法需要根据cmd类型处理与之对应的逻辑。此处需要处理的逻辑是new-peer,前端处理该逻辑的方法是handleRemoteNewPeer,其处理逻辑如下
ws.onmessage = (ev: MessageEvent) => { const message : Message = JSON.parse(ev.data); if (message.cmd === SIGNAL_TYPE_NEW_PEER) { handleRemoteNewPeer(message); } } /** * 创建offer,设置本地offer并且发送给对端 * @param message */ const handleRemoteNewPeer = async (message : Message) => { console.log("handleRemoteNewPeer..."); // 保存remoteUserId remoteUserId.value = message.remoteUserId; // 创建offer const offer = await pc.value.createOffer(); // 保存本地offer pc.value.setLocalDescription(offer); // 发送offer const offerMsg = new Message(); offerMsg.cmd = SIGNAL_TYPE_OFFER; offerMsg.data = offer; offerMsg.remoteUserId = remoteUserId.value; offerMsg.userId = userId.value; ws.send(JSON.stringify(offerMsg)); }
handleOffer
@OnMessage public void onMessage(String message) throws JsonProcessingException { Message data = objectMapper.readValue(message, Message.class); if (data.getCmd().equals(SIGNAL_TYPE_JOIN)) { handleJoin(data); }else if (data.getCmd().equals(SIGNAL_TYPE_OFFER)) { handleOffer(data); } } /** * 转发offer * @param message */ private void handleOffer(Message message) throws JsonProcessingException { String remoteUserId = message.getRemoteUserId(); sendMsgByUserId(message, remoteUserId); }
handleResponseJoin
ws.onmessage = (ev: MessageEvent) => { const message : Message = JSON.parse(ev.data); if (message.cmd === SIGNAL_TYPE_NEW_PEER) { handleRemoteNewPeer(message); } else if (message.cmd === SIGNAL_TYPE_RESP_JOIN) { handleResponseJoin(message); } } /** * 保存remoteUserId * @param message */ const handleResponseJoin = (message : Message) => { console.log("handleResponseJoin..."); remoteUserId.value = message.remoteUserId; }
handleRemoteOffer
ws.onmessage = (ev: MessageEvent) => { const message : Message = JSON.parse(ev.data); if (message.cmd === SIGNAL_TYPE_NEW_PEER) { handleRemoteNewPeer(message); } else if (message.cmd === SIGNAL_TYPE_RESP_JOIN) { handleResponseJoin(message); } else if (message.cmd === SIGNAL_TYPE_OFFER) { handleRemoteOffer(message); } } /** * 保存远端offer, 创建answer */ const handleRemoteOffer = async (message : Message) => { console.log("handleRemoteOffer..."); // 保存远端offer pc.value.setRemoteDescription(message.data); // 创建自己的offer(answer) const answer = await pc.value.createAnswer(); // 保存自己的answer pc.value.setLocalDescription(answer); // 发送answer const answerMsg = new Message(); answerMsg.cmd = SIGNAL_TYPE_ANSWER; answerMsg.userId = userId.value; answerMsg.remoteUserId = remoteUserId.value; answerMsg.data = answer; ws.send(JSON.stringify(answerMsg)); }
测试
客户端A
客户端B
测试发现没有问题
按照序号编写方法
handleAnswer
@OnMessage public void onMessage(String message) throws JsonProcessingException { Message data = objectMapper.readValue(message, Message.class); if (data.getCmd().equals(SIGNAL_TYPE_JOIN)) { handleJoin(data); }else if (data.getCmd().equals(SIGNAL_TYPE_OFFER)) { handleOffer(data); }else if (data.getCmd().equals(SIGNAL_TYPE_ANSWER)) { handleAnswer(data); } } /** * 转发answer * @param message */ private void handleAnswer(Message message) throws JsonProcessingException { String remoteUserId = message.getRemoteUserId(); sendMsgByUserId(message, remoteUserId); }
handleRemoteAnswer
ws.onmessage = (ev: MessageEvent) => { const message : Message = JSON.parse(ev.data); if (message.cmd === SIGNAL_TYPE_NEW_PEER) { handleRemoteNewPeer(message); } else if (message.cmd === SIGNAL_TYPE_RESP_JOIN) { handleResponseJoin(message); } else if (message.cmd === SIGNAL_TYPE_OFFER) { handleRemoteOffer(message); } else if (message.cmd === SIGNAL_TYPE_ANSWER) { handleRemoteAnswer(message); } } /** * 保存远端answer * @param message */ const handleRemoteAnswer = (message : Message) => { console.log("handleRemoteAnswer..."); pc.value.setRemoteDescription(message.data); } const join = () => { // ... // todo: pc.onicecandidate && pc.ontrack pc.value.onicecandidate = (event : RTCPeerConnectionIceEvent) => { if (event.candidate) { // 发送candidate const candidateMsg = new Message(); candidateMsg.cmd = SIGNAL_TYPE_CANDIDATE; candidateMsg.data = event.candidate; candidateMsg.userId = userId.value; candidateMsg.remoteUserId = remoteUserId.value; ws.send(JSON.stringify(candidateMsg)); } else { console.log("there is no candidate availiable..."); } } pc.value.ontrack = (event : RTCTrackEvent) => { // 保存远端视频流 remoteStream.value = event.streams[0]; remoteVideo.value.srcObject = event.streams[0]; } // ... }
请注意,这里发送ice candidate是webrtc内部实现的,我们需要在创建peerConnection的时候,为pc相应事件赋值,提供回调函数。
webrtc在内部会自动判断发送candidate的时机
handleCandidate
@OnMessage public void onMessage(String message) throws JsonProcessingException { Message data = objectMapper.readValue(message, Message.class); if (data.getCmd().equals(SIGNAL_TYPE_JOIN)) { handleJoin(data); }else if (data.getCmd().equals(SIGNAL_TYPE_OFFER)) { handleOffer(data); }else if (data.getCmd().equals(SIGNAL_TYPE_ANSWER)) { handleAnswer(data); }else if (data.getCmd().equals(SIGNAL_TYPE_CANDIDATE)) { handleCandidate(data); } } /** * 转发candidate * @param data */ private void handleCandidate(Message message) throws JsonProcessingException { String remoteUserId = message.getRemoteUserId(); sendMsgByUserId(message, remoteUserId); }
handleRemoteCandidate
ws.onmessage = (ev: MessageEvent) => { const message : Message = JSON.parse(ev.data); if (message.cmd === SIGNAL_TYPE_NEW_PEER) { handleRemoteNewPeer(message); } else if (message.cmd === SIGNAL_TYPE_RESP_JOIN) { handleResponseJoin(message); } else if (message.cmd === SIGNAL_TYPE_OFFER) { handleRemoteOffer(message); } else if (message.cmd === SIGNAL_TYPE_ANSWER) { handleRemoteAnswer(message); } else if (message.cmd === SIGNAL_TYPE_CANDIDATE) { handleRemoteCandidate(message); } } /** * 处理对端发送的candidate * @param message */ const handleRemoteCandidate = (message : Message) => { console.log("handleRemoteCandidate..."); pc.value.addIceCandidate(message.data); }
测试
客户端A
客户端B
WebRTC,写完啦!!!!
这时候可能有不少同学会有疑惑,流程图不还有内容吗?实际上,剩下的内容都是在复用之前写好的方法,实际上所有的逻辑都已完成!
吐血,写了2万字
<script setup lang='ts'> import { Message, SIGNAL_TYPE_ANSWER, SIGNAL_TYPE_CANDIDATE, SIGNAL_TYPE_JOIN, SIGNAL_TYPE_NEW_PEER, SIGNAL_TYPE_OFFER, SIGNAL_TYPE_RESP_JOIN } from "@/api/webrtc"; import { use } from "echarts"; import { useId } from "element-plus"; import { off } from "process"; import { onMounted } from "vue"; import { ref } from "vue"; const localVideo = ref<HTMLVideoElement>(); const localStream = ref<MediaStream>(); const remoteVideo = ref<HTMLVideoElement>(); const remoteStream = ref<MediaStream>(); const pc = ref<RTCPeerConnection>(); const userId = ref<string>(Math.random().toString(36).substr(2)); const remoteUserId = ref<string>(); const ws = new WebSocket("ws://localhost:1000/ws2/" + userId.value); ws.onopen = (ev: Event) => { console.log("连接成功 userId = " + userId.value); } ws.onmessage = (ev: MessageEvent) => { const message : Message = JSON.parse(ev.data); if (message.cmd === SIGNAL_TYPE_NEW_PEER) { handleRemoteNewPeer(message); } else if (message.cmd === SIGNAL_TYPE_RESP_JOIN) { handleResponseJoin(message); } else if (message.cmd === SIGNAL_TYPE_OFFER) { handleRemoteOffer(message); } else if (message.cmd === SIGNAL_TYPE_ANSWER) { handleRemoteAnswer(message); } else if (message.cmd === SIGNAL_TYPE_CANDIDATE) { handleRemoteCandidate(message); } } ws.onclose = (ev) => { console.log("连接关闭 userId = " + userId.value); } /** * 处理对端发送的candidate * @param message */ const handleRemoteCandidate = (message : Message) => { console.log("handleRemoteCandidate..."); pc.value.addIceCandidate(message.data); } /** * 保存远端answer * @param message */ const handleRemoteAnswer = (message : Message) => { console.log("handleRemoteAnswer..."); pc.value.setRemoteDescription(message.data); } /** * 保存远端offer, 创建answer */ const handleRemoteOffer = async (message : Message) => { console.log("handleRemoteOffer..."); // 保存远端offer pc.value.setRemoteDescription(message.data); // 创建自己的offer(answer) const answer = await pc.value.createAnswer(); // 保存自己的answer pc.value.setLocalDescription(answer); // 发送answer const answerMsg = new Message(); answerMsg.cmd = SIGNAL_TYPE_ANSWER; answerMsg.userId = userId.value; answerMsg.remoteUserId = remoteUserId.value; answerMsg.data = answer; ws.send(JSON.stringify(answerMsg)); } /** * 保存remoteUserId * @param message */ const handleResponseJoin = (message : Message) => { console.log("handleResponseJoin..."); remoteUserId.value = message.remoteUserId; } /** * 创建offer,设置本地offer并且发送给对端 * @param message */ const handleRemoteNewPeer = async (message : Message) => { console.log("handleRemoteNewPeer..."); // 保存remoteUserId remoteUserId.value = message.remoteUserId; // 创建offer const offer = await pc.value.createOffer(); // 保存本地offer pc.value.setLocalDescription(offer); // 发送offer const offerMsg = new Message(); offerMsg.cmd = SIGNAL_TYPE_OFFER; offerMsg.data = offer; offerMsg.remoteUserId = remoteUserId.value; offerMsg.userId = userId.value; ws.send(JSON.stringify(offerMsg)); } onMounted(() => { localVideo.value = document.querySelector("#localVideo"); remoteVideo.value = document.querySelector("#remoteVideo"); }) // 加入房间 const join = async () => { console.log("join..."); // 获取本地视频流 localStream.value = await navigator.mediaDevices.getUserMedia({ audio: true, video: true }); localVideo.value.srcObject = localStream.value; // 创建peerConnection pc.value = new RTCPeerConnection(); // 将本地流的控制权交给pc localStream.value.getTracks().forEach(track => { pc.value.addTrack(track, localStream.value) }); // todo: pc.onicecandidate && pc.ontrack pc.value.onicecandidate = (event : RTCPeerConnectionIceEvent) => { if (event.candidate) { // 发送candidate const candidateMsg = new Message(); candidateMsg.cmd = SIGNAL_TYPE_CANDIDATE; candidateMsg.data = event.candidate; candidateMsg.userId = userId.value; candidateMsg.remoteUserId = remoteUserId.value; ws.send(JSON.stringify(candidateMsg)); } else { console.log("there is no candidate availiable..."); } } pc.value.ontrack = (event : RTCTrackEvent) => { // 保存远端视频流 remoteStream.value = event.streams[0]; remoteVideo.value.srcObject = event.streams[0]; } // 加入房间 const message = new Message(); message.userId = userId.value; message.cmd = SIGNAL_TYPE_JOIN; ws.send(JSON.stringify(message)); } </script> <template> <el-button @click="join">加入</el-button> <div id="videos"> <video id="localVideo" autoplay muted playsinline>本地窗口</video> <video id="remoteVideo" autoplay playsinline>远端窗口</video> </div> </template> <style lang='scss' scoped> </style>
package com.fgbg.webrtc.websocket; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import com.fgbg.webrtc.model.Client; import com.fgbg.webrtc.model.Message; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import javax.websocket.OnClose; import javax.websocket.OnMessage; import javax.websocket.OnOpen; import javax.websocket.Session; import javax.websocket.server.PathParam; import javax.websocket.server.ServerEndpoint; import java.util.HashSet; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import static com.fgbg.webrtc.model.Constant.*; @Component @Slf4j @ServerEndpoint("/ws2/{userId}") public class WebSocket2 { //与某个客户端的连接会话,需要通过它来给客户端发送数据 private Client client; // 存储用户 private static Map<String, Client> clientMap = new ConcurrentHashMap<>(); // 存储房间 private static Map<String, Set<String>> roomMap = new ConcurrentHashMap<>(); // 为了简化逻辑, 只有一个房间->1号房间 static { roomMap.put("1", new HashSet<String>()); } private ObjectMapper objectMapper = new ObjectMapper(); @OnOpen public void onOpen(Session session, @PathParam(value="userId")String userId) { log.info("userId = " + userId + " 加入房间1"); Client client = new Client(); client.setRoomId("1"); client.setSession(session); client.setUserId(userId); this.client = client; clientMap.put(userId, client); } @OnClose public void onClose() { String userId = client.getUserId(); clientMap.remove(userId); roomMap.get("1").remove(userId); log.info("userId = " + userId + " 退出房间1"); } @OnMessage public void onMessage(String message) throws JsonProcessingException { Message data = objectMapper.readValue(message, Message.class); if (data.getCmd().equals(SIGNAL_TYPE_JOIN)) { handleJoin(data); }else if (data.getCmd().equals(SIGNAL_TYPE_OFFER)) { handleOffer(data); }else if (data.getCmd().equals(SIGNAL_TYPE_ANSWER)) { handleAnswer(data); }else if (data.getCmd().equals(SIGNAL_TYPE_CANDIDATE)) { handleCandidate(data); } } /** * 转发candidate * @param data */ private void handleCandidate(Message message) throws JsonProcessingException { String remoteUserId = message.getRemoteUserId(); sendMsgByUserId(message, remoteUserId); } /** * 转发answer * @param message */ private void handleAnswer(Message message) throws JsonProcessingException { String remoteUserId = message.getRemoteUserId(); sendMsgByUserId(message, remoteUserId); } /** * 转发offer * @param message */ private void handleOffer(Message message) throws JsonProcessingException { String remoteUserId = message.getRemoteUserId(); sendMsgByUserId(message, remoteUserId); } /** * 处理加入房间逻辑 * @param message 前端发送给后端的数据 */ private void handleJoin(Message message) throws JsonProcessingException { String userId = message.getUserId(); String roomId = message.getRoomId(); // 保存用户加入的房间 clientMap.get(userId).setRoomId(roomId); // 在对应房间中加入用户 roomMap.get(roomId).add(userId); int size = roomMap.get(roomId).size(); if (size == 1) { // 人数为1, 什么都不做 return; }else if (size == 2) { String remoteUserId = null; // 一个两人的房间, 只要id不是自己, 那就是remoteUserId for (String id : roomMap.get(roomId)) { if (!id.equals(userId)) { remoteUserId = id; break; } } // new-peer Message newPeerMsg = new Message(); newPeerMsg.setCmd(SIGNAL_TYPE_NEW_PEER); newPeerMsg.setRoomId(roomId); /** * 当前逻辑是由第二个用户出发的, remoteUserId是站在第二个用户的视角上获取的 * new-peer逻辑是返回给第一个用户的, 从第一个用户的视角来看, remoteUserId是他 * 的userId */ newPeerMsg.setUserId(remoteUserId); newPeerMsg.setRemoteUserId(userId); // 发送消息 sendMsgByUserId(newPeerMsg, remoteUserId); // resp-join Message respJoinMsg = new Message(); respJoinMsg.setCmd(SIGNAL_TYPE_RESP_JOIN); respJoinMsg.setRoomId(roomId); respJoinMsg.setUserId(userId); respJoinMsg.setRemoteUserId(remoteUserId); // 发送消息 sendMsgByUserId(respJoinMsg, userId); }else if (size > 2) { log.error("人数超过2人, 拒绝加入"); } } /** * 根据远端用户id, 转发信息 */ private void sendMsgByUserId(Message msg, String remoteId) throws JsonProcessingException { Client client = clientMap.get(remoteId); client.getSession().getAsyncRemote().sendText(objectMapper.writeValueAsString(msg)); } }
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。