WebRTC实现多人视频聊天之信令服务器设计

基础设想

信令服务器主要负责转发 SDP ，当然，我也可以选择将我的业务逻辑写在里面（生产环境不推荐这样做）。选择基于Web Sockets主要是因为这块比较熟悉，并且双向连接，我们能够基于此做的事非常多。我相信信令服务器必须要有的一项功能便是服务端主动向客户端推送消息，同样的道理，具有以上功能的都可以考虑用来做信令服务器，如何更好的融入系统，这才是影响我们选择的重要因素。

这里需要引入一个房间的概念来将多人规定在一个域内，并且我能够通过房间号获取到所有人的信息（包括web sockets的连接信息）。

基础架构如下：

服务端设置了多种消息格式，用于处理不同的消息（消息格式与客户端是对应的），具体消息格式为：

USER_IDENTITY("USER_IDENTITY","用户身份认证"),SDP_OFFER("SDP_OFFER","type 为 offer 的 desc"),SDP_ANSWER("SDP_ANSWER","type 为 answer 的 desc"),SDP_CANDIDATE("SDP_CANDIDATE","type 为 offer 的 desc"),CLIENT_MULTIPLY_CONNECTION_CREATEQUE("CLIENT_MULTIPLY_CONNECTION_CREATEQUE","多个连接创建请求"),SERVER_CALLEESRESP("SERVER_CALLEESRESP","成员信息响应");

本文福利，免费领取C++音视频学习资料包、技术视频，内容包括（音视频开发，面试题，FFmpeg ，webRTC ，rtmp ，hls ，rtsp ，ffplay ，srs）↓↓↓↓↓↓见下面↓↓文章底部点击免费领取↓↓

Web sockets

服务端的web sockets 需要保存用户与服务端的session以及组装便于 web scokets 接收消息时，进行消息处理的数据结构。

Web sockets 收发消息处理

服务端web socket 主要负责消息的处理以及转发还有相关业务逻辑的保存，这里抽象一个MsgCenter类来负责处理各类消息。抽象一个SignalingChannel来保存user、对等点（多个），房间号信息。通过SignalingChannel能够获取当前用户的对等点用户，以便转发 SDP。

// signalingChannel 类
/*** 请求房间号*/private String roomId;/*** 调用者*/private User caller;/*** 远程对等用户名称集合*/private Set<String> callees;public SignalingChannel(User user,String roomId) {this.caller = user;this.roomId = roomId;this.callees = new HashSet<>();}/*** 新增远程对象* @author DJZ-HXF* @date 2019/6/12 15:58* @param userName*/public void addCallee(String userName){callees.add(userName);}public void addCallees(Set<String> userNames){callees.addAll(userNames);}// MessageCenter 类
/*** 消息对应的处理器*/private static Map<String, BiConsumer<? extends Map<String,String>,Session>> handlers = new HashMap<>();public static void handleMsg(String msg, Session session) {MessageWrapper messageWrapper = JSONObject.parseObject(msg, MessageWrapper.class);BiConsumer consumer = handlers.get(messageWrapper.getMsgType());if (consumer != null) {if(messageWrapper.getMsgBody() instanceof JSONArray){JSONArray temp = (JSONArray) messageWrapper.getMsgBody();// 组装name 和 sdp 的映射Map<String,String> result = new HashMap<>();temp.forEach(jsonObject->{if(jsonObject instanceof JSONObject){Map tempMap = (Map) jsonObject;result.put(tempMap.get("userName").toString(),tempMap.get("sdp").toString());}});consumer.accept(result,session);}else{consumer.accept(messageWrapper.getMsgBody(), session);}}}/*** 新增消息处理器** @param msgType    消息类型* @param biConsumer 消息处理者* @author DJZ-HXF* @date 2019/6/12 16:30*/public static <T> void addHandler(String msgType, BiConsumer<? extends Map<String,String>, Session> biConsumer) {handlers.put(msgType, biConsumer);}

1.接收到认证消息时，保存用户信息。

// 处理用户认证消息MessageCenter.addHandler(MsgConsts.USER_IDENTITY.getValue(), (Map<String, String> map, Session session) -> {User user = new User();user.setName(map.get("name"));user.setRoomId(map.get("roomId"));sessionIdUserMap.put(session.getId(), user);userNameSessionMap.put(user.getName(), session);});

2.接收到多个连接创建请求（即开始请求多人通话）消息时，，建立SignalingChannel


// 处理多个连接建立请求 并响应成员信息MessageCenter.addHandler(MsgConsts.CLIENT_MULTIPLY_CONNECTION_CREATEQUE.getValue(), (Map<String, String> map, Session session) -> {User user = new User();user.setName(map.get("name"));user.setRoomId(map.get("roomId"));Set<String> waitCalleeNames = new HashSet<>();synchronized (SdpWebSocket.class){// 获取房间里获取尚未建立信道的成员Set<String> hasSignalingChannleUsers = Optional.ofNullable(roomIdSignalingChannelsMap.get(user.getRoomId())).orElseGet(HashSet::new).stream().map(u -> u.getCaller().getName()).collect(Collectors.toSet());sessionIdUserMap.forEach((id, u) -> {if (user.getRoomId().equals(u.getRoomId()) && !hasSignalingChannleUsers.contains(u.getName()) && !user.getName().equals(u.getName())) {waitCalleeNames.add(u.getName());}});SignalingChannel signalingChannel = new SignalingChannel(user, user.getRoomId());signalingChannel.addCallees(waitCalleeNames);sessionIdSignalingChannelMap.put(session.getId(), signalingChannel);if (!roomIdSignalingChannelsMap.containsKey(user.getRoomId())) {roomIdSignalingChannelsMap.put(user.getRoomId(), new HashSet<>());}roomIdSignalingChannelsMap.get(user.getRoomId()).add(signalingChannel);}if(waitCalleeNames.size()==0){return ;}// 响应成员信息MessageWrapper<Set<String>> result = new MessageWrapper();result.setMsgType(MsgConsts.SERVER_CALLEESRESP.getValue());result.setMsgBody(waitCalleeNames);try {session.getBasicRemote().sendText(JSONObject.toJSONString(result));} catch (IOException e) {e.printStackTrace();}});

远程对等用户名称集合为房间内成员并且尚未建立`signalchannel`的成员。

并且信令服务器响应将与之建立连接的成员信息，以便客户端创建对应数量的连接。

3.接收到多个 type 为 offer 的 desc和成员名称的消息时, 获取其SignalingChannel , 然后向其匹配的成员发送该desc和自己的名称。

// 处理type为offer的desc,MessageCenter.addHandler(MsgConsts.SDP_OFFER.getValue(), (Map<String, String> userNameSdp, Session session) -> {SignalingChannel signalingChannel = sessionIdSignalingChannelMap.get(session.getId());// 向远程发送offersignalingChannel.getCallees().forEach(userName -> {if(userNameSdp.containsKey(userName)){Session remoteSession = userNameSessionMap.get(userName);if (remoteSession != null && remoteSession.isOpen()) {try {SdpWrapper sdpWrapper = new SdpWrapper();sdpWrapper.setSdp(userNameSdp.get(userName));sdpWrapper.setUserName(signalingChannel.getCaller().getName());MessageWrapper<SdpWrapper> result = new MessageWrapper();result.setMsgType(MsgConsts.SDP_OFFER.getValue());result.setMsgBody(sdpWrapper);remoteSession.getBasicRemote().sendText(JSONObject.toJSONString(result));} catch (IOException e) {e.printStackTrace();}}}});});

4.接收到单个 type 为 answer 的desc和成员名称的消息时，检测房间里的signalingChannel集合，如果在信道的远程对等点集合中包含了该成员名称，那么向该调用者发送answer和自己的名称。

// 处理type为answer的desc,MessageCenter.addHandler(MsgConsts.SDP_ANSWER.getValue(), (Map<String, String> map, Session session) -> {SignalingChannel signalingChannel = sessionIdSignalingChannelMap.get(session.getId());SdpWrapper sdpWrapper = new SdpWrapper();sdpWrapper.setUserName(signalingChannel.getCaller().getName());sdpWrapper.setSdp(JSONObject.toJSONString(map.get("sdp")));Set<SignalingChannel> signalingChannels = roomIdSignalingChannelsMap.get(signalingChannel.getRoomId());// 向其调用者发送 answersignalingChannels.forEach(sc -> {if(sc.getCaller().getName().equals(map.get("userName"))){Session remoteSession = userNameSessionMap.get(sc.getCaller().getName());if (remoteSession != null && remoteSession.isOpen()) {MessageWrapper<SdpWrapper> result = new MessageWrapper();result.setMsgType(MsgConsts.SDP_ANSWER.getValue());result.setMsgBody(sdpWrapper);try {remoteSession.getBasicRemote().sendText(JSONObject.toJSONString(result));} catch (IOException e) {e.printStackTrace();}}}});});

5.接收到单个 candidate和成员名称的消息时，获取其SingalingChannel，然后向其对等点中的该成员发送该 candidate和自己的名称，并且还需要检测房间里的signalingChannel集合，如果该信道的调用者为该成员名称，那么向该调用者发送answer和自己的名称。

// 处理 candidateMessageCenter.addHandler(MsgConsts.SDP_CANDIDATE.getValue(), (Map<String, String> map, Session session) - {SignalingChannel signalingChannel = sessionIdSignalingChannelMap.get(session.getId());SdpWrapper sdpWrapper = new SdpWrapper();sdpWrapper.setUserName(signalingChannel.getCaller().getName());sdpWrapper.setSdp(JSONObject.toJSONString(map.get("sdp")));MessageWrapper<SdpWrapper> result = new MessageWrapper();result.setMsgType(MsgConsts.SDP_CANDIDATE.getValue());result.setMsgBody(sdpWrapper);// 向对等体发送 candidatesignalingChannel.getCallees().forEach(userName -> {if(userName.equals(map.get("userName"))){Session remoteSession = userNameSessionMap.get(userName);if (remoteSession != null && remoteSession.isOpen()) {try {remoteSession.getBasicRemote().sendText(JSONObject.toJSONString(result));} catch (IOException e) {e.printStackTrace();}}}});Set<SignalingChannel> signalingChannels = roomIdSignalingChannelsMap.get(signalingChannel.getRoomId());signalingChannels.forEach(sc -> {if (sc.getCaller().getName().equals(map.get("userName"))) {Session remoteSession = userNameSessionMap.get(sc.getCaller().getName());if (remoteSession != null && remoteSession.isOpen()) {try {remoteSession.getBasicRemote().sendText(JSONObject.toJSONString(result));} catch (IOException e) {e.printStackTrace();}}}});});

本文福利，免费领取C++音视频学习资料包、技术视频，内容包括（音视频开发，面试题，FFmpeg ，webRTC ，rtmp ，hls ，rtsp ，ffplay ，srs）↓↓↓↓↓↓见下面↓↓文章底部点击免费领取↓↓