WebSocket协议及其在实时通信中的重要性

本文深入介绍了WebSocket协议及其在实时通信中的重要性。从HTTP的限制到WebSocket的优势，再到连接建立和实际应用，全面阐述了WebSocket的工作原理及其在实际业务中的应用场景。

一、引言

在生产中，有时需要服务端向客户端发送消息，但是在传统的 HTTP 协议中，是请求-响应模式，也就是说每个请求都是独立的，是由客户端向服务器发送请求，服务器处理请求并返回响应，然后连接就会关闭。

这种请求-响应模式并不能支持后端发起请求，为了解决传统 HTTP 协议在实时通信中的限制，WebSocket协议被引入。

二、WebSocket 介绍

WebSocket 是一种网络传输协议。它允许客户端和服务器之间进行双向通信，而不需要每次请求都重新建立连接。可在单个 TCP 连接上进行全双工通信，位于 OSI 模型的应用层。

WebSocket 通信始于 HTTP 握手，之后升级到WebSocket协议。具有以下优势：

双向通信：WebSocket支持全双工通信，允许服务器主动向客户端推送数据，而不需要客户端发送请求。
较低的延迟：WebSocket 通过在建立连接后保持持久连接的方式，避免了重复建立和关闭连接的开销。可以减少延迟，实现更快的数据传输和实时更新。
减少网络流量：与轮询方式相比，WebSocket 采用事件驱动的方式，只在有新数据时才发送更新，避免了不必要的网络流量和服务器负载。
兼容性：WebSocket 协议已经得到了广泛的支持。

三、其他主动推送

	短轮询	长轮询	iframe流	SSE	WebSocket	Socket.IO
实时性	较差	较差	较好	较好	较好	较好
网络开销	较大	较大	较小	较小	较小	较小
协议支持	HTTP 协议	HTTP 协议	HTTP 协议	HTTP 协议	WebSocket 协议	自适应
跨域支持	较差	较差	较差	支持	支持	支持
兼容性	较好	较好	较好	较好	较好	较好
双向通信	单向通信	单向通信	单向通信	单向通信	双向通信	双向通信
实现复杂度	相对简单	相对简单	相对简单	相对复杂	相对复杂	相对复杂
延迟	较高	相对较低	相对较低	较低	最低	较低

1.短轮询

优势：

实现简单。简单场景的快速的解决方案。
劣势：
请求频繁。频繁的请求和响应会导致较高的网络开销和延迟。
资源占用高。对服务器资源的占用较高，每次请求需要建立连接和断开连接。

2.长轮询

优势：

相比短轮询，能够减少一部分请求的频繁性。客户端在收到响应后会立即发起新的请求。
相比短轮询速度相较快。服务器在有数据时才会响应，能够更快地将数据推送给客户端。
劣势：
延迟高。存在一定程度的延迟，客户端需要等待服务器有数据时才能收到响应。
资源占用高。服务器需要维护大量的长连接，可能会影响服务器的性能。

3.iframe流

优势：

通过在 iframe 中加载长连接资源来实现数据推送，相对于传统的轮询方式，可以降低一定程度的延迟。
劣势：
受到浏览器同源策略的限制。
复杂应用场景难以管理和维护。

4.SSE（Server-Sent Events）

优势：

基于 HTTP，在不需要额外的握手过程的情况下实现服务器向客户端的数据推送，降低了通信的开销。
相对于传统的轮询方式，能够实现较低的延迟。
劣势：
仅支持单向通信，无法实现客户端到服务器的双向通信。

5.Socket.IO

优势：

提供了跨浏览器的双向通信能力，可以自动选择最佳的通信方式，包括 WebSocket、轮询等，从而实现较低的延迟。
支持实时双向通信。
劣势：
需要额外的库支持，增加代码的复杂性。

6.WebSocket

优势：

提供了实时的双向通信能力，实现最低的延迟。
与 HTTP 不同，WebSocket 在建立连接后能够直接实现服务器和客户端之间的双向通信，而不需要频繁地发起新的连接。
劣势：
部署和维护复杂。

7.如何选择

双向通讯：WebSocket、Socket.IO
实时性：WebSocket >= Socket.IO > SSE ≈ iframe流 ≈ 长轮询 > 短轮询
仅单向通讯：SSE
场景简单且不复杂：长轮询、短轮询

四、WebSocket 应用

1.传统 HTTP 的限制

HTTP 是请求-响应模式，也就是说每个请求都是独立的。
WebSocket 是全双工通信。全双工（Full Duplex）是指在发送数据的同时也能够接收数据，两者同步进行。

用一个例子来解释一下两个的区别：

WebSocket（ws）就像你在餐厅里用餐，旁边有一位专门跟着你的服务员。你点菜，服务员会立刻把你的要求传达给厨房，菜做好后立刻送到你桌上，甚至如果厨房需要你的建议，服务员也能立即传达给你。

而 HTTP 就像整个餐厅共用一批服务员。可能 A 服务员会给你送第一道菜，但送第二道菜的时候会换成 B 服务员。而且服务员之间不共享信息。也就是说，如果你向 A 服务员点了菜，当你问 B 服务员的时候，B 会回答你“很快就上了”，但实际上 B 并不知道这个事情，只是在安慰你的情绪。

从资源使用上讲，相比于 HTTP，WebSocket 更具优势。相比于每次通讯需要建立连接，WebSocket只需要维持 TCP 即可。
![[Pasted image 20231209204843.png]]

2.WebSocket 连接的建立

WebSocket 连接流程

建立 TCP 连接：WebSocket 连接首先需要建立一个基础的 TCP 连接，这是因为WebSocket 是基于 TCP 的。
发送特殊的 HTTP 请求（WebSocket 握手）：客户端会发送一个特殊的 HTTP 请求，这个请求被称为 WebSocket 握手请求。这个请求包含一些特殊的头部信息，其中包括 Upgrade 和 Connection 字段，告诉服务器希望升级到 WebSocket 协议。服务器收到这个请求后，如果支持 WebSocket，就会进行协议升级。
服务器确认协议升级：如果服务器支持 WebSocket，它会发送一个类似的响应，这个响应也包含特殊的头部信息，告诉客户端协议已经升级到 WebSocket。这样，从此之后，客户端和服务器之间的通信就不再遵循传统的 HTTP 协议，而是遵循 WebSocket 协议。
使用 WebSocket 协议进行通讯：一旦协议升级完成，客户端和服务器就可以使用 WebSocket 协议进行实时的双向通信，发送和接收数据帧而无需频繁地建立和断开连接。
![[Pasted image 20231209204917.png]]

WebSocket 握手

创建HTTP请求，对连接进行升级参数如下：

# 请求头
Request Headers
# websocket版本
Sec-WebSocket-Version: 13
# 唯一口令，base64编码
Sec-WebSocket-Key: 2icxlyJOYxKXJpCXa8T14Q==
# 连接升级为websocket协议
Connection: Upgrade
Upgrade: websocket
Sec-WebSocket-Extensions: permessage-deflate; client_max_window_bits
Host: 127.0.0.1:9095# 返回值
Response Headers
# 升级为websocket协议
Upgrade: websocket
Connection: upgrade
# 口令
Sec-WebSocket-Accept: stCq5oQ38vohTpeBYUTMb0IU8Fo=
Sec-WebSocket-Extensions: permessage-deflate;client_max_window_bits=15

![[Pasted image 20231209202336.png]]

![[Pasted image 20231209202607.png]]

3.分布式环境中的 Session 共享

分布式问题

在生产中，程序往往是以分布式进行部署的，即启动多份程序，通过代理提高并发能力。

如下图，用户A 通过负载在节点1上建立了 WebSocket 连接，连接的 Session 也存储在节点1。当用户A 的数据通过其他方式到达系统时，通过负载代理，最终到达节点4，当节点4对数据进行发送时，发现此节点没有用户A的Session，这样就导致了分布式情况下出现发送不了的情况。

WebSocket 的 Session 是继承自 Closeable，不能像 HttpSession 一样，把内容序列化到Redis中，每个客户端都会与服务器之间建立独立的 WebSocket 连接，Session 存储在建立连接的服务中，那么就引申出 WebSocket 的分布式问题。

![[Pasted image 20231129215436.png]]

解决方法

解决方法大概有三种，分别为中间件广播处理、服务间建立WebSocket连接、请求代理哈希转发。

中间件广播处理：利用中间件广播的能力，当出现需要发送的消息时，广播所有节点，至少有一个节点能够处理此消息，发送给客户端。
服务间建立WebSocket连接：所有服务间建立WebSocket连接，分内外两部分连接，外部为用户发起的连接，内部为节点间的连接。通过记录用户连接关系，即可知消息需发送的客户端节点。
请求代理哈希转发：通过重写代理的方式，获取请求头中用户信息，哈希分配到指定节点处理操作。

下面是不同方法间的对比：

	中间件广播处理	服务间建立WebSocket连接	请求代理哈希转发
优势	简单易实现	点对点通讯，灵活性较高	定向请求，保证数据一致性
劣势	依赖于中间件，广播效率低	系统复杂，需维护大量连接	可能负载不均衡，需自定义负载策略
扩展性	扩展性好	随节点增加复杂	节点增加减少需重新计算哈希值

分析

第二种、第三种方法是可以解决分布式的问题，但在实现难易程度、扩展性方面相比第一种不具有优势。

第二种，需要维护众多WebSocket连接，需要节点间长连接，自动重连等功能，维护调试成本高。
第三种，需要对集群节点可用数量精确把控，若节点已经down掉，需要重新计算哈希值，以免代理到此节点上，维护成本高。

综上所述，选择第一种方案相比来讲，在开发、维护方面更加具有优势。

![[Pasted image 20231129215455.png]]

五、WebSocket案例

1.业务场景和需求

消息数据通过三方接口接入到系统中，服务端主动发送消息到小程序，小程序进行页面跳转，展示消息（由于小程序不支持SSE，选择WebSocket解决主动推送问题）。

要求有较高实时性。由于消息不知什么时候发送到系统中，所以连接在跳转前一直维持。

2.分析

由于消息会不定时接入到系统中，请求通过负载到不同节点，需要保持连接，且解决分布式问题。分析结果如下：

分布式问题。消息接收节点与用户建立 WebSocket 连接节点，两节点不一定一致。
连接空挡。连接到达最大时间后自动断开，与下次连接间存在时间差。所以需要重复连接，这样会导致多个连接存在，消息多次发送问题。
恶意连接。恶意使用不存在的 key 进行恶意连接。

针对上述问题方案如下：

使用MQ广播解决分布式问题。接收消息广播的方式分发。
在 WebSocket 连接成功后也进行广播，目的是断开此人其他的连接，保证连接有且仅有一个。
在连接建立前重写钩子函数，对用户进行校验。

![[Pasted image 20231129220451.png]]

3.关键代码

对Session进行封装，扩展Session数据，添加Session管理类，封装相应的方法

public class SessionExt {  private WebSocketSession session;  private String uniqueId;... 省略 get set 方法
}

@Component  
public class SessionContainer {  private static final ConcurrentHashMap<String, SessionExt> SESSION_MAP =  new ConcurrentHashMap<>(WebSocketConstants.INT_16);  /**  * 获取 session 数据  *  * @param key key  * @return session  */    public SessionExt getSessionExt(String key) {  return SESSION_MAP.getOrDefault(key, null);  }  /**  * 添加 session 数据  *  * @param key        key  * @param sessionExt sessionExt  */    public void addSessionExt(String key, SessionExt sessionExt) {  SESSION_MAP.put(key, sessionExt);  }  /**  * 添加 session 数据  *  * @param key        key  * @param sessionExt sessionExt  */    public void addSessionExtAndClose(String key, SessionExt sessionExt) {  SessionExt sessionExtOld = SESSION_MAP.get(key);  if (sessionExtOld != null) {  sessionExtOld.closeSession();  }  SESSION_MAP.put(key, sessionExt);  }  /**  * 删除 session 数据  *  * @param key key  */    public void delSessionExt(String key) {  SESSION_MAP.remove(key);  }  
}

连接创建、关闭后的广播回调，关闭多余连接

 
public void handleOpenProcessing(String key, String uniqueId) {  SessionExt sessionExt = SESSION_CONTAINER.getSessionExt(key);  if (sessionExt != null) {  String uniqueIdOld = sessionExt.getUniqueId();  if (!uniqueIdOld.equals(uniqueId)) {  log.debug("[websocket]广播，连接后置处理，关闭用户之前连接");  sessionExt.closeSession();  SESSION_CONTAINER.delSessionExt(key);  }  }  log.info("[websocket]广播，连接后置处理，完成");  
}  public void handleCloseProcessing(String key, String uniqueId) {  log.debug("[websocket]广播处理 关闭连接，key：" + key);  SessionExt sessionExt = SESSION_CONTAINER.getSessionExt(key);  if (sessionExt != null) {  String uniqueIdOld = sessionExt.getUniqueId();  if (uniqueIdOld.equals(uniqueId)) {  sessionExt.closeSession();  SESSION_CONTAINER.delSessionExt(key);  }  }  log.info("[websocket]广播，关闭后置处理，完成");  
}