Web 实时消息推送的两种方案：WebSocket 与 SSE

在接触服务端消息主动推送的时候，做业务遇到需要一个”新消息提醒”、“订单状态变化”、“进度条实时更新”这种，其实第一反应是加一个定时器轮询接口。但是很明显，在你的业务压力不是很大的时候，这种方式会白白浪费很多宝贵的系统资源。。。因此，在一个事件完成之后，服务端主动推送消息是很有必要的。

从轮询到推送，实时消息到底难在哪

HTTP 的本质就不支持服务器主动说话

要理解实时推送难在哪，得先接受一个有点反直觉的事实：HTTP 从设计之初就不是为了”服务器主动找客户端”而生的。

在经典的 HTTP 请求响应模型里，通信永远是客户端发起的：

1
2
3
4
5
6
客户端                                      服务端
  |                                           |
  | ----------- 请求 (Request) -------------> |
  |                                           |  处理业务
  | <---------- 响应 (Response) ------------- |
  |                                           |

这个模型有三个隐含前提：

  1. 通信由客户端发起，服务端只负责应答；
  2. 服务端处理完就忘掉这次连接，不保证还有下一次；
  3. 服务端手里没有一张”当前在线的客户端”名单，它无法主动构造出一个响应。

于是矛盾就出现了：业务上我们需要，服务端有新数据时立刻告诉客户端，这样的时候，但协议层面服务端根本没有一条现成的路可以走。

历史上为了绕开这个矛盾，工程师们想出了一系列曲线救国的办法，它们构成了实时推送技术的完整演进阶梯。

短轮询

最朴素的做法是短轮询（Short Polling）：客户端用定时器每隔几秒发一次请求问一下有没有新数据。

1
2
3
4
5
6
7
setInterval(async () => {
  const res = await fetch('/api/messages/latest');
  const data = await res.json();
  if (data.hasNew) {
    render(data);
  }
}, 3000);

但代价也很直观。假设 1 万个在线用户，轮询间隔 3 秒，那么 QPS 就是：

1
10000 / 3 ≈ 3333 QPS

这 3333 QPS 里，绝大多数请求的结果都是没有新数据。RFC 6455 在讲 WebSocket 的设计动机时，把这种做法的毛病总结得很清楚，大概就是说，服务端被迫为每个客户端维护多条底层 TCP 连接；线路协议开销巨大，每条客户端到服务端的消息都拖着完整的 HTTP 头；客户端脚本还得自己维护请求与响应之间的映射关系。

其实，短轮询最致命的其实是第一条。每一次轮询都是一次完整的请求响应周期

而如果你为了省资源把间隔拉长到 30 秒，那么最坏情况下要等 30 秒才看到新消息。这就是短轮询的问题：实时性和资源消耗只能二选一，没有中间地带。

长轮询

长轮询（Long Polling）是第一次真正意义上的改进。它的思路是把短轮询中立刻返回空的响应，改成了请求先挂住，等到有数据或者超时再返回。

sequenceDiagram
    participant C as 客户端
    participant S as 服务端
    C->>S: GET /api/messages/poll
    Note over S: 没有新数据，把请求挂起（不返回）
    Note over S: ...等待中...
    S-->>C: 数据到达，返回响应
    C->>S: 立即发起下一次 GET
    Note over S: 再次挂起

它带来的收益非常明显，空响应的次数大幅减少。没有新数据时，请求一直挂在服务端，不消耗带宽也不浪费序列化开销。延迟也降下来了，因为一旦有数据，响应是立刻推出去的，客户端收到后又马上发起下一轮，这是一种近实时方案。

但长轮询引入了新的复杂度，而且这些复杂度都不好处理：

  • 连接状态变成了业务状态。 服务端必须为每个挂起的请求保存一个”待唤醒”的句柄，还要给它配上超时时间。
  • 超时与重连的边界很微妙。 超时时间设得太短，请求会不断返回空响应，退化成短轮询；设得太长，中间的代理会先替你断掉（nginx 默认 60 秒，见后文）。
  • 重连风暴。 服务端一旦重启或抖动，所有挂起的连接会在同一瞬间断开，然后客户端会在同一瞬间重新发起连接，形成一波新的压力峰值。
  • 仍然不是真正的推送。 每次收到数据后都要重新建一次请求，请求头、鉴权、路由全部重来一遍。

HTTP 流式推送能做到一次请求，持续输出

再往前一步就是 HTTP 流式推送（HTTP Streaming）。既然长轮询的问题是每次都断开重连，那就不断开：服务端返回一个永不结束的响应，用分块传输的方式持续往同一个连接里写数据，客户端边读边解析。

1
2
3
4
5
6
7
HTTP/1.1 200 OK
Content-Type: text/event-stream
Transfer-Encoding: chunked

<chunk>data: 第一条消息</chunk>
<chunk>data: 第二条消息</chunk>
...（连接保持打开）

这一次，一个连接对应一整条会话，请求头只发一次，鉴权只做一次。这就是 SSE 的技术底座。

不过裸的 HTTP 流式推送也是有问题的：响应体是什么格式？客户端怎么切分消息？断线了怎么知道断在哪里？这些如果不统一约定，就是各个团队自己造轮子。

SSE 把这个做法标准化了。 它规定了两件事，一是 MIME 类型必须是 text/event-stream，二是事件流必须遵循一份严格的 ABNF 语法；再加上浏览器原生提供的 EventSource 接口，把连接管理、消息解析、断线重连这一整套规范都定下来了。

WHATWG 规范对它的定位说得很直白，我用中文讲一下大意：这套东西存在的目的，是让服务器能够通过 HTTP（或者专门的服务器推送协议）把数据推给网页，为此规范引入了 EventSource 这个接口。 换句话说，它不是发明了新的传输方式，而是给”服务端往浏览器推数据”这件事定了一套标准的做法。

WebSocket 与 SSE 是两条完全不同的设计路线

走到这里，实时推送已经分裂成了两条路线，它们的设计哲学完全不同：

WebSocket 认为 HTTP 不适合双向通信，那就在 HTTP 的基础上做一次”协议升级”，之后脱离 HTTP，用自己的帧格式进行通信。 它的目标是真双向、低开销的实时通信。

SSE 认为 HTTP 已经够用了，问题只是缺一个格式约定和客户端 API，那就把它标准化。 它要解决的是服务端到客户端的单向高效推送。

这两个东西都是实打实的标准：

方案 标准来源 标准化时间 传输层
WebSocket IETF RFC 6455（Proposed Standard） 2011 年 12 月 独立协议，借 HTTP 完成握手
SSE WHATWG HTML Living Standard 第 9.2 节 随 HTML 标准持续演进 就是 HTTP

WebSocket 的设计目标也很明确：为浏览器应用提供一种不依赖多条 HTTP 连接的双向通信机制。

这个协议要解决的，是让一段运行在受控环境里的不可信客户端代码，能和一个”同意接受这段代码通信”的远端主机之间建立双向通信；它采用的安全模型，是浏览器世界里通用的那套基于来源（Origin）的模型。

这句话里有两个信息值得留意，一是它明确假设客户端代码是不可信的，所以后面才会有掩码、Origin 校验这些设计；二是它把安全责任交给了 Origin 模型，这正是它和普通 HTTP 接口最大的不同，后面讲安全时我会展开。

整个协议由两部分组成，一个开场的握手，加上基本的分帧（framing），两者都跑在 TCP 之上。

一条贯穿全文的结论

WebSocket 和 SSE 是两种不同的取舍。

  • 如果客户端只需要被动接收，而你的团队又不想为了一套新协议去改网关、改鉴权、改监控，那么 SSE 几乎是免费的午餐；
  • 如果确实需要客户端频繁上行（聊天、协同编辑、游戏操作），或者消息频率和体量都很大，那 WebSocket 才是那个便宜的选项，因为用 SSE 模拟双向反而更贵。

什么时候不该用 WebSocket，Spring 官方文档在 WebSocket 那一章里给的判断标准，我觉得比大多数博客都靠谱。它说的是这么一层意思：延迟高低本身不是决定性的因素。如果消息量相对较低（比如监控网络故障这种场景），那么 HTTP 流式或轮询就已经是有效的方案了；真正让 WebSocket 成为最佳选择的，是”低延迟 + 高频 + 大流量”这三者同时成立。

提炼成一句就是：光有低延迟需求还不足以让你上 WebSocket，只有三者叠加，才是它最合适的场合。 这个判断标准和前面那条设计哲学是完全一致的。

WebSocket 协议原理

握手机制

WebSocket 连接的第一步，是一次看起来完全正常的 HTTP 请求。

RFC 6455 在协议概览里给了完整的例子。客户端请求长这样：

1
2
3
4
5
6
7
8
GET /chat HTTP/1.1
Host: server.example.com
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
Origin: http://example.com
Sec-WebSocket-Protocol: chat, superchat
Sec-WebSocket-Version: 13

服务端响应长这样：

1
2
3
4
5
HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
Sec-WebSocket-Protocol: chat

这个握手，每一行都有讲究。

请求方必须满足的条件

RFC 6455 第 4.1 节用 MUST 逐条规定了客户端的要求。下面这份清单里的”必须”，对应的都是规范里的强制要求，不是建议：

要求 具体内容
方法 必须是 GET，HTTP 版本必须至少 1.1
Host 必须包含主机名，非默认端口时还要带端口
Upgrade 必须包含 websocket 关键字
Connection 必须包含 Upgrade 令牌
Sec-WebSocket-Key 必须是一个随机选取的 16 字节值做 base64 编码，且每连接都必须重新随机
Origin 浏览器客户端必须携带；非浏览器客户端可选
Sec-WebSocket-Version 必须是 13
Sec-WebSocket-Protocol 可选，逗号分隔的子协议列表，按偏好排序
Sec-WebSocket-Extensions 可选，希望使用的协议级扩展

关于 Sec-WebSocket-Key，它的本质是一个每连接唯一的随机 nonce，编码成 base64 只是为了放在 HTTP 头里方便传输，解码之后必须正好是 16 字节。规范特意强调了这个值必须”每个连接都重新随机选取”，不能复用——这一点很重要，因为它决定了这个字段是一个真正的挑战值，而不是某个固定标识。

这里还有一个很容易被误解的点，我在后面讲 Sec-WebSocket-Accept 时会展开：这个 Key 不是用来做认证或加密的，它只是一个”挑战值”。

服务端如何校验：SHA-1 + 固定 GUID

服务端收到请求后，要计算 Sec-WebSocket-Accept。拆开看是三步：

  1. 取客户端 Sec-WebSocket-Key 的原始字符串值（注意，是 header 里那段 base64 文本本身，不是解码之后的 16 字节）；
  2. 拼接上固定的魔数 GUID：258EAFA5-E914-47DA-95CA-C5AB0DC85B11；
  3. 对拼接结果做 SHA-1，得到一个 20 字节的摘要，再对它做 base64 编码，就是 Sec-WebSocket-Accept。

关于那个 GUID，规范给了一句解释，说明为什么它不容易冲突。大意是：这个 GUID 写成了字符串形式，它的取值使得那些”不理解 WebSocket 协议”的网络端点，几乎不可能碰巧用到它。

关键就是这后半句：一个不理解 WebSocket 协议的端点，不可能碰巧把这个 GUID 当成正常数据。 这正是整个握手校验能成立的前提。

用规范里的示例验证一遍，dGhlIHNhbXBsZSBub25jZQ== 经过上述计算得到的正是 s3pPLMBiTxaQ9kYGzzhZRbK+xOo=。这个对应关系是规范的官方向量，可以直接拿来写单元测试。

自己实现时要注意一个细节，因为规范明确说了服务端不需要对 Sec-WebSocket-Key 做 base64 解码，直接拿字符串参与拼接就行。所以服务端只需要校验它的 base64 解码结果是 16 字节，然后按字符串拼接计算摘要即可。

另外，服务端在读取客户端握手时有几项必须验证的内容，我用中文列一下，这样你对照代码时不容易漏：请求必须是 HTTP/1.1 或更高的 GET；Host 头必须包含服务端的 authority；Upgrade 头必须包含 websocket（大小写不敏感）；Connection 头必须包含 Upgrade 令牌；Sec-WebSocket-Key 解码后必须是 16 字节；Sec-WebSocket-Version 必须是 13。这几项里任何一项不满足，服务端就应该停止处理这个握手，返回一个合适的 HTTP 错误码（比如 400）。

为什么一个安全协议会用 SHA-1

SHA-1 早就被弃用了，为什么 WebSocket 握手还在用它？

官方给了很明确的一句话，中文大意是：这个握手不依赖 SHA-1 的任何安全属性，包括抗碰撞性和抗第二原像攻击性。

这句话的分量比它看起来要重。SHA-1 之所以被淘汰，是因为它的抗碰撞性被攻破了，可以构造出两个不同输入得到相同摘要。但 WebSocket 握手根本不需要抗碰撞性：

  • 它要防的不是”有攻击者能构造出碰撞”，而是”有不理解协议的东西碰巧也返回了正确的 Accept”；它防的是误连，真正的安全性来自 TLS（wss://）和 Origin 校验；
  • 而这个校验之所以能成立，靠的是前面那个 GUID 的唯一性，一个不懂 WebSocket 的端点不会把 258EAFA5-E914-47DA-95CA-C5AB0DC85B11 当成自己的正常数据；
  • 所以攻击者就算能算出这个摘要也毫无意义，因为算法是公开的，摘要里没有任何 secret。

换句话说，SHA-1 在这里只是一个便宜的、所有语言都内置的一致性校验函数。

顺带一个值得知道的细节：RFC 8441（HTTP/2 上的 WebSocket）明确说明第 10.8 节不再适用，因为扩展 CONNECT 方案里 Sec-WebSocket-Key 和 Sec-WebSocket-Accept 这套机制被 :protocol 伪首部取代了。

校验失败的行为

规范用一句概括性的话说明了客户端的校验底线，大意是：如果 Sec-WebSocket-Accept 的值和期望值不符、这个头缺失、或者 HTTP 状态码不是 101，那么连接就不会被建立。

而官方逐条列出了客户端必须检查的项，任何一条不满足，客户端都必须让连接失败：

  • 状态码不是 101，就按普通 HTTP 处理；
  • 响应缺少 Upgrade 头，或者值不是 websocket（大小写不敏感）；
  • 响应缺少 Connection 头，或者不包含 Upgrade 令牌；
  • 响应缺少 Sec-WebSocket-Accept，或者值不等于上面算出来的结果；
  • 响应带上了一个客户端没请求过的扩展；
  • 响应带上了一个客户端没请求过的子协议。

最后两条经常被忽略。它们的实际含义是：服务端不能在客户端没同意的前提下擅自启用扩展或子协议。 比如客户端没有请求 permessage-deflate 压缩扩展，服务端就不能自己开——否则客户端解帧会直接错乱。

握手之后的连接状态

握手成功后，底层的 TCP 连接并不会重建，它只是换了一套语言。关键点在于，同一个 TCP socket，握手前说 HTTP，握手后说 WebSocket 帧。 这也是为什么 WebSocket 能复用 80/443 端口、能穿过大部分防火墙，因为它在握手阶段和一个普通 HTTP 请求没有区别。

规范把这个状态命名为”WebSocket 连接已建立（Established）“，并且把此后连接所处的状态称为 OPEN 状态。你在日志里看到这两个词（Established / OPEN），说的就是握手已经走完、双方可以开始发帧了。

帧格式是 WebSocket 真正的数据结构

握手只是开场，WebSocket 真正有意思的部分在帧（Frame）格式上。RFC 6455 给出的帧结构图如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
 0                   1                   2                   3
 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-------+-+-------------+-------------------------------+
|F|R|R|R| opcode|M| Payload len |    Extended payload length    |
|I|S|S|S|  (4)  |A|     (7)     |             (16/64)           |
|N|V|V|V|       |S|             |   (if payload len==126/127)   |
| |1|2|3|       |K|             |                               |
+-+-+-+-+-------+-+-------------+ - - - - - - - - - - - - - - - +
|     Extended payload length continued, if payload len == 127  |
+ - - - - - - - - - - - - - - - +-------------------------------+
|                               |Masking-key, if MASK set to 1  |
+-------------------------------+-------------------------------+
| Masking-key (continued)       |          Payload Data         |
+-------------------------------- - - - - - - - - - - - - - - - +
:                     Payload Data continued ...                :
+ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - +
|                     Payload Data continued ...                |
+---------------------------------------------------------------+

这张图是理解 WebSocket 一切性能特性与实现细节的基础。这里有一个容易被忽略的规范细节：这张图比后面的 ABNF 定义更有权威性。 规范明确说了，如果图和 ABNF 之间存在冲突，以图为准。这种情况在协议文档里并不常见，值得记一笔。

按字段逐个拆。

FIN：最后一片标记

FIN 只有 1 个 bit，含义是”这是本消息的最后一个分片”。绝大多数情况下，一个消息就是一个帧，此时 FIN = 1。只有做分片时才会出现 FIN = 0 的中间帧。

规范还补了一句：第一个分片本身也可以是最后一个分片——也就是说，单帧消息本质上就是”只有一个分片的分片消息”，这两种情况在实现上可以统一处理。

RSV1 / RSV2 / RSV3：扩展位

三个保留位，默认必须是 0。如果收到非 0 值，而本次连接又没有协商出能解释这些位的扩展，接收方必须让连接失败。

规范对这三个位的要求是硬性的：除非某个已协商的扩展明确定义了非 0 值的含义，否则它们必须为 0；一旦收到非 0 值而又没有任何扩展能解释它，接收方必须判定连接失败。

这里就能看出 WebSocket 的扩展机制是怎么设计的：扩展就是给这些保留位赋予语义。最典型的例子是 permessage-deflate 压缩扩展，它用 RSV1 表示”本消息的载荷是压缩过的”。

Opcode：4 bit 的操作

规范定义的值如下：

值 含义 类别
%x0 continuation frame（续帧） 数据帧
%x1 text frame（文本帧） 数据帧
%x2 binary frame（二进制帧） 数据帧
%x3-%x7 保留给后续非控制帧 数据帧
%x8 connection close（关闭） 控制帧
%x9 ping 控制帧
%xA pong 控制帧
%xB-%xF 保留给后续控制帧 控制帧

规范对它的定义是：操作码决定载荷数据该如何解释；如果收到一个未知的操作码，接收方必须判定连接失败。 这一点在实现时要严格对待——不能”看不懂就忽略”，那会造成协议状态错乱。

这里有个很好记的规律：操作码最高位为 1 的是控制帧。 所以 0x8、0x9、0xA 的最高位都是 1，它们是控制帧；0x0、0x1、0x2 是数据帧。

MASK：掩码位

一个 bit，表示载荷是否被掩码。规则是不对称的，这一点后面单独讲：

  • 客户端发给服务端的帧，这一位必须是 1；
  • 服务端发给客户端的帧，这一位必须是 0。

Payload length：三种长度的紧凑编码

这是 WebSocket 帧里设计最巧妙的一个字段。它只有 7 个 bit，但要表达的长度可能很大，所以用了转义值的设计：

  • 值在 0-125 之间：这个值就是载荷长度；
  • 值等于 126：后面紧跟的 2 字节（16 位无符号整数）才是真正的载荷长度；
  • 值等于 127：后面紧跟的 8 字节（64 位无符号整数）才是真正的载荷长度；
  • 多字节长度一律按网络字节序（大端）表示。

关于 127 这种情形，规范还加了一条容易被忽略的约束：那 8 字节里最高位必须是 0。 也就是说实际能表达的最大长度是 2 的 63 次方减一，而不是无符号 64 位的满量程。这是一道防止长度字段被滥用的护栏。

规范还有两条硬约束，也就是必须用最少字节数编码。一个 124 字节的载荷不能被编码成 126, 0, 124 这种浪费形式。这条约束的存在是为了防止有人用它做流量放大或者绕过长度限制。

另外，载荷长度统计的是 Extension data + Application data 的总和，所以用了压缩扩展之后，压缩后它算的是压缩后的长度，但应用层拿到的是解压后的长度；所以限流要按应用层长度来限，不然压缩炸弹能直接绕过。

顺便给出长度分界点的实际含义，这在估算帧开销时很有用：

载荷长度 帧头开销（服务端 → 客户端） 帧头开销（客户端 → 服务端）
0 - 125 字节 2 字节 6 字节
126 - 65535 字节 4 字节 8 字节
≥ 65536 字节 10 字节 14 字节

对比一下 HTTP 每次都带几百字节的头部，WebSocket 在高频小消息场景下的开销优势是压倒性的。

掩码机制

  • 掩码密钥是一个 32 位随机值，由客户端选取；
  • 每一帧都必须重新生成，不能复用；
  • 必须来自强熵源，不能是伪随机可预测的；
  • 掩码使用的是异或（XOR）运算，而且不能改变载荷长度。

规范对掩码密钥的要求说得很重：这个密钥必须不可预测，必须来自强熵源，而且某一帧的密钥不能让服务端或代理轻易推断出下一帧的密钥。 它甚至解释了为什么——密钥的不可预测性，是阻止恶意应用”自行挑选线路上出现的字节”的关键。

载荷长度不包含掩码密钥的长度。 那 4 个字节的密钥属于帧头，不占载荷空间，所以掩码不会让数据变长，也不会影响按长度解析。

掩码的变换算法规范写成了一个逐字节的公式：第 i 个字节的变换结果，等于原始字节异或上”密钥里第 i 对 4 取模的那一个字节”。也就是密钥按 4 字节循环使用。

然后，掩码的目的不是保密。 密钥就写在同一个帧里，任何人都能解出来，它提供不了任何机密性。只不过是防止如下的情况发生：

  1. 攻击者先与一个自己控制的服务器建立连接，并在 HTTP 连接上执行 UPGRADE，也就是升级成 WebSocket；
  2. 然后，攻击者通过这个已升级的连接发送数据，而这些数据看起来像是一个针对某个已知资源的 GET 请求；
  3. 部署在中间的不合规代理无法区分这是 WebSocket 帧还是一个新的 HTTP 请求，于是把它当成真实请求转发出去；
  4. 响应被代理缓存——缓存就这样被投毒了。

这不是我编的假设场景，而是规范里记录的一次真实实验。规范说，在这个协议开发期间，有人做过一次实验，演示了一类针对代理的攻击，并且确实导致了部署在真实环境中的缓存代理被投毒。攻击的一般形式就是上面那四步。

这两句话解释了掩码设计的全部逻辑：

  • 掩码的作用是让攻击者无法控制”线路上实际出现的字节”。 他想让第 N 个字节等于某个特定值，就必须让”明文 XOR 随机密钥”的结果等于那个值，而密钥每帧都换且不可预测，这个构造就不可能成功；
  • 威胁模型只是”客户端发送的数据看起来像 HTTP 请求”，所以只需要对客户端到服务端这一个方向做掩码，服务端到客户端不需要。

规范里还有一句很实在的提醒，大意是：即便有了掩码保护，那些不合规的 HTTP 代理，仍然会被”不做掩码的客户端和服务器”投毒。 也就是说，掩码只能保护双方都合规的场景。如果一个不合规的客户端根本不做掩码，那它连的那个不合规代理仍然会被投毒——掩码保护的是整个生态，不是单个连接。

掩码有方向性要求

掩码的方向性在规范里是用强制级别规定的，可以整理成四条：

规则 违反的后果
客户端发给服务端的所有帧必须掩码（无论是否走 TLS） 服务端必须关闭连接，可发 1002 协议错误
服务端发给客户端的帧必须不掩码 客户端必须关闭连接
掩码密钥每帧必须重新随机 视为协议违规
掩码密钥必须来自强熵源 视为协议违规

注意第一条括号里的那句补充：掩码与是否运行在 TLS 之上无关。 有人以为用了 wss:// 就可以省掉掩码，实际上加密和掩码防护的是完全不同的东西，因为 TLS 防的是链路上的窃听者，掩码防的是不合规的中间代理被投毒。两者不能互相替代。

这也解释了为什么用 Java 原生 socket 手写 WebSocket 客户端时，最容易犯的错就是忘记掩码，一对接标准服务端就立刻被 1002 拒绝。

手写掩码的代码

掩码其实就是一次循环异或，因为 XOR 是自反的，掩码和解掩码是同一段代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
/**
 * 按 RFC 6455 §5.3 对载荷做掩码（同时也是解掩码）。
 *
 * @param payload    载荷字节数组，会被原地修改
 * @param offset     载荷起始偏移
 * @param maskingKey 4 字节掩码密钥
 */
public static void applyMask(byte[] payload, int offset, byte[] maskingKey) {
    for (int i = 0; i < payload.length - offset; i++) {
        // 密钥按 4 字节循环使用：j = i MOD 4
        payload[offset + i] ^= maskingKey[i % 4];
    }
}

这段代码可以做单元测试验证：掩码两次应该还原成原文。

分片与重组代表一条消息可以拆成多个帧

WebSocket 的”消息（Message）“和”帧（Frame）“是两个不同层级的概念。一条应用层消息可以由多个帧拼成，这就是分片。

规范给出的定义是两条，大意如下：未分片的消息，就是单个帧，FIN 位为 1 且操作码不为 0；分片的消息，则先是一个 FIN 位为 0、操作码不为 0 的帧，后面跟着零个或多个 FIN 位为 0、操作码为 0 的帧，最后以一个 FIN 位为 1、操作码为 0 的帧收尾。

翻译成规则就是：

  1. 未分片消息：单个帧，FIN = 1，opcode 不是 0；
  2. 分片消息：第一片 FIN = 0 且 opcode 非 0（用真实类型），后面跟零到多个 FIN = 0 且 opcode 为 0 的续帧，最后以一个 FIN = 1 且 opcode 为 0 的帧结束。

接收方把所有片段的载荷依次拼接，直到遇到 FIN = 1，才算收到完整的一条消息。

规范里给的分片动机，大意是：分片最主要的目的，是允许在”消息刚开始发送时还不知道它有多大”的情况下把消息发出去，而不必先把整条消息缓冲起来。也就是支持流式产出，边生成边发，不用把整条消息堆在内存里。分片的次要用途还有多路复用，把大消息拆碎，让其他消息有机会插进去。

规范还列出了几条必须遵守的约束

约束 具体含义
分片顺序必须保持 一条消息的各个分片，必须按发送顺序交付给接收方
不同消息的分片不能交错 一条消息的分片不得插入另一条消息的分片之间，除非协商了能解释交错语义的扩展
所有分片类型必须一致 一条消息的所有分片类型都相同，由第一个分片的操作码决定
双方都必须支持接收分片和未分片消息 客户端和服务端都必须具备接收这两种形态消息的能力
分片大小任意 对于非控制帧，发送方可以把消息切成任意大小的分片
中间件不得改变控制帧的分片 控制帧不能分片，因此中间件不得试图改变控制帧的分片方式

其中第二条值得特别注意：它意味着在一条消息的分片序列中间，你不能插入另一条消息的分片。 所以说，用分片实现多路复用的可能性是存在的，但是受这条规则约束，实现起来并不简单。

关于控制帧的插入规则，规范说的是：控制帧可以插入到一条分片消息的中间，但控制帧本身不允许被分片；而且接收方必须具备处理”分片消息中间夹着控制帧”这种情况的能力。

还有关于控制帧的一些规则：

控制帧可以插在分片消息的中间，而且接收方必须能处理这种情况。

  • 这个设计的理由规范也写了，用的是 Ping 做例子。大意是：如果控制帧不能被插队，那么当它排在一条大消息后面时，一个 Ping 的延迟就会非常长。所以说，如果控制帧必须排在一条大消息的所有分片之后，那么 Ping 的延迟就会非常长。所以说规范允许它们”插队”，代价是接收方必须能正确处理这种交错。

所有控制帧的载荷长度都必须在 125 字节以内，并且都不允许分片。

  • 125 字节上限适用于所有控制帧（Close、Ping、Pong 三类都是控制帧），而且都不允许分片。

    125 这个数字不是随便取的：它正好让控制帧的长度字段停留在 7 位范围内（0-125），不需要扩展到 16 位或 64 位。这让控制帧永远是定长头部，可以被快速解析和插入。

控制帧：Close、Ping、Pong

WebSocket 定义了三种控制帧，它们是这个协议保持连接健康的全部手段。规范对控制帧的定位是：它们用来传递关于 WebSocket 连接本身的状态，而不是业务数据，并且它们可以插入到分片消息中间。

Close（opcode 0x8）：一次必须成对的握手

Close 帧的设计要点，规范说的是：Close 帧可以带一个 body（也就是帧里的 Application data 部分）来说明关闭原因；如果带了这个 body，那么它的前两个字节必须是一个网络字节序的 2 字节无符号整数，用来表示状态码。

所以 Close 帧的载荷结构是：

1
2
3
+----------------+---------------------------------+
| 2 字节状态码   | 可选的 UTF-8 原因文本            |
+----------------+---------------------------------+

有三个规范细节容易被忽略：

  • Close 帧从客户端发往服务端时必须掩码，因为它是客户端发出的帧；
  • 原因文本不保证人类可读，而且客户端不允许把它展示给最终用户，因为服务端可能在里面塞调试信息甚至内部数据；
  • 如果 Close 帧里没有原因数据，Close Reason 就是空字符串。

关闭握手的流程是成对的。如果一个端点收到了 Close 帧，而它之前还没有发过 Close 帧，那么它必须回一个 Close 帧；当双方都完成了”发出”和”收到”Close 的动作之后，这个端点才认为 WebSocket 连接已关闭，并且必须关闭底层的 TCP 连接。服务端必须立即关闭底层 TCP 连接，而客户端应该等服务端来关。另外，应用在发出 Close 帧之后，不得再发送任何数据帧。

提炼成四条：

  1. 收到 Close 且尚未发过 Close，必须回一个 Close；
  2. 双方都完成收发的动作后，连接才算关闭，必须关闭底层 TCP 连接；
  3. 服务端必须立即关闭底层 TCP 连接；客户端应该等服务端先关（这样能避免客户端发出最后一个数据后，服务端还没来得及读就断了）；
  4. 发出 Close 后不能再发任何数据帧。

这个 服务端先关、客户端后关 的顺序设计很实用。它避免了那种经典的 客户端以为发完了就断开，服务端其实还有数据没读到 的竞态问题。

Ping（0x9）与 Pong（0xA）：活性探测

Ping/Pong 的语义很简单，规范的第一条是

  • 收到 Ping 帧之后，端点必须回一个 Pong 帧，除非它已经收到过 Close 帧；而且它应该在条件允许的情况下尽快回复。
  • 收到 Ping 必须回 Pong，除非已经收到了 Close 帧。 回复要尽快。已经进入关闭流程了就不用再回 Pong 了。

Pong 的内容必须与 Ping 一致，作为对某个 Ping 帧的响应而发出的 Pong 帧，它的 Application data 必须与它所回应的那个 Ping 帧的 body 完全相同。这样 Ping 的发送方可以通过载荷匹配来确认这是对我这个 Ping 的回应，而不是别人的。

规范还允许合并应答，这是一个优化。大意是：如果一个端点收到了 Ping，而它之前收到的那些 Ping 都还没回过 Pong，那么它可以选择只为最近处理的那个 Ping 发一个 Pong。如果对端连续发了好几个 Ping，你可以只回最后一个。 这在高频 Ping 的场景下能省掉不少无谓的往返。

最后这条我特别喜欢，因为它把”心跳”和”Ping”解耦了。规范说的是：Pong 帧可以被主动发送，不必有对应的 Ping；这种主动的 Pong 充当一种单向心跳，而且不期望对方回应。完全可以当心跳包用。

官方还说明了 Ping/Pong 的用途定位，大意是：Ping 帧既可以当作保活手段，也可以用来验证对端是否仍然有响应。

然而这两件事在实践中是不同的：保活是为了防止中间代理因为空闲而断连；探测响应是为了发现 TCP 连接看起来还在，但对端进程已经卡死 的情况（半开连接）。

浏览器侧的局限

这里有个实践中的现实问题必须说清楚：浏览器的 JavaScript WebSocket API 不暴露 Ping/Pong。 你在浏览器里拿不到发送 Ping 的能力，也监听不到收到的 Ping。

这带来的实际影响是：

  • 服务端主动发 Ping 是可行的：浏览器会自动回 Pong（这是协议层行为，不经过 JS），服务端据此判断客户端是否还活着；
  • 反过来不行：需要浏览器主动保活时，只能走应用层，发一条自定义的 JSON 心跳消息。

这个差异直接影响了后面的选型结论：WebSocket 虽然有心跳能力，但在浏览器侧你只能依赖服务端发起。 这也是为什么 STOMP 那种”应用层心跳”（发一个 EOL 空行）反而更实用，因为它在应用层，浏览器完全可以控制。

关闭状态码

关闭状态码是排查线上问题的重要线索，而且这里有两个很多人不知道的细节：哪些码不允许发送，以及哪些码其实不是 RFC 6455 定义的。

我把 RFC 6455 第 7.4.1 节定义的码值和 IANA 后补注册的码值分开列，这样不会混淆来源。

RFC 6455 定义的码：

状态码 名称 含义
1000 Normal Closure 正常关闭，建立连接的目的已经达成
1001 Going Away 端点”要走了”，比如服务端下线或浏览器离开页面
1002 Protocol Error 协议错误，端点因协议错误终止连接
1003 Unsupported Data 收到了无法接受的数据类型（比如只懂文本的端点收到了二进制）
1004 Reserved 保留，具体含义未来可能定义
1005 No Status Rcvd 不允许在帧里发送；用于表示”实际上没有状态码”
1006 Abnormal Closure 不允许在帧里发送；用于表示”未发送或接收 Close 帧就异常关闭”
1007 Invalid frame payload data 载荷数据与消息类型不符，比如文本消息里出现非 UTF-8 数据
1008 Policy Violation 收到的消息违反了策略，这是一个通用状态码
1009 Message Too Big 收到的消息太大，无法处理
1010 Mandatory Ext. 客户端专用：期望服务端协商一个或多个扩展，但服务端没有返回
1011 Internal Error 服务端专用：遇到未预期的情况，无法完成请求
1015 TLS Handshake 不允许在帧里发送；表示因 TLS 握手失败而关闭

关于 1010，规范有一句特别说明：这个状态码不会被服务端使用，因为服务端有更直接的做法，它可以干脆让握手失败。所以1010 是客户端专用的，因为服务端如果不同意扩展，它可以在握手阶段直接拒绝，不需要专门发一个 1010。这句话解释了”为什么这个码只有客户端会用”。

顺便说一下 1003 和 1007 的区别，这两个很容易混：1003 说的是”数据类型”不被接受（比如只懂文本的端点收到二进制帧），1007 说的是”数据内容和消息类型不一致”（比如一个文本帧里塞了非法的 UTF-8 字节）。一个是类型层面，一个是内容层面。

IANA 后补注册的码（不属于 RFC 6455）：

状态码 名称 含义
1012 Service Restart 服务重启
1013 Try Again Later 服务暂时过载，请稍后重试
1014 Bad Gateway 服务端作为网关或代理时，从上游收到了无效响应（类似 HTTP 502）

它们是通过 IANA 的 WebSocket Close Code Number Registry 后补注册的扩展码。

除这些已定义的码之外，规范还划了保留区段，我按区段说明它们的用途：

区段 能不能用
0-999 不使用
1000-2999 保留给本协议、其未来修订版，以及那些写在永久且公开可得的规范里的扩展来定义
3000-3999 保留给库、框架和应用使用，这些码直接向 IANA 注册
4000-4999 保留给私有用途，因此不能被注册；可以按双方事先约定来使用

所以说，如果你想给业务定义一个专属的关闭原因，用 4000-4999 区间最省事，因为不需要注册，比如 4001 表示”登录态失效”、4002 表示”被其他设备踢下线”。

最后是关键的使用约束：1005、1006、1015 这三个码不允许作为 Close 帧的状态码发送。

官方对它们用的是同一句措辞：端点不得把这三个值中的任何一个，作为状态码放进 Close 控制帧里发送。 它们的作用是在应用层表示”本地推导出的关闭原因”：

  • 1005：“我收到了 Close 帧，但里面没有状态码”。规范规定，如果 Close 控制帧里不含状态码，那么”WebSocket 连接关闭码”就视为 1005；
  • 1006：“连接被关闭了，而且我根本没有收到过 Close 帧”。规范规定，如果连接已关闭而端点没有收到 Close 控制帧（比如底层传输连接直接断了），那么关闭码就视为 1006；
  • 1015：“因为 TLS 握手失败而关闭”。

在 Spring 里，服务端主动关闭时可以指定状态码。Spring 的 ExceptionWebSocketHandlerDecorator 在处理未捕获异常时，就是用 1011 关闭会话的。Spring 文档对这一行为的描述是，这个装饰器会捕获所有从 WebSocketHandler 方法里逃逸出来的异常，并以 1011 状态码关闭会话，表示服务端错误。

这个默认行为很合理：出错了要让客户端明确知道 是服务端的问题，不是客户端自己发错了。如果你自己实现限流、鉴权失效等逻辑，也应该主动用合适的码关闭，而不是让连接莫名其妙地断掉。

WebSocket 在 HTTP/2 和 HTTP/3 上怎么办？

HTTP/2 因为多路复用的特性，不允许有”连接级”的头部字段和状态码，而 Upgrade、Connection、101 恰好都是连接级的——这与 HTTP/2 的设计前提直接冲突。所以 RFC 8441 另辟蹊径，用 HTTP/2 的扩展 CONNECT 方法（Extended CONNECT）来引导 WebSocket。

它的机制有几个要点。

  • 第一，用 :protocol 伪首部指定目标协议。 规范允许在 CONNECT 请求的 HEADERS 上加一个 :protocol 伪首部，用来指明这条隧道上准备说哪种协议。对 WebSocket 来说，这个值必须是 websocket。和 HTTP/1.1 的 Sec-WebSocket-Protocol 不同，它是单值的。
  • 第二，服务端必须先通告能力。 这就是 SETTINGS_ENABLE_CONNECT_PROTOCOL 设置项，它的取值必须是 0 或 1。规范说，客户端在收到值为 1 的这个设置之后，才可以在新建流时使用扩展 CONNECT。而且发送方不得在已经发过 1 之后，再发送值为 0 的同一个设置项。所以说，一旦开了就不能关，避免能力反复变化造成状态不一致。
  • 第三，握手校验机制被替换掉了。 规范明确说：使用扩展 CONNECT 来引导 WebSocket 的实现，不再处理 RFC 6455 里的 Sec-WebSocket-Key 和 Sec-WebSocket-Accept，因为这套功能已经被 :protocol 伪首部取代了。这很合理：HTTP/2 有自己严格的首部校验机制

同时 Connection、Upgrade、Host 这些头在 HTTP/2 上是不得包含的，而 Origin、Sec-WebSocket-Version、Sec-WebSocket-Protocol、Sec-WebSocket-Extensions 这些仍然沿用。

在 HTTP/3 上，RFC 9220 做了同样的适配。其实在 HTTP/2 单条流上跑 WebSocket”的机制，同样适用于 HTTP/3，只是与 HTTP 版本相关的细节需要单独规定。语义与 HTTP/2 完全一致，SETTINGS_ENABLE_CONNECT_PROTOCOL 的值同样是 0x08。RFC 9220 里还有一条值得知道的细节：如果服务端宣称支持扩展 CONNECT，但收到的请求里 :protocol 是个未知或不支持的值，那它应该回一个 501（Not Implemented）。

HTTP/2 上的 WebSocket 并没有解决 TCP 层队头阻塞，因为一条 WebSocket 会话仍然映射到一条 HTTP/2 流，而所有流都跑在同一条 TCP 连接上。它带来的好处是：

  • 省掉一次独立的 TCP/TLS 连接建立；
  • 让 WebSocket 流量和其他 HTTP 流量共享一条连接，更好地利用连接资源；
  • 在 HTTP/2 环境下不再受 HTTP/1.1 每域名 6 个连接的限制。

SSE 规范与 EventSource

如果说 WebSocket 是另起炉灶，那 SSE（Server-Sent Events，服务器发送事件）就是就地改造。它选择的路线是：不改变 HTTP 的传输方式，只在 HTTP 之上定义一套严格的消息格式和客户端 API。

事件流格式

MIME 类型与编码

SSE 的规范开篇就定死了两件事：事件流格式的 MIME 类型是 text/event-stream，而且这种格式的事件流必须始终以 UTF-8 编码。

  • MIME 类型必须是 text/event-stream；
  • 必须是 UTF-8 编码，没有其他选择。它还规定解码时用的是标准的 UTF-8 解码算法

这一点在中文场景下需要留意：服务端如果按 GBK 写出流，浏览器解析出来的就是乱码，而且不会给你报错，只会静默乱码。Spring 侧只要正确设置了 produces = MediaType.TEXT_EVENT_STREAM_VALUE，框架会帮你处理编码。

ABNF 语法

规范给出的完整 ABNF 定义如下：

1
2
3
4
5
stream        = [ bom ] *event
event         = *( comment / field ) end-of-line
comment       = colon *any-char end-of-line
field         = 1*name-char [ colon [ space ] *any-char ] end-of-line
end-of-line   = ( cr lf / cr / lf )

拆解一下：

  • stream 可以由一个可选的 BOM 开头，后面跟零个或多个 event；
  • 一个 event 由若干 comment 或 field 行，加上一个 end-of-line（空行）组成；
  • comment 是以冒号 : 开头的行；
  • field 的格式是”字段名 + 冒号 + 可选空格 + 内容”；
  • 行分隔符可以是 CRLF、单独的 CR 或单独的 LF——三种都接受。
  • 字段名是逐字比较的，SSE 的字段名必须全小写。

四个标准字段

字段 作用
data 事件的数据内容，可重复出现
event 事件类型名，默认是 message
id 事件 ID，会写入浏览器的 last event ID
retry 重连时间，必须是 ASCII 数字

再加上以冒号开头的注释行（comment）。这就是 SSE 的全部语法了，非常简单。

data 的拼接规则

data 可以被拆成多行，拼接规则在规范里有明确的示例。看这个流：

1
2
3
data: YHOO
data: +2
data: 10

规范说它会产生一个 message 事件，而 event.data 的值是 "YHOO\n+2\n10"（其中的 \n 表示换行符）。

也就是：

  1. 每个 data 字段的值按出现顺序用换行符 LF 连接；
  2. 最后那个换行符会被去掉。

规范里对这两步的表述很精确：每处理一个 data 字段，就把它的值追加到 data 缓冲区，同时追加一个 LF 字符；等到真正派发事件时，如果缓冲区最后一个字符是 LF，就把这个字符去掉。所以是”每个 data 后面都加一个 LF，最后再删掉一个”，净效果就是多行之间用 LF 连接、末尾不留 LF。

这个规则的实际影响很大。假设你要推送一段格式化过的 JSON（带缩进、带换行），你不能直接把 JSON 塞在 data: 后面，因为一旦内容里有真实的换行符，解析就断了。正确做法是每一行都加一个 data: 前缀，让多个 data 字段来承载多行内容：

1
2
3
4
data: {
data:   "orderId": "A1001",
data:   "status": "PAID"
data: }

客户端拿到的 event.data 是：

1
2
3
4
{
  "orderId": "A1001",
  "status": "PAID"
}

这个字符串可以直接被 JSON.parse 解析。这也是为什么 SSE 推送 JSON 时，服务端必须做多行拆分成多个 data 字段的处理，而不是简单拼一个 data: 前缀。 忘了这一点，你的 JSON 就会在第一个换行处被截断。

空行才是消息边界

最后一条事件也必须以一个空行结尾，流的结束本身不足以触发派发。

所以你在写 SSE 服务端时，一次完整的事件发送必须是：

1
2
data: 内容\n
\n

两个换行，缺一不可。这也是很多人第一次自己写 SSE 遇到的现象：数据明明发出去了，客户端就是不触发 onmessage。

规范里还有一个例子说明了这点：

1
2
3
4
5
6
data

data
data

data:
  • 第一块触发一个 data 为空字符串的事件；
  • 中间一块触发一个 data 为单个换行符的事件（因为两个 data 字段，第一个为空）；
  • 最后一块被丢弃，因为它后面没有跟空行。

顺便说一下这里隐含的两条规则。

  • 空的 data 块不派发事件，如果派发时 data 缓冲区是空字符串，就清空缓冲区直接返回，什么都不发（这一点我在后面还会单独展开）。
  • 流中途被截断时，未完成的事件直接丢弃，SSE 的原意是：读到文件末尾时，任何挂起的数据都必须丢弃；如果文件在一条事件中间结束，也就是还没到最后的空行，那么这条不完整的事件不会被派发。

冒号后的空格只忽略一个

1
2
3
data:test

data: test

规范说这两个流会产生完全相同的两个事件，原因是冒号后面的空格如果存在，会被忽略。

只去掉一个前导空格，不是把空白都 trim 掉。 这个区别在解析时很关键：

写法 解析出的 data 值
data:abc abc
data: abc abc
data: abc abc（开头有一个空格）
data: abc abc（开头有两个空格）

所以如果你需要推送的内容本身以空格开头，比如 Markdown 里的缩进代码块，必须自己多写一个空格来抵消被忽略的那一个。这个细节在网络协议实现里很常见。

另外，一个只有 data 而没有冒号的行也是合法的。规范里的规则是，如果这一行非空且不含冒号，那么整行就是字段名，字段值是空字符串。 所以裸 data 等价于 data:。

注释行的处理

以冒号开头的行是注释行。规范对它的处理只有一个动作：直接忽略整行。 它不进入任何缓冲区，不产生事件，也不影响后面的解析，这正是心跳该用的形式。

完整的示例流

规范给了一个包含注释、id、event 和多行 data 的综合示例，我照着它加上了中文注释：

1
2
3
4
5
6
7
8
9
10
11
12
: test stream
              <- 这一行是注释，以冒号开头，本身不触发任何事件

data: first event
id: 1
              <- 空行，触发第一个事件（类型默认 message，data = "first event"，last event ID = "1"）

data:second event
id
              <- id 字段没有值，把 last event ID 重置为空字符串

data:  third event

最后一个事件要注意：data: third event 里冒号后面有两个空格，按规则只忽略一个，所以它的 data 值是一个以单个空格开头的内容，也就是 " third event"。

EventSource 接口与自动重连

浏览器端的连接保持

SSE 最大的特点不在格式，而在浏览器原生帮你做重连。这一点用普通 fetch 流式读取是做不到的，你必须自己写重连逻辑。

客户端使用起来极其简单：

1
2
3
4
5
6
7
8
9
10
const source = new EventSource('/api/notifications/stream');

source.onopen = () => console.log('连接已建立');
source.onmessage = (event) => {
  console.log('收到默认 message 事件', event.data);
};
source.onerror = (event) => {
  // 浏览器会自动重连，这里只做状态提示
  console.warn('连接异常，readyState =', source.readyState);
};

如果要区分事件类型，用服务端的 event 字段配合 addEventListener：

1
2
3
4
5
6
7
8
source.addEventListener('order-paid', (event) => {
  const payload = JSON.parse(event.data);
  updateOrder(payload.orderId, payload.status);
}, false);

source.addEventListener('stock-changed', (event) => {
  // ...
}, false);

另外，规范定义 EventSource 时可以传第二个参数 { withCredentials: true }，它会把连接请求的凭证模式设为 include，也就是跨域时也带上 Cookie。这个选项在做前后端分离部署时经常需要。

响应必须是 200 且 Content-Type 正确

这一条是 SSE 服务端最基础的硬性要求，规范里的原意是：如果响应的状态码不是 200，或者响应的 Content-Type 不是 text/event-stream，就判定连接失败。这不是重试，而是失败

而这个机制恰好解释了一个非常好用的技巧，用 204 让浏览器停下来。如果服务端返回 204 No Content，浏览器就不会再重连，并且会把 readyState 置为 CLOSED。

这是 SSE 一个非常好用但常被忽略的能力。因为客户端会在连接关闭后重连；而服务端可以用 HTTP 204 No Content 这个响应码来告诉它不要再重连了。但是其实，任何非 200 响应都会让浏览器放弃重连。所以说，不要用 5xx 来表达”暂时不可用”。如果你在服务端限流时返回 503，浏览器不会重试，而是彻底放弃。如果你希望客户端重试，正确的做法是建立连接后立刻关闭，这会触发正常的重连流程。

重定向

规范里说：事件流请求可以像普通 HTTP 请求一样，用 HTTP 301 和 307 重定向。

301 和 307 重定向都是支持的，行为和普通 HTTP 请求一致。这里要留意 301 是永久重定向，浏览器可能缓存它；滚动发布时用 307 更安全。

其中，事件源的 origin 取的是重定向之后最终 URL 的 origin。规范明确说是重定向之后的 URL。如果你把 SSE 从 a.example.com 重定向到 b.example.com，那么客户端收到的事件 origin 就是 b.example.com，跨域校验也按这个来。

readyState 的三个状态

EventSource 的 readyState 有三种状态，它们的含义在规范里定义得很清楚：

常量 数值 含义
CONNECTING 0 连接尚未建立，或者曾经关闭、用户代理正在重连
OPEN 1 连接已打开，正在派发收到的消息
CLOSED 2 连接已关闭且不会重连（发生致命错误，或调用了 close()）

这里有个重要的实践价值：readyState 是区分临时抖动和请求真的寄了的唯一手段。

  • 如果是 0，说明浏览器正在自动重连，你什么都不用做；
  • 如果是 2，说明这次是真的不重试了，通常是因为服务端返回了 204，或者调用了 close()。

EventSource 还提供了 close() 方法，它的作用是中止所有为这个对象发起的 fetch 过程，并把 readyState 置为 CLOSED。注意它和 浏览器自己判定连接失败 的效果是一样的，都是进入 CLOSED，都不会再重连。

重连算法与 Last-Event-ID

SSE 规范里定义的重连状态是这样的：一个”重连时间”（单位毫秒），它初始必须是一个由实现决定的取值，很可能在几秒这个量级；还有一个最后事件 ID 字符串，它初始必须是空字符串。

注意两点：

  • 初始重连时间是实现自定义的，规范只给了个大概范围：大概几秒量级；
  • 初始的 last event ID 是空字符串。

而重连的完整步骤，规范是这么写的：

  1. 排队一个任务：如果 readyState 已经是 CLOSED 就中止；否则把 readyState 设为 CONNECTING，并派发一个 error 事件；
  2. 等待一个等于重连时间的延迟；
  3. 可选地再等一会儿。 如果上一次尝试失败了，用户代理可能会引入一个指数退避的延迟，以避免给一个可能已经过载的服务器雪上加霜；另一种情况是，如果操作系统报告当前没有网络连接，用户代理可能会等操作系统通告网络恢复之后再重试；
  4. 等前面那个任务执行完；
  5. 排队一个任务：如果 readyState 不是 CONNECTING 就返回；否则如果 last event ID string 非空，就把它以 UTF-8 编码后设置为请求头的 Last-Event-ID，然后重新发起 Fetch。

续传部分才是 SSE 真正有价值的设计。服务端可以在事件里带上 id 字段：

1
2
id: 10086
data: {"orderId":"A1001","status":"PAID"}

浏览器收到后会把 10086 记为”last event ID string”。当连接断开、浏览器重新发起请求时，它会自动把这个值放进请求头：

1
2
3
GET /api/notifications/stream HTTP/1.1
Accept: text/event-stream
Last-Event-ID: 10086

服务端拿到 Last-Event-ID，就可以从 10086 之后继续推送，实现 断线不丢消息。

关于 id 字段还有三个细节，规范里都写明了：

  • id 字段的值如果含有 NUL 字符，整个字段会被忽略。只有当字段值不含 NUL 时才把它写入 last event ID 缓冲区，否则就忽略这个字段；而且Last-Event-ID 头只在 last event ID 字符串非空时才会被发送，空字符串时不发这个头。
  • id 字段没有值时，会把 last event ID 缓冲区置为空字符串，此后重连就不再发送 Last-Event-ID 头了。
  • 如果某条事件没有 id 字段，而之前的事件设置过，那么这条事件的 lastEventId 仍然是之前那个值。 也就是说 last event ID 是”粘住”的，规范明确说那个缓冲区不会被重置。这一条很实用：你不需要每条事件都带 id。

data 为空的块不派发事件

如果 data 缓冲区是空字符串，就把 data 缓冲区和事件类型缓冲区都置为空，然后直接返回。

也就是说，下面这个流：

1
2
3
event: ping

data: real-content

第一个块（只有 event: ping，没有 data）不会触发任何事件。空行只是把缓冲区清空而已。

这个规则解释了一个常见的困惑：“我明明发了心跳事件，为什么前端 addEventListener('ping') 一次都没触发？”——因为心跳块里没有 data 字段，浏览器认为它是空的，直接跳过。

心跳要用注释行（: 开头），而不是空的 event: 行。注释行在规范里的处理是忽略这一行，它既不产生事件，也不会干扰后面的解析，这才是心跳的正确形式。

retry 字段的严格性

retry 字段用于告诉浏览器下次重连等多久，单位是毫秒：

1
retry: 3000

值必须”只由 ASCII 数字组成”，否则整个字段被忽略。这意味着下面这些写法全都无效，而且不会有任何报错：

1
2
3
4
5
retry: 3s          ← 含非数字字符，忽略
retry: 3000ms      ← 含非数字字符，忽略
retry: -1          ← 含符号，忽略
retry: 3 000       ← 含空格，忽略
retry: 0x0BB8      ← 含字母，忽略

正确的写法就是一个纯数字：retry: 3000。

值得一提的是，浏览器会在响应体的 SSE 数据流里解析 retry 字段，这和其他协议里”用响应头配置客户端行为”的习惯不同，所以如果你想调整重连间隔，写在事件流里就行，不要去找某个 HTTP 响应头。

生产环境的注意事项

规范里有一节叫 Authoring notes，也就是”给实现者的提醒”。这一节的内容含金量很高，几乎每一条都对应一个我见过的线上问题。

老代理会偷偷断流，所以要定期发注释心跳

众所周知，一些遗留代理服务器在某些情况下会在较短超时之后丢弃 HTTP 连接；为了防范这类代理，作者可以每隔 15 秒左右加一个注释行（也就是以冒号开头的那种行）。来作为保活心跳

在 Spring 里对应 SSE 的写法就是定期 emitter.send(SseEmitter.event().comment("keep-alive"))，具体代码我在后面写。

这个数字（15 秒）非常值得记住。因为 nginx 的 proxy_read_timeout 默认是 60 秒；

规范在这一节里还提了一个同源的坑。它提醒说：HTTP 分块（chunking）可能对这套协议的可靠性产生意料之外的负面影响，尤其是当分块是由另一个不了解其时序要求的层来做的时候；如果这成了问题，可以在提供事件流时关闭分块。

也就是说，HTTP 分块如果由不理解 SSE 时序要求的中间层来做，可能带来可靠性问题。 实践中的典型表现就是 nginx 的 proxy_buffering（默认开启）把数据攒在缓冲区里批量下发，这个问题我在讲网关时会专门讲。

还有一条关于缓冲的提醒，规范是针对用户代理说的，但同样适用于服务端的写入逻辑：这类资源的连接预期是长期存在的，用户代理应当确保使用了合适的缓冲；具体来说，“以单个 LF 作为行结束”的行缓冲是安全的，而块缓冲、或者按不同行结束符预期的行缓冲，都可能导致事件派发出现延迟。

块缓冲会导致事件派发延迟。 如果你把响应包装在一个块缓冲的 writer 里，就会破坏实时性。

不要用 IP 地址关联连接和用户

想把多个事件源连接互相关联、或者把连接和之前提供的某个文档关联起来的作者，可能会发现依赖 IP 地址是行不通的，因为单个客户端可能有多个 IP 地址（它走了多个代理服务器），而单个 IP 地址也可能对应多个客户端（它们共用一个代理服务器）。更好的做法是在提供文档时给它带一个唯一标识符，然后在建立连接时把这个标识符作为 URL 的一部分传过去。

因为单个客户端可能有多个 IP（走了多级代理），单个 IP 也可能对应多个客户端（共享出口）。 所以：

  • 不要用 IP 做 SSE 会话的唯一标识；
  • 不要用 IP 做限流维度的唯一依据（会把同一个公司出口后面的所有用户当成一个人）；
  • 正确做法是在页面渲染时生成一个唯一标识，建立 SSE 连接时作为 URL 参数带上。

这个建议不仅对 SSE 成立，对 WebSocket 的会话管理同样成立，对其他的类似的会话管理也是这样。

浏览器连接数限制的现实影响

关于连接数的提醒，那些支持”每服务器连接数限制”的客户端，在一个站点打开多个页面时可能遇到麻烦，因为如果每个页面都和同一个域建了一个 EventSource 的话。作者可以规避这个问题，办法包括：为每条连接使用不同的域名（这个机制相对复杂）、允许用户按页面启用或禁用 EventSource 功能、或者用一个共享 worker 共享单个 EventSource 对象。

这个数字可以从 Chromium 的源码里得到最硬的一手证据。在 net/socket/client_socket_pool_manager.cc 里，连接池的每主机上限是这样定义的：

1
2
3
4
5
6
7
8
// Default to allow up to 6 connections per host. Experiment and tuning may
// try other values (greater than 0).  Too large may cause many problems, such
// as home routers blocking the connections!?!?  See http://crbug.com/12066.
std::array<size_t, kSocketPoolTypesSize> g_max_sockets_per_group =
    std::to_array<size_t>({
        6,   // kNormal，史上最神秘的6
        255  // kWebSocket
    });

这里有两个信息同样重要，但第二个经常被忽略：

  1. HTTP/1.1 普通连接的每主机默认上限是 6，注释还解释了为什么不设更大——太大可能被家用路由器阻断连接；
  2. WebSocket 走的是独立的连接池，上限是 255，不是 6。

WebSocket 连接是长期存在的，应当和普通连接区别对待；这里用 255 作为上限，这样 wss 的限制就和 ws 一致。另外注意 Firefox 用的上限是 200。而 Firefox 用的上限是 200。所以这个数字是各浏览器自己定的，不统一。

MDN 的 EventSource 文档页也给出了数字 6，并且明确了它的作用域，大意是：这个限制是按”浏览器 + 域名”计的，也就是说你可以跨所有标签页对 www.example1.com 打开 6 条 SSE 连接，同时对 www.example2.com 再开 6 条，也就是这个配额是跨标签页共享的。所以：

  • 用户开 6 个标签页，每个标签页都建一个 SSE 连接，配额就满了；
  • 第 7 个标签页的 SSE 连接会被排队阻塞，页面看起来”卡住了”；
  • 更糟的是，这些被占用的连接会挤压该域名下其他普通 XHR 请求的配额，导致整个站点都变慢。

MDN 还给出了 HTTP/2 下的情况，使用 HTTP/2 时，同时打开的 HTTP 流数量上限由服务端和客户端协商决定，默认是 100。

所以说 HTTP/2 下不再有 6 连接的限制，取而代之的是”并发流的数量”，默认 100 且在服务端和客户端之间协商。这也是把 SSE 部署在 HTTP/2 网关后面的一个额外收益。

规避手段一般这样：

  1. 在同一个标签页内复用同一个 EventSource 全局实例（最简单也最有效，用 event 字段区分业务类型）；
  2. 用 SharedWorker 在所有标签页之间共享一个 EventSource（规范明示的方案）；
  3. 升到 HTTP/2（不改代码就能缓解）；
  4. 为每个连接使用不同域名（规范说这个机制”相对复杂”，一般不建议）。

一般情况下第一条是必须做的，因为绝不因为”页面上有三个模块需要实时数据”就开三条 SSE 连接。

WebSocket 与 SSE 的对比与选型

十个维度的正面对比

维度 WebSocket SSE
通信方向 全双工，双向对等 单向，只能服务端到客户端
底层协议 独立协议（RFC 6455），借 HTTP 完成握手后脱离 HTTP 就是 HTTP，长连接持续响应
握手方式 GET + Upgrade: websocket，服务端回 101 Switching Protocols 普通 GET，服务端回 200 并保持响应打开
数据格式 二进制帧，文本或二进制载荷，格式由应用自定义 text/event-stream，UTF-8 文本，有严格 ABNF
消息边界 协议层用 FIN + opcode 界定消息边界 应用层用空行界定事件边界
自动重连 无，必须自己实现 浏览器原生自动重连，可能带退避
断点续传 无原生机制 有，靠 id 字段 + Last-Event-ID 请求头
保活机制 协议级 Ping/Pong 控制帧 无协议级机制，靠注释行心跳（规范建议 15 秒）
浏览器 API WebSocket，事件驱动，需自己处理重连 EventSource，声明式，重连内建
二进制支持 原生支持，0x2 二进制帧 不支持，只能文本（二进制需 base64 编码，体积膨胀约 33%）
数据压缩 有标准扩展 permessage-deflate（用 RSV1 位标记） 无标准扩展，靠 HTTP 层 gzip（对长连接流式内容效果有限）
消息头开销 服务端到客户端 2 字节起（短消息） 每个事件至少 data: + 换行 + 空行，约 8-10 字节起
HTTP 语义 握手后不再有 HTTP 语义，状态码、Cookie、缓存都失效 完全保留 HTTP 语义，Cookie、CORS、鉴权、重定向、状态码全部可用
代理/网关兼容 需要专门配置转发 Upgrade，部分企业代理直接不支持 就是一个普通 HTTP 响应，只要关掉缓冲就能用
连接数限制 不受 HTTP/1.1 每域名 6 连接限制影响（升级后已不是 HTTP） HTTP/1.1 下受每域名 6 连接限制；HTTP/2 下无此问题
服务端资源占用 每连接一个会话对象 + 容器缓冲区（Tomcat 默认文本/二进制各 8KB） 每连接一个 Servlet 异步请求 + 一个 emitter 对象
跨域控制 同源策略不生效，必须服务端主动校验 Origin 受 CORS 约束，EventSource 可配 withCredentials
浏览器兼容 IE10+，所有现代浏览器 IE 全系列不支持，其他现代浏览器全部支持
实现复杂度 高（握手、路由、订阅管理、重连、心跳都要自己考虑） 低（MVC 里返回一个 SseEmitter 就行）

这张表信息量比较大，把更关键的差异展开说一下

HTTP 语义的保留

如果只能记住一条差异，我建议记住这条。

SSE 是一个普通 HTTP 响应。 这句话听起来平淡，但它的工程含义非常巨大：

  • 鉴权体系完全不用改。 你的 JWT 拦截器、Session Cookie、网关的认证过滤器，全都原样生效，因为这就是一个 GET 请求。
  • CORS 机制原样生效。 跨域时配置 Access-Control-Allow-Origin 就行，和普通接口一模一样。
  • HTTP 状态码语义可用。 401 拒绝连接、403 无权限、204 让客户端停止重连、307 重定向到另一个节点——这些在 SSE 里都是优雅且标准的表达方式。
  • 网关、日志、监控不用动。 访问日志里能看到这个请求，APM 能追踪它，压测工具能打它。

而 WebSocket 握手之后，这一切都消失了：

  • 握手完成后连接里流动的就不再是 HTTP 了，标准 HTTP 中间件看不到”请求”，只看到一堆字节；
  • 你的 Spring Security 过滤器链只在握手那一次执行，之后的消息不会再经过它——这就是为什么 Spring Security 需要专门提供 @EnableWebSocketSecurity 和 ChannelInterceptor 来做消息级授权；
  • 网关需要显式配置才能正确转发升级请求，配置错了会静默降级成普通 HTTP 导致连接失败；
  • APM 工具通常看不到 WebSocket 里的”逐条消息”，只能看到连接建立和关闭。

Spring Security 官方文档里有一句很关键的说明：Spring Security 只保护 clientInboundChannel，它并不试图保护 clientOutboundChannel。它给出的理由是性能，每一条入站消息通常对应多条出站消息；而它给出的建议是：与其去保护出站消息，不如把订阅这个动作保护好。

Spring Security 只管入站消息，不管出站消息。这就是 WebSocket 强大但昂贵的具体体现：你获得了双向能力，代价是要自己重建一整套在 HTTP 里的安全与治理能力。

重连与状态恢复

SSE 的自动重连 + Last-Event-ID 组合，本质上提供了一套消息可靠性原语。

考虑这个场景：用户正在看一个实时订单看板，网络抖动了 3 秒。

用 SSE 时：

  1. 浏览器检测到连接断开，readyState 变成 CONNECTING；
  2. 等待退避时间后自动重连，带上 Last-Event-ID: 10086；
  3. 服务端从 10086 之后的事件开始补发；
  4. 用户端看不到任何异常，数据也没有缺口。

用 WebSocket 时：

  1. 客户端收到 onclose（或者干脆因为网络中断，什么都没收到，只能靠心跳超时发现）；
  2. 你需要在 onclose 里实现退避重连（否则所有客户端会同时重连，打出一波流量尖峰）；
  3. 重连成功后，你需要在应用协议里加一个”同步”消息：客户端告诉服务端”我最后收到的消息序号是 10086”，服务端补发差额；
  4. 这套逻辑你得自己设计、自己实现、自己测试。

这个差异在业务上意味着什么？如果你做的是”通知提醒”、“行情推送”、“进度同步”，这类允许重连后补齐的场景，SSE 省掉的工作量是实打实的。而如果你做的是”聊天”、“实时协同”、“游戏操作”，这类每一帧都要求即时且不可重放的场景，那你需要的是 WebSocket 的低延迟双向通道。

选型方法

我在实践中总结出三个问题，按顺序问下去，基本能确定方向。

graph TD
    A[需要服务端主动推送] --> B{客户端需要频繁上行吗?}
    B -- 是, 且上行量级与下行相当 --> C[WebSocket]
    B -- 否, 或只有偶尔的控制指令 --> D{单个连接的消息频率高吗?}
    D -- 高, 如行情/协同/游戏 --> C
    D -- 低到中, 如通知/进度/公告 --> E{服务端到客户端是单向语义吗?}
    E -- 是 --> F[SSE 优先]
    E -- 否, 需要真正的双向会话 --> C
    C --> G{能否接受改造网关/鉴权/监控?}
    G -- 否 --> F
    G -- 是 --> H[WebSocket + STOMP]

第一个问题：客户端需要频繁上行吗？

这是最关键的一刀。

  • 需要：聊天、协同编辑、在线游戏、实时协作白板、远程控制。这些场景的特征是”客户端产生的消息量和时间敏感度与下行相当”，用 SSE 模拟就要在下面再开一条 POST 通道，两条通道的时序无法保证，反而更复杂。选 WebSocket。
  • 不需要：通知、公告、行情展示、任务进度、日志滚动、状态看板。客户端最多偶尔发个”订阅某只股票”的控制指令，用普通 POST 就够了。选 SSE。

上行消息是否需要和下行消息共享顺序保证？如果需要（比如聊天里”我发了一条消息，紧接着收到别人的回复”，顺序必须正确），那么两条独立通道会带来麻烦，WebSocket 更合适；如果不需要，SSE + POST 完全可行。

第二个问题：消息频率和体量有多大？

Spring 官方认为：只有”低延迟 + 高频 + 大流量”三者叠加，才是 WebSocket 最合适的场合。

  • 低频：短轮询就够了，别上长连接。为了一条消息维持一个长连接，成本反而更高。
  • 中频：SSE 最划算。连接开销低、实现简单、有自动重连。
  • 高频大流量：WebSocket 更合适，因为它的帧头开销小（短消息只要 2 字节）、支持二进制（不用 base64 膨胀）、支持 permessage-deflate 压缩。

第三个问题：你的基础设施能承受多大的改造成本？

这是个现实问题，但往往被忽略。

SSE 需要改的：

  • nginx 关掉 proxy_buffering（或者应用返回 X-Accel-Buffering: no）；
  • 调大 proxy_read_timeout 并配合心跳；
  • 容器异步超时时间要设够。

WebSocket 需要改的：

  • 上面全部，再加上：nginx 配置 Upgrade 和 Connection 头转发（如果漏了，连接会直接降级失败）；
  • 负载均衡要支持 WebSocket（四层或七层都要配置）；
  • 鉴权要从 HTTP 过滤器体系迁移到握手 + 消息两层；
  • 监控要新增 WebSocket 会话指标；
  • 如果要做多实例广播，还得引入消息代理。

三个常见误区

最后澄清三个我在讨论中经常遇到的误区。

  • SSE 是过时技术，WebSocket 更先进。

    不是。它们是同一时期（2011 年前后）标准化的两套方案，解决不同问题。SSE 不是”WebSocket 的简化版”，它有自己的独门能力——自动重连和断点续传是 WebSocket 里没有对应物的。而且 SSE 在 HTTP/2 时代反而更有价值，因为多路复用让它彻底摆脱了连接数限制。

  • SSE 只能服务端单向，所以功能弱。

    单向是 SSE 的设计边界，不是缺陷。事实上，绝大多数实时推送需求本质上就是单向的，通知、告警、行情、进度、日志、大屏。强行用双向协议做单向需求，带来的只是额外的复杂度和安全负担。

    而且单向只针对推送通道。你完全可以在 SSE 之外用普通的 REST 接口做上行，这在很多场景下比在长连接里做上行更好：上行天然是请求响应语义，用 HTTP 表达最自然。

  • SSE 是纯文本，所以性能一定比 WebSocket 差。

    这个结论要分场景看：

    • 对于高频小消息，WebSocket 的 2 字节帧头确实优于 SSE 的约 10 字节事件开销，而且 SSE 的事件开销随 data 行数线性增长（每个换行都要加一个 data: 前缀）；
    • 对于低频消息，两者的开销都可以忽略，此时 SSE 的实现简单度和可靠性优势更重要；
    • 对于二进制数据，WebSocket 完胜，因为 SSE 必须 base64（体积膨胀约 33%）。

    所以正确的说法是：在高频场景下 WebSocket 的性能优势明显，在低频场景下这个差异可以忽略。 不要为了性能去选一个用不上的复杂度。

Spring 下的 WebSocket 实战

从这一部分开始进入落地。

路线选择：原生 WebSocket 还是 STOMP

Spring 7 提供了两条平行的 WebSocket 路线，它们解决的是不同层次的问题。

对比项 原生 WebSocket STOMP over WebSocket
启用注解 @EnableWebSocket @EnableWebSocketMessageBroker
核心抽象 WebSocketHandler / WebSocketSession Message / MessageChannel / SimpMessagingTemplate
消息语义 无，文本或二进制，格式自定义 有，STOMP 帧 + destination 路由
消息路由 只有连接级路由（一个 URL 一个 handler） 目的地级路由（/app/xxx 到方法，/topic/xxx 到订阅）
订阅管理 需要自己实现 内建（simple broker 或外部 broker）
用户维度寻址 需要自己维护 session 与用户的映射 内建（/user/ 前缀 + SimpMessagingTemplate）
安全集成 只有握手鉴权 握手鉴权 + 消息级鉴权 + 订阅级鉴权
多实例广播 需要自己实现 通过外部 broker 中继实现
上手成本 低 中
适用场景 简单双向通信、自定义协议 企业级消息推送（推荐）

除了一两个端点的极简场景，企业项目都应该直接上 STOMP。 但原生 API 也必须理解，因为 STOMP 是构建在它之上的。

原生 WebSocket：最小可用实现

写一个 Handler

Spring 的做法是实现 WebSocketHandler，或者更方便地继承 TextWebSocketHandler（只处理文本）或 BinaryWebSocketHandler（只处理二进制）。官方示例长这样：

1
2
3
4
5
6
7
public class MyHandler extends TextWebSocketHandler {

    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) {
        // ...
    }
}

TextWebSocketHandler 提供了一组可以覆盖的回调方法，覆盖了连接的生命周期：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
public class NotifyWebSocketHandler extends TextWebSocketHandler {

    private static final Logger log = LoggerFactory.getLogger(NotifyWebSocketHandler.class);

    /** 连接建立后回调 */
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        // session.getId() 是这个连接的唯一标识
        // getAttributes() 里能拿到握手拦截器塞进来的东西
        Object userId = session.getAttributes().get("userId");
        log.info("WebSocket 已连接, sessionId={}, userId={}", session.getId(), userId);
    }

    /** 收到文本消息时回调 */
    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        String payload = message.getPayload();
        // 注意：这里不建议做重业务逻辑，它会占用 WebSocket 容器的 IO 线程
        session.sendMessage(new TextMessage("echo: " + payload));
    }

    /** 传输层出错时回调 */
    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
        log.warn("WebSocket 传输异常, sessionId={}", session.getId(), exception);
        if (session.isOpen()) {
            session.close(CloseStatus.SERVER_ERROR);
        }
    }

    /** 连接关闭后回调 */
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        log.info("WebSocket 已断开, sessionId={}, code={}, reason={}",
                session.getId(), status.getCode(), status.getReason());
    }
}
  • handleTextMessage 跑在 WebSocket 容器的 IO 线程上，不要在里面做耗时操作。 如果需要查库、调远程服务，正确做法是丢到业务线程池里异步处理，处理完再回到 session 上发送。这与 Servlet 里”不要在 Filter 里做重活”是同一个道理。

注册 handler 与路由

通过 @EnableWebSocket 加 WebSocketConfigurer 把 handler 绑到 URL 上：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {

    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(notifyHandler(), "/ws/notify")
                // 生产环境不要用 setAllowedOrigins("*")，下面会解释
                .setAllowedOriginPatterns("https://app.example.com")
                .addInterceptors(new AuthHandshakeInterceptor());
    }

    @Bean
    public WebSocketHandler notifyHandler() {
        return new NotifyWebSocketHandler();
    }
}

另外，这个示例是给 Spring MVC 应用用的，应该放在 DispatcherServlet 的配置里。

但 Spring WebSocket 本身并不依赖 Spring MVC，如果你想把它集成到别的 HTTP 处理环境里，可以用 WebSocketHttpRequestHandler。

关于 setAllowedOrigins 和 setAllowedOriginPatterns，两个方法的定位是：

  • setAllowedOrigins(String...)：设置允许跨域请求的源。默认不允许任何源。而且 Javadoc 特别提醒：CORS 规范不允许在 allowCredentials=true 的同时使用 "*"；
  • setAllowedOriginPatterns(String...)：setAllowedOrigins 的替代方案，支持更灵活的模式。当两者都设置时，后者优先。

从 Spring 4.1.5 起，WebSocket 和 SockJS 的默认行为就已经是 只接受同源请求 了。Spring 官方列出了三种可选行为：

  • 只允许同源（默认）：这是最安全的默认值。这种模式下，如果启用了 SockJS，Iframe 传输的 HTTP 响应头 X-Frame-Options 会被设成 SAMEORIGIN，并且 JSONP 传输会被禁用（因为它没法校验请求来源）。代价是 IE6 和 IE7 不受支持。
  • 允许指定的来源列表：每一个来源都必须以 http:// 或 https:// 开头。这种模式下，如果启用了 SockJS，IFrame 传输会被禁用，代价是 IE6 到 IE9 不受支持。
  • 允许所有来源：传入 *。这种模式下所有传输都可用，但安全性完全交给你自己了。

这个检查主要是给浏览器客户端设计的，因为除了浏览器，没有任何东西能阻止其他类型的客户端修改 Origin 头的值。Origin 校验只对浏览器客户端有意义。 非浏览器客户端可以随意伪造 Origin 头。所以别把它当成认证手段，它只是一道防 CSRF 的防线，毫无身份认证。

握手拦截器：把用户信息带进会话

原生 WebSocket 最常见的一个需求是：在握手时把当前登录用户识别出来，并存到会话里，因为后续收到消息时你拿不到 HTTP 请求上下文了。

Spring 提供了 HandshakeInterceptor。官方对它的介绍是：定制开场 HTTP WebSocket 握手请求最简单的方式，就是用一个 HandshakeInterceptor，它暴露了握手前后两个方法；你可以用这样的拦截器来阻止握手，或者把任意属性提供给 WebSocketSession。

它有两个方法：

  • beforeHandshake：握手前
  • afterHandshake：握手后

下面是我在实际项目里用的写法，它做三件事：解析 token、写入会话属性、拒绝非法握手。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
@Component
public class AuthHandshakeInterceptor implements HandshakeInterceptor {

    private final TokenService tokenService;

    public AuthHandshakeInterceptor(TokenService tokenService) {
        this.tokenService = tokenService;
    }

    @Override
    public boolean beforeHandshake(ServerHttpRequest request,
                                   ServerHttpResponse response,
                                   WebSocketHandler wsHandler,
                                   Map<String, Object> attributes) {
        // 方式一：从查询参数取 token
        // 注意：浏览器原生的 WebSocket API 无法自定义请求头，所以 token 通常走 query
        String token = null;
        if (request instanceof ServletServerHttpRequest servletRequest) {
            token = servletRequest.getServletRequest().getParameter("token");
        }

        // 方式二：如果项目用 Cookie + Session，可以直接从 Cookie 里读会话
        // principal 也可以从 request.getPrincipal() 拿到（由安全过滤器写入）

        Long userId = tokenService.parseUserId(token);
        if (userId == null) {
            // 返回 false 直接拒绝握手，客户端会收到握手失败
            response.setStatusCode(HttpStatus.UNAUTHORIZED);
            return false;
        }

        // 这些属性会跟随整个 WebSocket 会话，后续通过 session.getAttributes() 取
        attributes.put("userId", userId);
        attributes.put("handshakeTime", System.currentTimeMillis());
        return true;
    }

    @Override
    public void afterHandshake(ServerHttpRequest request,
                               ServerHttpResponse response,
                               WebSocketHandler wsHandler,
                               @Nullable Exception exception) {
        // 一般不需要做什么
    }
}

怎么确定这个连接是谁，Spring 还提供了 AbstractHandshakeHandler#determineUser 这个方法，它的默认实现就是取 ServerHttpRequest#getPrincipal()；如果你的项目需要匿名用户也分配一个名字，经常会覆盖它。另外 DefaultHandshakeHandler 负责校验 Origin、协商子协议这些细节，需要更高级的定制时可以从它继承。

对于身份验证，很搞的是，浏览器原生的 WebSocket 构造函数不支持自定义请求头。

1
2
// 这是唯一的形式，没法加 header
const ws = new WebSocket('wss://api.example.com/ws/notify?token=' + token);

所以 token 只能通过两种方式传：

  1. URL 查询参数（最常见，但要注意 URL 会被记录到访问日志里，敏感 token 会泄漏到日志）；
  2. Cookie（如果 token 存在 Cookie 里，浏览器会自动带上，但跨站场景受 SameSite 限制）。

如果确实需要自定义 header，比如标准的 Authorization: Bearer xxx，有两条路：

  • 转用 SockJS 或者 STOMP over WebSocket，因为 STOMP 的 CONNECT 帧本身可以带自定义头，Spring 也支持从 STOMP 头里取 token；
  • 用非浏览器客户端，它们大多支持自定义头。

这个原生 WebSocket 在鉴权上的一个痛点，也是很多人转向 STOMP 的原因之一。

并发发送问题

官方说：当直接使用 WebSocketHandler API，而不是间接地通过 STOMP 消息机制使用它的时候，应用必须自行同步消息的发送，因为底层标准的 WebSocket 会话（JSR-356）不允许并发发送；一个可选做法是用 ConcurrentWebSocketSessionDecorator 把 WebSocketSession 包装起来。

这个限制带来的现象非常隐蔽：如果你有两个线程同时往同一个 session 发消息，可能会抛 IllegalStateException: The remote endpoint was in state [TEXT_FULL_WRITING]，也可能表现为消息内容交错损坏。而且它不是每次都必现，只在并发碰撞时出现，测试环境很难复现，上了生产才爆。

Spring 提供了 ConcurrentWebSocketSessionDecorator 来解决。它的作用是：包装一个 WebSocketSession，以保证同一时刻只有一个线程能发送消息；如果某次发送很慢，其他线程后续尝试发送更多消息时就无法获取刷新锁，这些消息会被缓冲起来；在这种情况下，配置的缓冲区大小上限和发送时间上限会被检查，一旦超限，会话会被关闭。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public class NotifyWebSocketHandler extends TextWebSocketHandler {

    /** 单个会话的发送超时时间（毫秒） */
    private static final int SEND_TIME_LIMIT = 5_000;

    /** 单个会话允许积压的发送缓冲区大小（字节） */
    private static final int SEND_BUFFER_SIZE_LIMIT = 512 * 1024;

    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        // 注意：这里只是演示装饰器的用法，实际需要把装饰后的 session 存起来复用
        WebSocketSession concurrent = new ConcurrentWebSocketSessionDecorator(
                session, SEND_TIME_LIMIT, SEND_BUFFER_SIZE_LIMIT);
        // 后续所有发送都走 concurrent
    }
}

这两个参数的含义很关键：

  • sendTimeLimit：一次发送允许占用的最长时间。超时后 Spring 会关闭这个会话；
  • sendBufferSizeLimit：该会话允许积压的缓冲区上限。超了也会关闭会话。

另外这个装饰器还有一个 OverflowStrategy 参数可以选择溢出的处理方式，默认是 TERMINATE，也就是超限时抛异常并终止会话；另一个可选值是 DROP，表示丢弃最旧的消息而不是断开连接。对于”消息可以丢、但连接不能断”的场景，DROP 更合适。

为什么需要缓冲区上限？因为如果客户端网络很慢（或者干脆是个恶意客户端，故意不读数据），服务端往这个 session 发的消息会不断堆积在内存里，最终把 JVM 内存撑爆。这是一个典型的慢消费者攻击面。给每个会话设一个缓冲区上限，超过就断开，是标准的防护手法。

STOMP 路线下不需要手动做这件事，因为 Spring 的消息基础设施已经内置了 sendTimeLimit 和 sendBufferSizeLimit 配置。

服务端容器配置：缓冲区、超时、装饰器

原生的缓冲区大小默认值往往不够用，尤其是消息体稍大一点时。Spring 提供了 ServletServerContainerFactoryBean 来配置 Jakarta WebSocket 容器：

1
2
3
4
5
6
7
8
9
10
11
12
13
@Bean
public ServletServerContainerFactoryBean createWebSocketContainer() {
    ServletServerContainerFactoryBean container = new ServletServerContainerFactoryBean();
    // 文本消息缓冲区，默认值参考容器默认（Tomcat 为 8192 字节）
    container.setMaxTextMessageBufferSize(64 * 1024);
    // 二进制消息缓冲区
    container.setMaxBinaryMessageBufferSize(64 * 1024);
    // 会话空闲超时
    container.setMaxSessionIdleTimeout(30 * 60 * 1000L);
    // 异步发送超时
    container.setAsyncSendTimeout(10_000L);
    return container;
}

这几个参数在 Tomcat 11 上的含义如下

配置项 Tomcat 11 默认值 Jetty 12.1 默认值
文本消息缓冲区上限 8192 字节 65536 字节
二进制消息缓冲区上限 8192 字节 65536 字节
单帧最大载荷 由消息缓冲区约束 65536 字节
阻塞发送写超时 20000 毫秒 通过 blockingSendTimeout 配置
关闭握手等待超时 30000 毫秒 —
空闲超时 见下方说明 由 setIdleTimeout 控制

不同容器的默认值相差 8 倍。所以说，永远显式设置这些值是有必要的，不要依赖任何容器的默认值。

另外，Spring 的 Javadoc 对 ServletServerContainerFactoryBean 这四个 setter 也没有给出默认值，因为它们的默认行为取决于具体的容器实现。

关于装饰器，官方还提到 Spring 默认会加上两个 WebSocketHandlerDecorator。官方的说法是，在使用 WebSocket 的 Java 配置或 XML 命名空间时，日志和异常处理这两个实现是默认提供的；其中 ExceptionWebSocketHandlerDecorator 会捕获所有从 WebSocketHandler 方法里逃逸出来的未捕获异常，并以 1011 状态码关闭 WebSocket 会话，表示服务端错误。

也就是：

  • ExceptionWebSocketHandlerDecorator：捕获所有未处理异常，用状态码 1011 关闭会话；
  • LoggingWebSocketHandlerDecorator：记录连接与消息日志。

顺便说明一下它的注册方式：这些装饰器是通过 WebSocketTransportRegistration#addDecoratorFactory 或 setDecoratorFactories 挂上去的。默认的 decoratorFactories 初始是空列表，所以官方文档说的”默认添加”指的是 Spring 的配置类会替你注册，而不是框架内部有一个隐式的默认列表，理解这一点很重要，因为你自己设置 setDecoratorFactories 的时候是有可能把它覆盖掉的。

如果你需要自定义行为，比如统一埋点、统一限流，可以继承 WebSocketHandlerDecorator 自己加一层：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
public class MetricsWebSocketHandlerDecorator extends WebSocketHandlerDecorator {

    public MetricsWebSocketHandlerDecorator(WebSocketHandler delegate) {
        super(delegate);
    }

    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        Metrics.counter("ws.connections.total").increment();
        super.afterConnectionEstablished(session);
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) throws Exception {
        Metrics.counter("ws.connections.closed").increment();
        // 按关闭码分类打点，这是排查断连原因最有效的手段
        Metrics.counter("ws.close.code." + closeStatus.getCode()).increment();
        super.afterConnectionClosed(session, closeStatus);
    }
}

按关闭码分类打点是排查 WebSocket 断连问题的第一手段。1006 占比高说明网络或代理在掐连接；1011 占比高说明业务代码在抛异常；1000 占比高则说明是正常的客户端关闭。

原生 WebSocket 的局限：为什么企业项目需要 STOMP

写到这里，原生 WebSocket 的问题已经很清楚了。Spring 官方文档用了很长一段来解释 WebSocket 作为低层传输协议的定位，大意是：WebSocket 是一种低层传输协议，和 HTTP 不同，它不对消息内容规定任何语义。这意味着，除非客户端和服务端就消息语义达成一致，否则没有任何办法对消息进行路由或处理。

所以说，WebSocket 不规定消息内容的语义。 它只负责把字节从一端搬到另一端，至于这个字节是什么，协议一无所知，也无法路由。

对比 HTTP ，在 HTTP 和 REST 里，应用被建模成很多个 URL；要与应用交互，客户端就以请求响应的方式访问这些 URL；服务端根据 HTTP 的 URL、方法和头部把请求路由到合适的处理器。而在 WebSocket 里，通常只有一个 URL 用于初始连接，此后所有应用消息都在同一条 TCP 连接上流动，这指向了一种完全不同的、异步的、事件驱动的消息架构。

这段话点出了 WebSocket 架构上的本质变化：

HTTP / REST WebSocket
应用建模 很多个 URL 通常只有一个连接 URL
路由依据 URL + method + header 无（握手后所有消息走同一条连接）
交互方式 请求响应 异步事件驱动
状态 无状态（或 Session 承载） 有状态的连接会话

所以用原生 WebSocket 做业务时，你实际上要自己发明一套应用层协议：

  • 消息里要带一个指令类型字段来做路由（等价于 URL）；
  • 要自己实现”订阅”和”取消订阅”（等价于 broker 的订阅表）；
  • 要自己维护 哪个 session 订阅了哪个主题 的映射；
  • 要自己实现给某个用户的所有会话推送；
  • 要自己处理序列化、错误码、超时、心跳。

这就是 STOMP 存在的意义。Spring 官方文档列出了用 STOMP 作为子协议，好处如下：

  1. 不用自己发明一套消息协议和消息格式；
  2. 有现成的 STOMP 客户端可用（包括 Spring Framework 提供的 Java 客户端）；
  3. 可以选择使用消息代理（比如 RabbitMQ、ActiveMQ 等）来管理订阅和广播消息；
  4. 业务逻辑可以组织到任意多个 @Controller 里，按 STOMP 的 destination 头来路由消息，而不是像处理裸 WebSocket 消息那样，一条连接只能对应一个 WebSocketHandler。
  5. 可以用 Spring Security 基于 STOMP 的目的地和消息类型来做授权。

STOMP 协议速览

STOMP 是什么。Spring 文档对它的定位是：

  • 全称面向文本的简单消息协议，最初是为 Ruby、Python、Perl 这类脚本语言连接企业消息代理而设计的；
  • 它只覆盖常用消息模式的一个最小子集，不是一个完备的消息 API；
  • 可以跑在任何可靠的双向流式协议上，TCP 和 WebSocket 都行，这一点很重要，它意味着 STOMP 和 WebSocket 是解耦的；
  • 虽然是文本协议，但载荷可以是文本或二进制。

帧结构

STOMP 的帧结构受 HTTP 启发，STOMP 1.2 规范 给出的形式是：

1
2
3
4
5
COMMAND
header1:value1
header2:value2

Body^@

规范对每一部分的定义，我整理成中文如下：

  • 命令：帧的第一行，以行结束符收尾。这个行结束符是”可选的 CR + 必需的 LF”，也就是 \n 或 \r\n 都可以；
  • 头部：命令之后是零个或多个 <key>:<value> 形式的头部条目，每个条目也以行结束符收尾；
  • 空行：一个额外的行结束符，标记头部结束、Body 开始；
  • Body：消息内容；
  • NULL 字节（^@ 表示，ASCII 的 0x00）：帧结束符。这个 NULL 字节之后还可以跟多个行结束符。

拆解成五步：

  1. 命令：第一行，以 EOL 结束；
  2. 头部：零到多个 key:value，每行一个；
  3. 空行：一个额外的 EOL，标记头部结束、Body 开始；
  4. Body：消息内容；
  5. NULL 字节：帧结束符。

那个 NULL 字节是 STOMP 解析的关键。如果 Body 里含有 NULL 字节，必须带 content-length 头，否则接收方无法确定帧边界，它会把 Body 里的 NULL 当成结束符。这一条在传二进制内容时特别重要。

还有几条解析规则值得知道：

  • 只有 SEND、MESSAGE 和 ERROR 这三种帧可以带 Body，其他帧都不得带 Body；
  • 命令和头部的名字是区分大小写的；
  • 除了 CONNECT 和 CONNECTED 帧以外，其他帧里的头部如果含有 CR、LF 或冒号，必须做转义（用 C 风格的 \r、\n、\c、\\）；而未定义的转义序列（比如 \t）必须被视为致命的协议错误；
  • 在 STOMP 1.2 里，客户端和服务端都绝不允许对头部值做空格裁剪或补齐。

常用命令

命令 方向 作用
CONNECT / STOMP 客户端到服务端 建立会话，可带 accept-version、host、heart-beat
CONNECTED 服务端到客户端 会话建立成功，返回协商出的 version
SEND 客户端到服务端 向某个 destination 发消息
SUBSCRIBE 客户端到服务端 订阅某个 destination，必须带唯一 id
UNSUBSCRIBE 客户端到服务端 取消订阅，需要 id
MESSAGE 服务端到客户端 推送消息，带 destination、message-id、subscription
ACK / NACK 客户端到服务端 确认/否认消费（需要 ack 模式支持）
BEGIN / COMMIT / ABORT 客户端到服务端 事务
DISCONNECT 客户端到服务端 优雅断开，可带 receipt
RECEIPT 服务端到客户端 对带 receipt 的帧的处理确认
ERROR 服务端到客户端 出错，服务端发完 ERROR 后必须关闭连接

服务端不能发送”未经请求”的消息。 所有服务端消息都必须是对某个客户端订阅的响应，subscription 头必须匹配客户端订阅的 id。这条规则意味着：客户端必须先订阅，服务端才能推。实践中前端要先发 SUBSCRIBE，后面才能收到 MESSAGE。

关于 CONNECT 帧，规范规定了两项必需的头：

  • accept-version（客户端支持的 STOMP 版本列表）
  • host（客户端想连接的虚拟主机名）；
  • 可选的头包括 login、passcode 和 heart-beat。

destination 的语义

关于 destination，destination 的含义在 STOMP 规范里是被有意留成不透明的。它可以是任意字符串，其语义和语法完全由 STOMP 服务端自己定义。不过很常见的一种做法是把它写成类似路径的字符串，其中 /topic/.. 表示发布订阅（一对多），/queue/ 表示点对点（一对一）的消息交换。

这一点在 Spring 的 simple broker 上要注意，对于内置的简单 broker，/topic 和 /queue 这两个前缀没有任何特殊含义，它们只是为了区分”发布订阅”和”点对点”而约定的命名习惯。当你使用外部 broker 时，要去查那个 broker 的 STOMP 文档，了解它支持什么样的目的地和前缀。

心跳机制

STOMP 有协议级的心跳，它用一个 heart-beat 头同时表达”我能发多快”和”我希望你发多快”。具有如下规则

  • 这个头必须包含两个以逗号分隔的正整数；就是 heart-beat:<cx>,<cy>：cx是发送方能保证的最小发送间隔（0 表示不能发）；cy是发送方希望收到的间隔（0 表示不想收）。
  • 第一个数字表示”发送方自己能做到什么”（也就是它外发心跳的能力）：0 表示它不能发送心跳，否则表示它能保证的两次心跳之间的最小毫秒间隔
  • 第二个数字表示”发送方希望收到什么”（也就是它希望对方发来的心跳频率）：0 表示它不想收到心跳，否则表示它期望的两次心跳之间的毫秒间隔。

对于客户端到服务端方向的心跳，如果客户端的 cx 是 0（客户端不能发心跳），或者服务端的 sy 是 0（服务端不想收心跳），那就不会有心跳；否则心跳间隔就是 cx 和 sy 里的较大值。

也就是取双方声明的较大值：客户端说”我最快 10000ms 能发一次”，服务端说”我希望 20000ms 收一次”，那实际就是 20000ms 一次。很明显，不这样取值会有问题

心跳的发送方式也很有意思，在某个方向上，如果期望每 n 毫秒有一次心跳，那么发送方必须至少每 n 毫秒在连接上发送一次新数据；如果发送方没有真实的 STOMP 帧要发，它必须发送一个行结束符（EOL）；而如果接收方在一个至少 n 毫秒的时间窗口内没有收到任何新数据，它可以把这条连接视为已死。

如果没有真实的 STOMP 帧要发，就发一个 EOL（换行符）来充当心跳。也就是说，STOMP 的心跳就是一个空行。这样做的好处是接收方不需要区分心跳帧和数据帧，不增加解析复杂度，而且其实很明显，任何在连接上收到的新数据，都表示远端还活着。

最后一条实践提醒：缺少 heart-beat 头必须被当作 heart-beat:0,0 处理，也就是”不能发、也不想收”。

在 Spring Boot 4 上实现 STOMP

服务端配置

核心是 @EnableWebSocketMessageBroker 加上实现 WebSocketMessageBrokerConfigurer：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
@Configuration
@EnableWebSocketMessageBroker
public class StompConfig implements WebSocketMessageBrokerConfigurer {

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        // 1. 启用内置的 simple broker，并声明它负责哪些目的地前缀
        //    注意：simple broker 下 /topic 与 /queue 没有任何特殊语义，只是命名约定
        registry.enableSimpleBroker("/topic", "/queue")
                // STOMP 心跳需要一个 TaskScheduler（见下方说明）
                .setHeartbeatValue(new long[]{10_000, 10_000})
                .setTaskScheduler(heartbeatScheduler());

        // 2. 声明应用目的地前缀：以 /app 开头的消息会被路由到 @MessageMapping 方法
        registry.setApplicationDestinationPrefixes("/app");

        // 3. 用户目的地前缀，默认就是 /user，这里显式声明以便阅读
        //    注意：/user 不能落进上面 broker 的前缀范围，否则会被 broker 抢走
        registry.setUserDestinationPrefix("/user");

        // 4. 开启发布顺序保证（有性能开销，按需开启）
        registry.setPreservePublishOrder(true);
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        // 接收顺序保证配置在"端点"上，不是在 MessageBrokerRegistry 上（6.1 起支持）
        registry.setPreserveReceiveOrder(true);

        registry.addEndpoint("/ws/stomp")
                // 生产环境应配置具体的来源，而不是用 *
                .setAllowedOriginPatterns("https://app.example.com")
                // 握手拦截器（鉴权、埋点）
                .addInterceptors(new AuthHandshakeInterceptor(tokenService));

        // 如果需要兼容老浏览器，再额外注册一个带 SockJS 的端点
        // registry.addEndpoint("/ws/stomp-sockjs")
        //         .setAllowedOriginPatterns("https://app.example.com")
        //         .withSockJS();
    }

    @Bean
    public TaskScheduler heartbeatScheduler() {
        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
        scheduler.setPoolSize(1);
        scheduler.setThreadNamePrefix("ws-heartbeat-");
        return scheduler;
    }
}

这里有几个配置项值得展开说明。

enableSimpleBroker 与心跳。 Spring 文档说：简单 broker 只有在配置了任务调度器的情况下才支持 STOMP 心跳；要配置调度器，你可以声明自己的 TaskScheduler bean，并通过 MessageBrokerRegistry 设置进去。如果没有配 TaskScheduler，simple broker 的心跳是不工作的。

另外如果用 Spring 内置 WebSocket 配置里自动声明的那个 TaskScheduler，需要加 @Lazy 来打破循环依赖。一般是自己声明一个专用的 TaskScheduler来避免这个问题，也方便给它起个可识别的线程名前缀。

关于心跳的值，Javadoc 里的语义需要说清楚：setHeartbeatValue(long[] heartbeat) 的第一个元素是服务端发送心跳的间隔，第二个元素是期望客户端发送心跳的间隔，0 表示不做心跳；默认是 [0, 0]（也就是不心跳），而一旦设置了 TaskScheduler，默认值就变成 [10000, 10000]（毫秒）。

关于 SockJS 的一个必要澄清。他就像是棍父一样天天被说弃用了结果你点开荷兰方片军官还是看得到，从来没有弃用过，但是很多人都在说，所以 SockJS 在今天依然是一个可选项

关于两种心跳会不会打架，不会，STOMP 层面的心跳协商成功之后，SockJS 自身的心跳会被禁用。

最后是 SockJS 的兼容性代价：启用 SockJS 并与 Spring Security 配合时，需要放宽两项 Web 层配置，也就是帧选项和 CSRF 豁免，这个我在讲安全时会详细写。如果你的用户群已经不需要支持 IE，那么 SockJS 的价值会大幅下降，跟弃用了差不多，此时原生 WebSocket 更简单、开销更低。

三条消息通道与消息流

理解 STOMP 的配置，必须先理解消息是怎么流的。Spring 明确列出了三条通道：

  • clientInboundChannel：用于传递从 WebSocket 客户端收到的消息；
  • clientOutboundChannel：用于把服务端的消息发送给 WebSocket 客户端；
  • brokerChannel：用于在服务端应用代码内部向消息 broker 发送消息。
通道 方向 用途
clientInboundChannel 客户端 → 服务端 接收 WebSocket 客户端发来的消息
brokerChannel 应用内部 → broker 服务端代码主动往 broker 发消息
clientOutboundChannel 服务端 → 客户端 把服务端消息写回 WebSocket 客户端

官方文档还给了一个完整的流程示例：

  1. 客户端连接到 http://localhost:8080/portfolio，一旦 WebSocket 连接建立，STOMP 帧就开始在上面流动；
  2. 客户端发一个 SUBSCRIBE 帧，destination 头是 /topic/greeting。收到并解码后，消息被送到 clientInboundChannel，然后被路由到消息 broker，由 broker 保存这个客户端订阅；
  3. 客户端发一个 SEND 帧到 /app/greeting。/app 这个前缀的作用是把它路由到注解控制器；前缀被剥掉之后，剩下的 /greeting 部分就映射到 GreetingController 里的 @MessageMapping 方法；
  4. GreetingController 的返回值被转成一个 Spring Message，载荷来自返回值，destination 头默认是 /topic/greeting（由入站 destination 把 /app 换成 /topic 得来）。这条消息被送到 brokerChannel，由消息 broker 处理；
  5. 消息 broker 找到所有匹配的订阅者，通过 clientOutboundChannel 给每一个发送一个 MESSAGE 帧，最后由这个通道把消息编码成 STOMP 帧，发到 WebSocket 连接上。

注意第 4 步里的那个自动转换规则，非常关键：@MessageMapping 方法的返回值，默认目的地是把入站目的地的 /app 换成 /topic。

1
2
入站 destination: /app/greeting
出站 destination: /topic/greeting

这个隐式规则让代码很简洁，但也容易让人困惑”我的消息到底发到哪去了”。如果目的地不符合预期，就用 @SendTo 显式指定。

flowchart LR
    A[浏览器客户端] -->|STOMP 帧| B[WebSocket 会话]
    B --> C[clientInboundChannel]
    C -->|/app 前缀| D[MessageMapping 控制器]
    C -->|/topic /queue| E[Message Broker]
    D -->|返回值| F[brokerChannel]
    F --> E
    E --> G[clientOutboundChannel]
    G --> B
    B -->|MESSAGE 帧| A
    H[业务服务/定时任务] -->|SimpMessagingTemplate| F

编写消息控制器

STOMP 的控制器写起来和 Spring MVC 非常像，这是它最大的易用性优势。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@Controller
public class OrderNotifyController {

    private final SimpMessagingTemplate messagingTemplate;
    private final OrderService orderService;

    public OrderNotifyController(SimpMessagingTemplate messagingTemplate, OrderService orderService) {
        this.messagingTemplate = messagingTemplate;
        this.orderService = orderService;
    }

    /**
     * 处理客户端发到 /app/order/subscribe 的消息。
     * 注意：/app 前缀会被剥掉，所以映射写 /order/subscribe
     */
    @MessageMapping("/order/subscribe")
    public void subscribeOrder(@Payload OrderSubscribeCmd cmd, Principal principal) {
        // principal 是握手时建立的用户身份
        // 实际项目中这里应该做权限校验：这个用户能不能看这个订单
        String destination = "/topic/order/" + cmd.orderId();
        // 主动推一条当前状态，让客户端立刻有数据可渲染
        messagingTemplate.convertAndSend(destination, orderService.currentStatus(cmd.orderId()));
    }

    /**
     * 带模板变量的 destination，用 @DestinationVariable 取
     */
    @MessageMapping("/stock/{symbol}/watch")
    public void watchStock(@DestinationVariable String symbol, Principal principal) {
        // ...
    }

    /**
     * 用 @SubscribeMapping 处理"只订阅一次拿初始数据"的场景。
     * 返回值直接回给订阅者，不经过 broker，不保留订阅。
     */
    @SubscribeMapping("/dashboard/init")
    public DashboardVO initDashboard(Principal principal) {
        return dashboardService.loadFor(principal.getName());
    }

    /**
     * 用 @SendTo 显式指定广播目的地
     */
    @MessageMapping("/chat/send")
    @SendTo("/topic/chat/room1")
    public ChatMessage broadcast(ChatMessage message, Principal principal) {
        return message.withSender(principal.getName());
    }

    /**
     * 用 @SendToUser 只发给消息发送者自己
     */
    @MessageMapping("/trade")
    @SendToUser("/queue/position-updates")
    public TradeResult executeTrade(Trade trade, Principal principal) {
        return tradeService.execute(trade, principal.getName());
    }

    /** 处理某个控制器内的异常 */
    @MessageExceptionHandler
    @SendToUser(destinations = "/queue/errors", broadcast = false)
    public ApplicationError handleException(MyBusinessException ex) {
        return new ApplicationError(ex.getMessage());
    }
}

关于方法参数，官方文档给出了完整的支持列表，我整理成表：

方法参数 说明
Message 访问完整消息
MessageHeaders 访问消息头
MessageHeaderAccessor / SimpMessageHeaderAccessor / StompHeaderAccessor 通过类型化访问器访问消息头
@Payload 访问载荷，由配置的 MessageConverter 转换（比如从 JSON）
@Header 访问特定消息头的值，必要时用 Converter 做类型转换
@Headers 访问全部消息头，参数必须能赋给 java.util.Map
@DestinationVariable 访问从 destination 模板变量中提取的值，必要时做类型转换
java.security.Principal 反映 WebSocket HTTP 握手时登录的用户

这个注解@Payload并不是必须的，因为如果没有其他参数匹配上，它就默认被当作载荷处理。所以 public void handle(ChatMessage msg) 和 public void handle(@Payload ChatMessage msg) 是等价的。这减少了大量样板代码。而且载荷参数支持 @Valid 自动校验，和 Spring MVC 完全一致。

关于返回值的默认行为，官方也说明：@MessageMapping 方法的返回值默认会经由匹配的 MessageConverter 序列化成载荷，并作为 Message 发往 brokerChannel，再由 broker 广播给订阅者；这条出站消息的目的地，就是入站消息的目的地加上 /topic 前缀。

另外，@SendTo 和 @SendToUser 可以同时用在同一个方法上，而且都支持类级别，在类级别时作为该类内方法的默认值，但方法级别的注解会覆盖类级别的。@MessageMapping 方法还可以返回 ListenableFuture、CompletableFuture 或 CompletionStage，也就是支持异步处理。

关于 @SubscribeMapping 的取舍，它的行为差异是：

  • @MessageMapping 的返回值：发到 broker，由 broker 广播给所有匹配的订阅者；
  • @SubscribeMapping 的返回值：默认直接通过 clientOutboundChannel 发给发起订阅的那个客户端，不经过 broker，也不会把这个订阅存下来给后续广播用。

不要试图把 broker 和控制器映射到同一个目的地前缀，除非你确实想让两者独立处理消息（包括订阅）。因为入站消息是并行处理的，一条给定消息到底由 broker 还是控制器先处理，是没有任何保证的。

关于怎么知道订阅已经就绪？如果你希望在订阅被 broker 保存、可以开始接收广播时收到通知，客户端应该请求 receipt（如果服务端支持的话，simple broker 并不支持）。 换句话讲，simple broker 下你是没有订阅确认这个能力的；要么用带 receipt 能力的 Java STOMP 客户端，要么在服务端注册一个 ExecutorChannelInterceptor 到 brokerChannel，实现 afterMessageHandled，在消息（包括订阅）被处理完之后做后续动作。

从任何地方推送消息

@MessageMapping 只能响应客户端的消息。但业务上更常见的是业务事件发生后主动推送，比如订单支付成功、库存变化、定时任务产出的报表。这时用 SimpMessagingTemplate。

如果你想把消息从应用的任何部分发给已连接的客户端呢？任何应用组件都可以把消息发到 brokerChannel，而最简单的做法就是注入一个 SimpMessagingTemplate 并用它来发消息。

任何组件都可以往 brokerChannel 发消息，最简单的方式就是注入 SimpMessagingTemplate：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
@Service
public class OrderEventPublisher {

    private final SimpMessagingTemplate messagingTemplate;

    public OrderEventPublisher(SimpMessagingTemplate messagingTemplate) {
        this.messagingTemplate = messagingTemplate;
    }

    /** 广播给所有订阅该订单频道的客户端 */
    public void publishOrderPaid(OrderPaidEvent event) {
        messagingTemplate.convertAndSend(
                "/topic/order/" + event.orderId(),
                new OrderStatusVO(event.orderId(), "PAID", event.paidAt()));
    }

    /** 只推给指定用户的所有会话 */
    public void notifyUser(String username, NotificationVO notification) {
        messagingTemplate.convertAndSendToUser(
                username,
                "/queue/notifications",
                notification);
    }
}

注意 convertAndSendToUser 的第二个参数是用户私有的目的地，客户端订阅时要加上 /user 前缀：

1
2
3
stompClient.subscribe('/user/queue/notifications', (frame) => {
  showNotification(JSON.parse(frame.body));
});

关于用户目的地的转换机制，应用可以发送针对特定用户的消息，Spring 的 STOMP 支持识别以 /user/ 为前缀的目的地来实现这一点。比如客户端可能订阅 /user/queue/position-updates 这个目的地；UserDestinationMessageHandler 会处理它，并把它转换成一个对该用户会话唯一的 destination（例如 /queue/position-updates-user123）。这样做的好处是，你可以订阅一个通用命名的目的地，同时又保证不会和其他订阅了同一目的地的用户发生冲突，让每个用户都能收到属于自己的那份更新。

/user/queue/position-updates 会被 UserDestinationMessageHandler 转换成 /queue/position-updates-user123 这种带用户标识的唯一目的地。 这样多个用户订阅同一个逻辑目的地时不会互相串消息。

有一个方向上的区别：订阅侧写的是 /user/...（不带用户名），而发送侧写的是 /user/{username}/...（带用户名），后者会被解析成该用户的一个或多个会话目的地。

必须正确配置 broker 前缀和应用前缀，否则 broker 会去处理本该由 UserDestinationMessageHandler 处理的 /user 前缀消息。 这就是为什么在配置里要把 enableSimpleBroker("/topic", "/queue") 和 setApplicationDestinationPrefixes("/app") 配清楚——不要让 /user 落进 broker 的前缀范围。

关于多会话的行为，默认情况下，如果同一个用户有多个会话（多标签页、多设备），消息会发给所有会话。 如果只想发给”当前这个发消息的会话”，用 @SendToUser(broadcast = false)。

还有一条提醒是关于未登录用户的，未认证的用户也能订阅 user destination，此时 @SendToUser 的行为等同于 broadcast=false。这说明 user destination 本身不是一道安全边界，别指望用它来做隔离——真正的隔离要靠 Spring Security 的授权规则。

线程池与性能调优

STOMP 的性能调优，核心是配置背撑三条通道的线程池，以及几个与消息发送相关的保护参数。

关于 clientInboundChannel 和 clientOutboundChannel 的默认线程数，存在三种说法：

来源 说法
Spring 参考文档 可用处理器数量的 2 倍
WebSocketMessageBrokerConfigurer Javadoc 大小为 1 的线程池
Spring Framework 7.0.9 源码 corePoolSize = 0、maxPoolSize = 1

很明显三者冲突了。源码和 Javadoc 一致，参考文档说法不同。但纠结默认值没有意义，无论默认是 1 还是 2×CPU，对生产级消息推送服务都远远不够，必须显式配置。这里一般是把默认值当成极小/未定义，不依赖它，直接设置成业务需要的值

调优方向取决于工作负载类型，如果注解方法里的处理主要是 CPU 密集型的，clientInboundChannel 的线程数应当保持在处理器数量附近；如果这些工作更偏 IO 密集、需要阻塞或等待数据库或其他外部系统，那么线程池大小大概就需要增加。

  • CPU 密集：clientInboundChannel 线程数保持在 CPU 核心数附近；
  • IO 密集（查库、调远程服务）：要适当增加线程数。

而且注意到，如果队列容量是默认的 Integer.MAX_VALUE，那么线程数永远不会超过核心线程数，因为所有超出核心数的任务都被塞进队列里排队了，永远触发不了”队列满了才扩容到 maxPoolSize”的条件。

所以说，正确的配置方式是同时设置队列容量，让扩容条件能被触发：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
@Configuration
@EnableWebSocketMessageBroker
public class StompThreadPoolConfig implements WebSocketMessageBrokerConfigurer {

    @Override
    public void configureClientInboundChannel(ChannelRegistration registration) {
        registration.taskExecutor()
                .corePoolSize(16)
                // 关键：队列容量必须有限，否则 maxPoolSize 形同虚设
                .maxPoolSize(64)
                .queueCapacity(500)
                .keepAliveSeconds(60);
        // 也可以注册拦截器做鉴权/埋点
        registration.interceptors(new MetricsChannelInterceptor());
    }

    @Override
    public void configureClientOutboundChannel(ChannelRegistration registration) {
        // 出站线程数取决于客户端网络状况：客户端越慢，越需要更多线程
        registration.taskExecutor()
                .corePoolSize(16)
                .maxPoolSize(64)
                .queueCapacity(1000);
    }
}

出站线程池的调优逻辑和入站不同，出站线程池的负载取决于客户端网络状况，这是应用无法控制的。所以 Spring 额外提供了两个保护参数。sendTimeLimit 和 sendBufferSizeLimit。 sendTimeLimit 控制”一次发送最多允许多久”，sendBufferSizeLimit 控制”最多能积压多少数据”。超过任一限制，该会话会被关闭。

同一时刻只有一个线程能往某个客户端发送，其他消息会被缓冲。

1
2
3
4
5
6
7
8
9
10
11
12
@Override
public void configureWebSocketTransport(WebSocketTransportRegistration registration) {
    registration
            // 一次发送最长 15 秒（默认是 10 秒，可按业务调整）
            .setSendTimeLimit(15_000)
            // 单个会话最多积压 512KB（这也是默认值）
            .setSendBufferSizeLimit(512 * 1024)
            // STOMP 消息大小上限（跨 WebSocket 帧分片重组后），默认 64K
            .setMessageSizeLimit(128 * 1024)
            // 连接建立后多久必须收到第一条消息，默认 60 秒
            .setTimeToFirstMessage(60_000);
}
配置项 默认值 说明
messageSizeLimit 64KB 不是单个 WebSocket 消息的大小，而是在 WebSocket 服务器层之上重组后的 STOMP 消息大小
sendTimeLimit 10 秒 入参必须大于 0，否则被忽略
sendBufferSizeLimit 512KB 小于等于 0 会让缓冲实际上被禁用
timeToFirstMessage 60 秒 超过这个时间没收到第一条消息，视为代理或网络异常并关闭

关于 sendTimeLimit，会话的时间上限只在”尝试发送更多消息”时才会被检查；所以如果只发了一条消息而它卡住了，那么除非之后再发一条消息、或者底层物理 socket 自己超时，否则这个会话不会超时。也就是说，如果只发了一条消息而且它卡住了，这个会话不会超时。

关于 messageSizeLimit，理论上，一条 WebSocket 消息的大小几乎可以不受限制。但在实践中，WebSocket 服务器会施加限制，例如 Tomcat 上是 8K，Jetty 上是 64K。出于这个原因，像 stomp-js/stompjs 这样的 STOMP 客户端会在 16K 的边界处切分较大的 STOMP 消息，把它们作为多条 WebSocket 消息发送，而这要求服务端进行缓冲和重新组装。

所以说，Spring 会自动调整 WebSocket 消息大小上限，保证至少能承载 16K 的 WebSocket 消息。 所以你可以放心地把 messageSizeLimit 设成业务需要的大小，不用去关心底层容器的限制。

会话事件与顺序保证

除了前面提到的 setPreservePublishOrder，Spring 还提供了一组会话生命周期事件，用于在连接建立/断开时做业务处理：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@Component
public class WebSocketSessionEventListener {

    private static final Logger log = LoggerFactory.getLogger(WebSocketSessionEventListener.class);

    /** 连接建立（收到 STOMP CONNECT 帧之后） */
    @EventListener
    public void onConnect(SessionConnectEvent event) {
        StompHeaderAccessor accessor = StompHeaderAccessor.wrap(event.getMessage());
        log.info("STOMP 会话建立, sessionId={}, user={}",
                accessor.getSessionId(), accessor.getUser());
    }

    /** 连接就绪 */
    @EventListener
    public void onConnected(SessionConnectedEvent event) {
        // 此时 broker 已经返回 CONNECTED 帧
    }

    /** 订阅事件 */
    @EventListener
    public void onSubscribe(SessionSubscribeEvent event) {
        StompHeaderAccessor accessor = StompHeaderAccessor.wrap(event.getMessage());
        log.info("订阅 destination={}, sessionId={}", accessor.getDestination(), accessor.getSessionId());
    }

    /** 断开事件 */
    @EventListener
    public void onDisconnect(SessionDisconnectEvent event) {
        StompHeaderAccessor accessor = StompHeaderAccessor.wrap(event.getMessage());
        log.info("STOMP 会话断开, sessionId={}, closeStatus={}",
                accessor.getSessionId(), event.getCloseStatus());
    }

    /** broker 可用性变化（接外部 broker 时很有用） */
    @EventListener
    public void onBrokerAvailability(BrokerAvailabilityEvent event) {
        if (!event.isBrokerAvailable()) {
            log.error("消息 broker 不可用，广播能力暂时丧失");
        }
    }
}

这几个事件的语义如下：

  • SessionConnectEvent：在收到一个新的 STOMP CONNECT 时发布（消息里包含 session ID、用户信息、客户端自定义头）；
  • SessionConnectedEvent：在 broker 返回 CONNECTED 帧之后不久发布，这时会话才算完全建立；
  • SessionSubscribeEvent / SessionUnsubscribeEvent：在收到 SUBSCRIBE / UNSUBSCRIBE 时发布；
  • SessionDisconnectEvent：会话结束时发布；
  • BrokerAvailabilityEvent：broker 可用性变化时发布。

关于 SessionDisconnectEvent，一条 DISCONNECT 消息可能来自客户端，也可能是 WebSocket 会话关闭时自动生成的；某些情况下，一个会话的断开事件可能被拦截多次。所以处理断开事件的组件必须幂等。如果不注意这个，例如断开事件里做”在线用户数减一”，没有做幂等保护，就会出现计数变成负数的问题，就跟我之前看到在线人数-17的一样搞笑了。

关于外部 broker 的重连行为，relay 会自动重连”系统”连接，但客户端的连接不会自动重连。*启用了心跳的情况下，客户端通常能在 10 秒内察觉到无响应，但重连逻辑需要客户端自己实现。

WebSocket 作用域

Spring 提供了一个 @WebSocketScope，可以声明生命周期与 WebSocket 会话绑定的 bean：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
@Component
@WebSocketScope
public class SessionContext {

    private final Map<String, Object> data = new ConcurrentHashMap<>();

    @PostConstruct
    public void init() {
        // 依赖注入完成后调用
    }

    @PreDestroy
    public void destroy() {
        // WebSocket 会话结束时调用
    }
}

它的作用就是，你可以声明一个处于 websocket 作用域的 Spring 管理 bean；你可以把它注入到控制器以及注册在 clientInboundChannel 上的任何通道拦截器里。

它的初始化时机是 第一次从控制器访问时，每个 WebSocket 会话一个实例，会话结束才销毁；@PostConstruct 和 @PreDestroy 生命周期方法都会被调用。这个能力适合放会话级的上下文缓存，比如该连接的订阅偏好、限流计数器。

顺便说一下怎么读会话属性。除了注入作用域 bean，你也可以直接从消息头里拿，官方示例是，在 @MessageMapping 方法上加一个 SimpMessageHeaderAccessor 参数，然后调 getSessionAttributes() 拿到那个 map。这个 map 就是挂在每个 WebSocket 会话上的属性集合。

监控：WebSocketMessageBrokerStats

Spring 在启用 @EnableWebSocketMessageBroker 时会自动注册一个统计 bean，这个能力在排查线上问题时非常有用。它的行为是：在使用 @EnableWebSocketMessageBroker 时，一些关键基础设施组件会自动收集统计信息和计数器；配置还会声明一个 WebSocketMessageBrokerStats 类型的 bean，把所有这些信息集中到一处，并默认以 INFO 级别每 30 分钟记录一次日志；这个 bean 可以通过 Spring 的 MBeanExporter 导出到 JMX，以便在运行时查看。

它收集的信息，官方文档列出的内容包括：

  • Client WebSocket Sessions：
    • Current：当前有多少客户端会话，还按 WebSocket 与 SockJS 的 HTTP streaming/polling 分别统计；
    • Total：累计建立过多少会话；
    • Abnormally Closed：
      • Connect Failures：会话建立后 60 秒内没收到任何消息就关闭的——官方说这通常意味着代理或网络有问题；
      • Send Limit Exceeded：因为超过发送超时或发送缓冲区上限而被关闭的会话，官方说这可能发生在慢客户端身上；
      • Transport Errors：传输错误导致的关闭，比如读写 WebSocket 连接失败、或 HTTP 请求/响应失败；
    • STOMP Frames：处理过的 CONNECT、CONNECTED、DISCONNECT 帧总数。
  • STOMP Broker Relay（接外部 broker 时）：
    • TCP Connections：代表客户端 WebSocket 会话建立的到 broker 的 TCP 连接数，应该等于客户端会话数 + 1（那个额外的 1 是应用内部发消息用的共享”系统”连接）；
    • STOMP Frames：转发给 broker 和从 broker 收到的各类帧总数。

另外它还会带上通道线程池的统计，以及 SockJS 任务调度器的统计。

这三个计数器（Connect Failures、Send Limit Exceeded、Transport Errors）是在生产环境排查 WebSocket 问题的第一手资料。

  • Connect Failures 突增 → 去查 nginx 的 proxy_read_timeout、查负载均衡的空闲超时；
  • Send Limit Exceeded 突增 → 说明有一批客户端网络很差（移动端弱网？），要调整 sendBufferSizeLimit 或者做消息降采样；
  • Transport Errors 突增 → 一般是网络层问题或者容器资源耗尽。

我在项目里会把它接到 Micrometer 上做告警，而不是等 30 分钟一次的日志。

客户端：stompjs 的正确用法

服务端讲完了，来看浏览器端。Spring 文档推荐的 JS 库是 stomp-js/stompjs，官方的说法是它是目前维护最活跃的 JavaScript 库。

官方给的示例很简洁：

1
2
3
4
5
6
const stompClient = new StompJs.Client({
    brokerURL: 'ws://domain.com/portfolio',
    onConnect: () => {
        // ...
    }
});

但生产环境要用好它，还有几个关键配置必须设置。下面是我实际项目里用的版本：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
import { Client } from '@stomp/stompjs';

const client = new Client({
  // 注意：如果页面是 https，这里必须是 wss，否则会被浏览器以混合内容为由拦截
  brokerURL: `${location.protocol === 'https:' ? 'wss' : 'ws'}://${location.host}/ws/stomp`,

  // 鉴权：浏览器原生 WebSocket 不支持自定义头，
  // 但 STOMP 的 CONNECT 帧可以带头部，所以可以在这里放 token
  connectHeaders: {
    Authorization: `Bearer ${accessToken}`
  },

  // 心跳：这里的值要和 nginx 的 proxy_read_timeout 匹配
  // 客户端希望 10 秒发一次心跳，服务端希望 10 秒发一次
  heartbeatIncoming: 10000,
  heartbeatOutgoing: 10000,

  // 断线重连间隔（毫秒）。stompjs 内置了重连，不要自己再写一套
  reconnectDelay: 5000,

  // 调试日志，生产环境应关掉
  debug: (msg) => console.debug('[STOMP]', msg),

  onConnect: (frame) => {
    console.info('STOMP 已连接', frame.headers['version']);

    // 订阅广播频道
    const topicSub = client.subscribe('/topic/order/A1001', (message) => {
      const payload = JSON.parse(message.body);
      renderOrder(payload);
    });

    // 订阅用户私有队列：注意 /user 前缀
    const userSub = client.subscribe('/user/queue/notifications', (message) => {
      showNotification(JSON.parse(message.body));
    });

    // 组件卸载时记得取消订阅，否则会泄漏
    // topicSub.unsubscribe();
    // userSub.unsubscribe();
  },

  onStompError: (frame) => {
    // STOMP 协议级错误（服务端发来的 ERROR 帧）
    console.error('STOMP 错误', frame.headers['message'], frame.body);
  },

  onWebSocketError: (event) => {
    console.error('WebSocket 层错误', event);
  },

  onWebSocketClose: (event) => {
    // 1000 是正常关闭；其他码要关注
    console.warn('WebSocket 已关闭, code =', event.code);
  }
});

client.activate();

// 页面卸载时优雅断开
window.addEventListener('beforeunload', () => {
  client.deactivate();
});

这里有几个必须强调的点：

  1. reconnectDelay 交给 stompjs，不要自己再写重连。 自己写会和库的内置重连冲突，导致连接数翻倍、服务器压力暴增。
  2. 心跳要和网关超时匹配。 如果 nginx proxy_read_timeout 是 60 秒，而客户端心跳是 90 秒，那连接会在客户端发出心跳之前就被 nginx 掐掉。心跳间隔应该显著小于 proxy_read_timeout，我的经验值是取其 1/4 到 1/6。
  3. 页面是 HTTPS 时必须用 wss://。 混合内容会被浏览器直接拦截，报错信息还很不直观。
  4. 组件卸载要 unsubscribe，页面卸载要 deactivate。 否则会累积无效订阅，服务端会持续往死连接发消息。
  5. token 过期问题。 STOMP 的 connectHeaders 只在 CONNECT 时发送一次，如果 token 在中途过期，连接不会自动断开。这需要业务上配合：要么 token 有效期足够长（覆盖会话时长），要么在 onStompError 里捕获服务端的过期通知并主动重连。

Spring 下的 SSE 实战

SSE 在 Spring 里有三种实现路径

路径 技术栈 核心类型 适用场景
Spring MVC + SseEmitter Servlet 栈 SseEmitter 现有 Spring MVC 项目改造，最常见
Spring WebFlux + Flux 反应式栈 Flux<ServerSentEvent<T>> 新项目、需要背压、连接数很大
纯 HTTP 手写流 任意 HttpServletResponse / OutputStream 需要完全控制输出格式

我给出的选择建议是：

  • 已经在用 Spring MVC 的存量项目：直接用 SseEmitter，改造成本最低；
  • 新建项目、预计连接数上万：考虑 WebFlux，它用少量线程就能撑住大量长连接；
  • 需要输出非标准格式（比如自定义的心跳、多路复用）：手写 OutputStream，但失去了框架帮你做的生命周期管理。

先看最常用的 SseEmitter。

Spring MVC：SseEmitter 全面解析

最小实现

Spring 官方文档给的例子非常简洁：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
@GetMapping(path="/events", produces=MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter handle() {
    SseEmitter emitter = new SseEmitter();
    // Save the emitter somewhere..
    return emitter;
}

// In some other thread
emitter.send("Hello once");

// and again later on
emitter.send("Hello again");

// and done at some point
emitter.complete();

三个要点：

  1. 注解上必须声明 produces = MediaType.TEXT_EVENT_STREAM_VALUE，也就是 text/event-stream。少了这个，浏览器不会把它当事件流处理，EventSource 会直接失败。
  2. 把 emitter 存起来（官方注释里的 “Save the emitter somewhere”），因为后续推送发生在别的线程里。
  3. 推送完一定要 complete()，否则这个连接会一直挂着，直到超时。

SseEmitter 继承自 ResponseBodyEmitter，而 ResponseBodyEmitter 是 Spring MVC 异步请求机制的一部分。

Java 对它的定位是：这是一个控制器方法返回值类型，用于异步请求处理，可以把一个或多个对象写进响应；

超时语义

SseEmitter 的无参构造和带超时构造的语义很明白。

带超时构造的说明是，用一个自定义的超时值创建 SseEmitter。默认情况下不设置这个值，此时会使用 MVC Java Config 或 MVC 命名空间里配置的默认值；如果那个也没设置，那么超时就取决于底层服务器的默认值。

也就是说超时值的三级优先级是：

  1. new SseEmitter(timeout) 显式指定的值；
  2. MVC Java Config 或 MVC namespace 里配置的默认值；
  3. 底层容器的默认值。

这个链条很重要，它解释了那个非常经典的线上问题：“我的 SSE 连接总是在 30 秒左右断开”。

原因通常就出在第三条上：Servlet 异步请求有一个超时值，而它是容器/容器配置决定的。关于这个默认值，社区的普遍经验值是 30 秒。

所以说不要依赖默认值够，而是必须显式设置。只要你不设，它就是一个你不确定的数字

去 MVC 配置里改默认值：

1
2
// 方式一：每个 emitter 单独指定
SseEmitter emitter = new SseEmitter(30 * 60 * 1000L); // 30 分钟
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
// 方式二：全局配置 MVC 异步支持的默认超时
@Configuration
public class WebMvcAsyncConfig implements WebMvcConfigurer {

    @Override
    public void configureAsyncSupport(AsyncSupportConfigurer configurer) {
        // 默认超时设为 30 分钟
        configurer.setDefaultTimeout(30 * 60 * 1000L);
        // 关键：指定用于流式写入的线程池
        configurer.setTaskExecutor(sseTaskExecutor());
    }

    @Bean
    public ThreadPoolTaskExecutor sseTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(8);
        executor.setMaxPoolSize(32);
        executor.setQueueCapacity(200);
        executor.setThreadNamePrefix("sse-writer-");
        executor.initialize();
        return executor;
    }
}

setDefaultTimeout 的语义是：在异步请求处理超时之前的时间量，单位毫秒。这个超时从主请求处理线程退出之后开始计，到请求因为要继续处理并发产生的结果而被再次派发时结束。如果这个值没有被设置，就使用底层实现的默认超时。

关于 setTaskExecutor，这个 AsyncTaskExecutor 用于两件事：一是以 Reactive 类型做流式输出时的阻塞写，二是执行控制器方法返回的 Callable 实例；

默认的 AsyncTaskExecutor 不适合生产环境负载。 所以显式配置一个带名字前缀、有界队列的线程池，是 SSE 上生产前的必做项。

Spring MVC 的 SSE 写入是阻塞的，跑在一个单独的线程上。 它直接决定了 MVC 方案的容量上限，每一个活跃的 SSE 连接，在写入时都要占用一个线程。而WebFlux 不需要为每次写操作分配额外线程。

event() 构建器：完整的事件表达能力

SseEmitter 提供了静态方法 event() 返回一个 SseEventBuilder，它能表达 SSE 规范里的全部字段。Javadoc 里给的示例是：

1
2
3
// static import of SseEmitter
SseEmitter emitter = new SseEmitter();
emitter.send(event().name("update").id("1").data(myObject));

这个构建器的方法和它们对应的 SSE 字段如下

构建器方法 写入的 SSE 字段 说明
id(String id) id 事件 ID，支撑断点续传
name(String eventName) event 注意方法名叫 name，写的是 event: 字段
reconnectTime(long reconnectTimeMillis) retry 告诉浏览器下次重连等多久
comment(String comment) 注释行 多行注释会被拆成多行注释
data(Object object) data 写入一行 data
data(Object object, MediaType mediaType) data 指定序列化方式
build() — 返回要写入的对象集合

关于 data 的两种重载：如果 data 参数是一个多行的 String，它会被拆成多个 data 字段行；如果 data 不是 String，它会被编码成 JSON。

这一点正好解决了我在讲 SSE 规范时提到的那个问题，多行内容必须拆成多个 data: 字段。而 Spring 的构建器帮你做了这件事，但前提是你要用带 MediaType 的那个重载，或者传入的是多行字符串。如果你直接 send(Object)，它等价于 event().data(obj)，也就是把对象格式化成单行的 data 行。

完整的用法，以一个消息推送服务的实现为例子：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
@Service
public class SsePushService {

    private static final Logger log = LoggerFactory.getLogger(SsePushService.class);

    /** 按业务频道维护连接。真实项目应加上用户维度索引，并考虑放 Redis 以便多实例共享 */
    private final Map<String, Set<SseEmitter>> channels = new ConcurrentHashMap<>();

    /**
     * 建立连接
     */
    public SseEmitter subscribe(String channel, Long lastEventId) {
        // 30 分钟超时：要显著大于前端期望的单次连接时长，同时避免连接泄漏
        SseEmitter emitter = new SseEmitter(30 * 60 * 1000L);
        channels.computeIfAbsent(channel, k -> ConcurrentHashMap.newKeySet()).add(emitter);

        // 生命周期回调：这三行是防内存泄漏的关键
        emitter.onCompletion(() -> {
            log.info("SSE 正常完成, channel={}", channel);
            remove(channel, emitter);
        });
        emitter.onTimeout(() -> {
            log.info("SSE 超时, channel={}", channel);
            // 超时后必须自己 complete，否则资源不会被释放
            emitter.complete();
            remove(channel, emitter);
        });
        emitter.onError((ex) -> {
            log.warn("SSE 异常, channel={}", channel, ex);
            remove(channel, emitter);
        });

        // 立即发一条事件，让客户端确认连接可用
        try {
            emitter.send(SseEmitter.event()
                    .name("connected")           // event 字段：事件类型
                    .id(String.valueOf(nowEventId()))  // id 字段：用于断点续传
                    .data(Map.of("channel", channel, "serverTime", System.currentTimeMillis())));
        } catch (IOException e) {
            remove(channel, emitter);
            throw new IllegalStateException("SSE 连接初始化失败", e);
        }

        // 如果客户端带了 Last-Event-ID，这里补发丢失的消息
        if (lastEventId != null) {
            replaySince(channel, lastEventId, emitter);
        }
        return emitter;
    }

    /**
     * 推送一条业务事件
     */
    public void publish(String channel, String eventName, Object payload) {
        Set<SseEmitter> subscribers = channels.getOrDefault(channel, Set.of());
        for (SseEmitter emitter : subscribers) {
            try {
                emitter.send(SseEmitter.event()
                        .name(eventName)                     // 自定义事件类型
                        .id(String.valueOf(nowEventId()))    // 事件 ID，支撑断点续传
                        .data(payload, MediaType.APPLICATION_JSON));  // 用 JSON 序列化
            } catch (IOException | IllegalStateException ex) {
                // 发送失败说明这个连接已经不可用，直接清理
                log.warn("SSE 推送失败, 移除连接, channel={}", channel, ex);
                remove(channel, emitter);
            }
        }
    }

    /**
     * 纯注释心跳：规范建议每 15 秒左右发一次
     */
    public void heartbeat(String channel) {
        for (SseEmitter emitter : channels.getOrDefault(channel, Set.of())) {
            try {
                emitter.send(SseEmitter.event().comment("keep-alive"));
            } catch (IOException | IllegalStateException ex) {
                remove(channel, emitter);
            }
        }
    }

    /**
     * 主动结束某个连接，让前端不要再重连
     */
    public void close(String channel) {
        for (SseEmitter emitter : channels.getOrDefault(channel, Set.of())) {
            emitter.complete();
        }
        channels.remove(channel);
    }

    private void remove(String channel, SseEmitter emitter) {
        Set<SseEmitter> set = channels.get(channel);
        if (set != null) {
            set.remove(emitter);
            if (set.isEmpty()) {
                channels.remove(channel, set);
            }
        }
    }
}

关于生命周期回调：

  • onCompletion：当异步请求因为任何原因完成时，由容器线程调用，包括超时和网络错误。Javadoc 说这个方法很有用，因为可以用它侦测到一个 ResponseBodyEmitter 实例已经不再可用了；
  • onTimeout：当异步请求超时时，由容器线程调用；
  • onError：当处理异步请求过程中发生错误时，由容器线程调用。

onCompletion 覆盖了包括超时和网络错误在内的所有情况，所以它其实是清理资源的兜底位置；而 onTimeout 和 onError 更适合做分类统计和日志。

而且，emitter 抛 IOException 时，不要自己去 complete 或 completeWithError。 容器会接管后续处理。这也是为什么我上面的代码在捕获异常后只做了从集合里移除，而没有再去调用 complete()。

comment("keep-alive") 对应 SSE 规范里的注释行，它不产生任何事件，只用于保活。这是 WHATWG 规范建议的每 15 秒发一个注释行。

断开检测是 Servlet 的先天不足

Servlet API 不提供”远程客户端已离开”的通知。

因此，在往响应做流式输出时，无论是通过 SseEmitter 还是 Reactive 类型，定期发送数据都很重要，因为一旦客户端已经断开，写操作就会失败。这个发送可以采用空内容的 SSE 事件。

这意味着什么？意味着当用户直接关掉浏览器标签页时，服务端不会立刻知道。 你不会有任何回调被触发，那个 SseEmitter 对象还安静地躺在你的 Map 里占着内存。

那怎么发现连接已经死了？只能是定期发数据，靠写入失败来发现。

因此这里，心跳不只是保活手段，它同时是连接探活手段。所以一个正确的 SSE 服务，心跳是必选项而不是可选项：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
@Component
public class SseHeartbeatScheduler {

    private final SsePushService pushService;

    public SseHeartbeatScheduler(SsePushService pushService) {
        this.pushService = pushService;
    }

    /**
     * 每 15 秒给所有频道发一次注释心跳。
     * 15 秒这个数字来自 WHATWG 规范的建议，同时远小于 nginx 默认的 60 秒读超时。
     */
    @Scheduled(fixedDelay = 15_000)
    public void heartbeat() {
        pushService.heartbeatAll();
    }
}

如果你觉得手动管理心跳和断连太麻烦，那就用 WebSocket 的 STOMP，它有内建心跳。

内存与资源管理清单

SSE 在 MVC 下最容易出的问题是连接泄漏，症状是内存缓慢增长、最后 OOM 或者被连接数打满。下面是我总结的必查清单：

  1. 有没有设 onCompletion / onTimeout / onError 三个回调？ 它们是释放资源的唯一入口。少一个都会泄漏。
  2. 超时后有没有主动调用 complete()？ onTimeout 触发时容器只是通知你超时了，不会自动清理 emitter。
  3. 心跳有没有在跑？ 没有心跳，客户端断开的连接永远不会被发现和清理。
  4. 超时时间是否合理？ 太长会导致”僵尸连接”堆积；太短会导致客户端频繁重连。 我的经验值是：超时时间设为前端期望单次连接时长的 1.5 到 2 倍。比如前端希望连接保持 30 分钟然后主动重连，那服务端超时可以设 45-60 分钟。
  5. 连接池有没有上限？ 单个用户的连接数、单个频道的订阅者数量都要有上限，否则一个恶意用户开一万个连接就能打垮服务。
  6. 推送失败后有没有移除？ send 抛异常时一定要把 emitter 从集合里摘掉，否则下次推送还会去试这个死连接。

第 5 点给出一个实际的防护写法：

1
2
3
4
5
6
7
8
9
10
11
/** 单用户在单频道最多允许的连接数 */
private static final int MAX_CONNECTIONS_PER_USER = 3;

public SseEmitter subscribe(String channel, Long userId, Long lastEventId) {
    long current = countByUser(channel, userId);
    if (current >= MAX_CONNECTIONS_PER_USER) {
        // 返回 noContent 让浏览器停止重连，避免它死循环
        throw new TooManyConnectionsException("同一用户连接数过多");
    }
    // ...
}

这里用 204 语义来拒绝是个技巧：任何非 200 响应都会让浏览器判定连接失败并彻底停止重连（前面讲过这个机制），所以 204 是最语义化的选择。反过来说，如果你用 500 拒绝，效果和 204 一样——浏览器也不会重试。这两种情况都不会”放大压力”，但也都会让客户端永久失联，所以只适合”这个用户确实不该再收到推送”的场景。如果只是想暂时限流，应该建立连接后立刻 complete()，让浏览器走正常的重连流程。

Spring WebFlux：反应式 SSE

如果你的项目是 WebFlux，或者连接数会非常大，那响应式方案是更好的选择。Spring 也提供了 ServerSentEvent 这个类型来表达 SSE 事件的各个字段。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
@RestController
public class ReactiveSseController {

    private final Sinks.Many<OrderEvent> sink = Sinks.many().multicast().onBackpressureBuffer();

    /**
     * 返回一个 Flux，框架会自动以 SSE 格式写出。
     * produces 同样必须是 text/event-stream
     */
    @GetMapping(path = "/api/orders/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<OrderEvent>> stream() {
        return sink.asFlux()
                .map(event -> ServerSentEvent.<OrderEvent>builder()
                        .id(event.eventId())        // 对应 SSE 的 id 字段
                        .event("order-update")      // 对应 event 字段
                        .data(event)                // 对应 data 字段
                        .retry(Duration.ofSeconds(5)) // 对应 retry 字段
                        .comment("ok")              // 对应注释行
                        .build());
    }

    /** 业务侧推送：往 sink 里写就是往所有订阅者推 */
    public void publish(OrderEvent event) {
        sink.tryEmitNext(event);
    }
}

ServerSentEvent 是用于 Spring 反应式 Web 支持的 Server-Sent Event 表示；Flux<ServerSentEvent> 或 Observable<ServerSentEvent> 是 Spring MVC 的 SseEmitter 在反应式侧的等价物。

它的 builder 方法对应的字段和 MVC 侧一样：id()、event()、retry()、comment()、data()。关于 retry，它的入参类型是 Duration，而不是毫秒数

WebFlux 侧关于返回值，Flux<ServerSentEvent>、Observable<ServerSentEvent> 或其他响应式类型，都会发出 Server-Sent Events；如果只需要写数据，ServerSentEvent 这个包装类型可以省略，但此时必须通过 produces 属性请求或在映射里声明 text/event-stream。

也就是说，如果你不关心 id / event / retry 这些字段，直接返回 Flux<MyData> 也可以，只是produces 千万不能忘。

一个提醒：Flux 的事件流是有背压语义的。 如果客户端消费不过来，反应式流会施加背压，而不是像 MVC 那样无脑往缓冲区里堆。这在保护服务端内存方面是个优势，但也要注意：某些背压策略下，慢客户端会导致事件堆积在内存里（比如用了 onBackpressureBuffer），这时仍然需要设置缓冲区上限和溢出策略。

1
2
3
// 更安全的写法：缓冲区有界 + 溢出时丢弃最旧的事件 + 记录告警
Sinks.Many<OrderEvent> sink = Sinks.many().multicast()
        .onBackpressureBuffer(1024, event -> log.warn("SSE 缓冲区溢出，丢弃事件: {}", event));

WebFlux 的方案还有一个隐性优势：它不需要为每个连接分配一个写线程。所以：

Spring MVC + SseEmitter WebFlux + Flux
每连接的线程开销 写入时占用一个线程 无额外线程（非阻塞 IO）
背压支持 有（通过 reactive types），但写入仍阻塞 原生支持
万级连接 需要仔细规划线程池 相对轻松
迁移成本 低（存量 MVC 项目） 高（需要整个栈反应式）

最后，WebFlux 是一个全栈选择，不是局部优化。 如果你只在 SSE 这一个接口上用 WebFlux，而其他接口还是阻塞式的 MVC，那你会失去 WebFlux 的大部分收益，还引入了两套编程模型的复杂度。要上就上全套。

纯 HTTP 手写流

有时候框架抽象反而碍事，比如你想做多路复用（一个 SSE 连接推多个频道）、或者想完全自己控制输出格式。这时可以直接操作 HttpServletResponse。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
@GetMapping("/api/raw/stream")
public void rawStream(HttpServletResponse response) throws IOException {
    // 1. 设置 SSE 要求的响应头
    response.setContentType(MediaType.TEXT_EVENT_STREAM_VALUE);
    response.setCharacterEncoding(StandardCharsets.UTF_8.name());
    response.setHeader("Cache-Control", "no-cache");
    response.setHeader("Connection", "keep-alive");
    // 2. 告诉 nginx 不要缓冲这个响应（nginx 官方文档确认支持此头）
    response.setHeader("X-Accel-Buffering", "no");

    PrintWriter writer = response.getWriter();

    try {
        while (true) {
            String payload = fetchNextEvent();
            // 3. 严格按 SSE 格式写：data 行 + 空行
            writer.write("id: " + nextId() + "\n");
            writer.write("event: update\n");
            // 多行内容要拆成多个 data 字段
            for (String line : payload.split("\n", -1)) {
                writer.write("data: " + line + "\n");
            }
            writer.write("\n");   // 4. 空行才是消息边界，绝对不能漏
            writer.flush();       // 5. 必须 flush，否则数据会留在缓冲区里

            Thread.sleep(1000);
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    } catch (IOException e) {
        // 客户端断开时会在这里抛异常，直接结束即可
    }
}

手写流的五个关键点，正好对应我在讲 SSE 规范时提到的要求：

  1. Content-Type 必须是 text/event-stream；
  2. 编码必须是 UTF-8；
  3. 消息以空行结尾（两个连续换行）；
  4. 多行内容拆成多个 data: 字段；
  5. 每次写完必须 flush()。

第 5 点在手写方案里尤其重要。很多人写完之后发现”服务端明明发了，客户端要等好久才收到”，就是因为没有 flush——数据卡在 PrintWriter 或者容器缓冲区里了。

关于 X-Accel-Buffering: no，nginx 官方文档确认了这个头的存在和作用，缓冲也可以通过 X-Accel-Buffering 响应头字段传入 yes 或 no 来开关；这个能力可以用 proxy_ignore_headers 指令禁用。

这个头的价值在于，你可以在应用层控制 nginx 的缓冲行为，而不用去改 nginx 配置。对于 SSE 这种”只有某些接口需要关缓冲”的场景，这比全局关掉 proxy_buffering 更精细。

前端实现要点

服务端写好了，前端也有几个坑要避开。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
/**
 * SSE 客户端封装。
 * 关键设计：整个应用只维护一个 EventSource 实例，避免触发浏览器的每域名连接数限制。
 */
class SseClient {
  constructor(url) {
    this.url = url;
    this.source = null;
    this.handlers = new Map();
    this.lastEventId = null;
  }

  connect() {
    // 注意：EventSource 的 URL 里不能放太长的 token（会被日志记录，且有长度限制）
    const url = this.lastEventId
        ? `${this.url}?lastEventId=${encodeURIComponent(this.lastEventId)}`
        : this.url;

    // withCredentials: true 才会带上跨域的 Cookie
    this.source = new EventSource(url, { withCredentials: true });

    this.source.onopen = () => {
      console.info('SSE 连接已建立');
    };

    // 默认事件类型是 message
    this.source.onmessage = (event) => {
      this.recordId(event);
      this.dispatch('message', event);
    };

    this.source.onerror = () => {
      // 关键：不要自己重建 EventSource！
      // readyState 为 0 表示浏览器正在自动重连，什么都不用做。
      // readyState 为 2 表示浏览器已放弃重连，这时才需要人工介入（比如提示用户刷新）。
      if (this.source.readyState === EventSource.CLOSED) {
        console.error('SSE 连接已关闭且不会重连，请检查服务端是否返回了 204');
      } else {
        console.warn('SSE 连接异常，浏览器正在自动重连');
      }
    };

    // 自定义事件类型必须用 addEventListener
    this.source.addEventListener('order-update', (event) => {
      this.recordId(event);
      this.dispatch('order-update', event);
    });
  }

  /** 记录 lastEventId，用于重连时主动带上 */
  recordId(event) {
    if (event.lastEventId) {
      this.lastEventId = event.lastEventId;
    }
  }

  on(eventName, handler) {
    if (!this.handlers.has(eventName)) {
      this.handlers.set(eventName, []);
    }
    this.handlers.get(eventName).push(handler);
    return this;
  }

  dispatch(eventName, event) {
    (this.handlers.get(eventName) || []).forEach((h) => {
      try {
        // 注意 event.data 是字符串，业务数据要自己解析
        h(event.data);
      } catch (e) {
        console.error('SSE 事件处理失败', eventName, e);
      }
    });
  }

  close() {
    if (this.source) {
      this.source.close();
      this.source = null;
    }
  }
}

// 全局单例，整个页面共用一条连接
export const sseClient = new SseClient('/api/notifications/stream');

这里有三个前端层面的关键认知：

  • 不要自己写重连。 EventSource 内置了重连，你在 onerror 里再 new 一个实例，就会出现两条连接甚至更多。判断依据是 readyState：0 表示浏览器正在重连，2 表示彻底关闭了。
  • lastEventId 其实不用你操心。 浏览器在自动重连时会自动带上 Last-Event-ID 请求头，你不需要手动拼 URL 参数。上面代码里手动带 lastEventId 只是为了处理”页面刷新后重连”这种浏览器状态丢失的场景——刷新之后内存里的 lastEventId 没了，但如果你把它存到 sessionStorage，就可以在刷新后继续续传。
  • 一个页面一条连接。 这是前面讲的每域名 6 连接限制的直接应对。如果页面有多个模块都需要实时数据，让它们复用同一个 EventSource，用 event 字段来区分，而不是各开一条连接。

对应的服务端取 Last-Event-ID 的写法：

1
2
3
4
5
6
7
8
9
@GetMapping(path = "/api/notifications/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter stream(
        @RequestParam(required = false) Long userId,
        // 浏览器自动重连时会带上这个头
        @RequestHeader(value = "Last-Event-ID", required = false) String lastEventId) {

    Long resumeFrom = parseEventId(lastEventId);
    return ssePushService.subscribe("user:" + userId, resumeFrom);
}

Last-Event-ID 是请求头，所以用 @RequestHeader 取。

安全相关

功能和性能的问题通常能在测试环境暴露，但安全问题的特点是：它在开发阶段完全正常，上线后才被利用。

这一部分主要基于 Spring Security 官方 WebSocket 安全文档。

同源策略对 WebSocket 不生效

浏览器不会对 WebSocket 连接执行同源策略。

文档给了一个非常具体的攻击场景来说明后果：

设想如下场景：一个用户访问 bank.com 并登录了自己的账户。然后同一个用户在浏览器里又打开一个标签页，访问 evil.com。同源策略保证了 evil.com 无法读取或写入 bank.com 的数据。

但是对 WebSocket 来说，同源策略不适用。事实上，除非 bank.com 明确禁止，否则 evil.com 可以代表这个用户读写数据。这意味着用户能通过 WebSocket 做的任何事情（比如转账），evil.com 都能代表他去做。

整理成攻击步骤：

  1. 用户登录了 bank.com，浏览器里有了登录态的 Cookie；
  2. 用户在另一个标签页打开了 evil.com；
  3. evil.com 的脚本发起一个到 bank.com 的 WebSocket 连接；
  4. 因为这个连接会自动携带 bank.com 的 Cookie，服务端认为这是合法用户；
  5. 同源策略不管 WebSocket，所以这个连接被建立；
  6. 于是 evil.com 就可以以用户身份通过 WebSocket 做任何事，包括转账。

文档把 SockJS 也纳入了警告范围，SockJS 因为要模拟 WebSocket，同样绕过了同源策略，所以用 SockJS 也必须显式防护。

Origin 校验与 Spring 的默认行为

好消息是，Spring 从 4.1.5 起就默认只接受同源请求了。Spring 的 WebSocket 和 SockJS 支持就把访问限制在当前域名内了；而 Spring Security 在此基础上又增加了一层保护，形成纵深防御。

前面已经讲过三种配置模式（只允许同源、允许指定列表、允许全部）：

  • 默认只允许同源，这是安全的默认值，不要轻易改成 *；
  • 用 setAllowedOriginPatterns 而不是 setAllowedOrigins，前者支持通配模式且是 Spring 推荐的方式；
  • Origin 头可以被非浏览器客户端伪造，所以它只防浏览器侧的攻击（CSRF 方向），不能替代认证。

Spring Security 会在此基础上再加一层，形成纵深防御。这就是官方说的 defense in depth。

消息级授权

Origin 校验只保护了”握手这一次”。握手成功之后，连接里流动的所有 STOMP 消息都不再经过 Web 安全过滤器链了。

新 API：@EnableWebSocketSecurity

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
@Configuration
@EnableWebSocketSecurity
public class WebSocketSecurityConfig {

    @Bean
    AuthorizationManager<Message<?>> messageAuthorizationManager(
            MessageMatcherDelegatingAuthorizationManager.Builder messages) {

        messages
            // 1. 没有 destination 的消息（比如 CONNECT、DISCONNECT、心跳）要求已认证
            .nullDestMatcher().authenticated()

            // 2. 允许任何人订阅错误队列（否则连接失败时客户端收不到错误信息）
            .simpSubscribeDestMatchers("/user/queue/errors").permitAll()

            // 3. 发往 /app 的应用消息要求 USER 角色
            .simpDestMatchers("/app/**").hasRole("USER")

            // 4. 订阅用户私有目的地和特定主题要求 USER 角色
            .simpSubscribeDestMatchers("/user/**", "/topic/notifications").hasRole("USER")

            // 5. 禁止客户端直接往 broker 目的地发消息（防止伪造系统通知）
            .simpDestMatchers("/topic/**", "/queue/**").denyAll()

            // 6. 其他一律拒绝，避免遗漏
            .anyMessage().denyAll();

        return messages.build();
    }
}

这几条规则用到的匹配器如下：

匹配器 匹配范围
nullDestMatcher() destination 头为 null 的消息，也就是 CONNECT、心跳、UNSUBSCRIBE、DISCONNECT 这类
simpDestMatchers(String...) 按目的地匹配，不区分消息类型
simpMessageDestMatchers(String...) 只匹配 MESSAGE 类型的消息
simpSubscribeDestMatchers(String...) 只匹配 SUBSCRIBE 类型的消息
simpTypeMatchers(SimpMessageType...) 按消息类型匹配
anyMessage() 兜底匹配

关于 @EnableWebSocketSecurity 这个注解的两个隐含行为：

  1. 任何入站的 CONNECT 消息都需要一个合法的 CSRF token，以此来强制同源策略。

    如果你用 @EnableWebSocketSecurity，客户端必须正确传 CSRF token，否则连 CONNECT 都会失败。

  2. 对于任何入站请求，SecurityContextHolder 都会被填充用户信息，放在 simpUser 头属性里。

区分 SUBSCRIBE 和 MESSAGE

客户端只能请求，不能广播。广播是服务端的特权。

设想一个聊天应用：

  • 系统可以通过 /topic/system/notifications 这个目的地，向所有用户发送通知类 MESSAGE；
  • 客户端可以通过 SUBSCRIBE 到 /topic/system/notifications 来接收通知。

我们当然希望客户端能够 SUBSCRIBE 到 /topic/system/notifications，但我们不希望允许它们向这个目的地发送 MESSAGE。因为如果我们允许向 /topic/system/notifications 发送 MESSAGE，客户端就能直接往这个端点发消息，从而冒充系统。

如果允许客户端往广播目的地发消息，客户端就能冒充系统。

总的来说，应用的常见做法是：拒绝任何发往以 broker 前缀（/topic/ 或 /queue/）开头的目的地的 MESSAGE，客户端只能往 /app/** 发，由服务端决定要不要广播出去。

理解 destination 转换，才能写对规则

  • 用户可以通过向 /app/chat 发消息来把消息发给某个特定用户；
  • 应用看到这条消息，确保 from 属性被指定为当前用户（我们不能信任客户端）；
  • 然后应用通过 SimpMessageSendingOperations.convertAndSendToUser("toUser", "/queue/messages", message) 把消息发给接收者；
  • 这条消息的目的地会被转换成 /queue/user/messages-<sessionid>。

注意最后一步的转换：

1
2
3
convertAndSendToUser("toUser", "/queue/messages", msg)
        ↓
目的地变成 /queue/user/messages-<sessionid>

在这个聊天应用里，我们想让客户端监听 /user/queue，它会被转换成 /queue/user/messages-<sessionid>。但是，我们绝不能让客户端有能力监听 /queue/*，因为那会让客户端看到所有用户的消息。

所以说，客户端订阅 /user/queue/messages 是安全的，但你必须禁止它订阅 /queue/*。因为 /queue/* 是转换后的真实目的地空间，里面装着所有用户的消息。如果你开放了 /queue/** 的订阅权限，任何一个客户端都能订阅到别人的私信。

授权规则要写在”客户端声明的目的地”上（/user/...），而不是”转换后的真实目的地”上（/queue/...），并且必须显式拒绝客户端直接访问转换后的空间。

Token 认证：从 STOMP 头里取凭证

前面提到，浏览器原生 WebSocket 不支持自定义请求头，但 STOMP 的 CONNECT 帧可以带头部。所以企业项目里标准的做法是把 JWT 放在 STOMP 的 CONNECT 头里。

STOMP 协议确实在 CONNECT 帧上提供了 login 和 passcode 这两个头。它们最初是为 STOMP over TCP 设计的，在那个场景下也是必需的。但对于 STOMP over WebSocket，Spring 默认会忽略 STOMP 协议层面的认证头，并假定用户已经在 HTTP 传输层完成了认证。

注意这个默认行为，Spring 默认忽略 STOMP 层的认证头，它假设你已经在 HTTP 传输层完成认证了。

所以有两条路线。

依赖 HTTP 层认证

对于 WebSocket 握手请求、或者 SockJS 的 HTTP 传输请求来说，通常已经存在一个已认证用户，可以通过 HttpServletRequest#getUserPrincipal() 拿到。Spring 会自动把这个用户与该用户创建的 WebSocket 或 SockJS 会话关联起来，并进而通过一个 user 头，关联到这条会话上传输的所有 STOMP 消息。简而言之，一个典型的 Web 应用除了它原本为安全所做的事之外，不需要做任何额外的事。

如果你的应用用 Cookie + Session 认证，那基本上什么都不用做：用户在 HTTP 层认证 → Session 里存着安全上下文 → WebSocket 握手时 Spring 自动关联用户 → 之后每条 STOMP 消息都会自动带上用户信息。

这就是用 WebSocket 时，能用 Cookie/Session 就用，别硬上 JWT。

从 STOMP CONNECT 头里取 token（前后端分离）

如果架构是前后端分离 + JWT，Session 不可用，那就需要自定义一个 ChannelInterceptor。

首先，RFC 6455 并没有规定服务端在 WebSocket 握手期间认证客户端的具体方式；而浏览器客户端只能用标准的认证头（比如 HTTP Basic）或 Cookie，不能发自定义头；SockJS 的 JavaScript 客户端同样不能在传输请求里发 HTTP 头。虽然 token 也可以放在查询参数里，但那样 token 可能会随着 URL 一起被无意中记录到服务端日志里

所以官方推荐的两步方案是：客户端在 connect 时把认证头放进 STOMP 头里，服务端用一个 ChannelInterceptor 处理它。而且官方特别强调，只需要在 CONNECT 这条消息上做认证并设置 user 头，Spring 会记住这个用户，并把它关联到同一会话后续的所有 STOMP 消息上。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
@Configuration
@EnableWebSocketMessageBroker
public class StompAuthConfig implements WebSocketMessageBrokerConfigurer {

    private final JwtService jwtService;

    public StompAuthConfig(JwtService jwtService) {
        this.jwtService = jwtService;
    }

    @Override
    public void configureClientInboundChannel(ChannelRegistration registration) {
        registration.interceptors(new ChannelInterceptor() {
            @Override
            public Message<?> preSend(Message<?> message, MessageChannel channel) {
                StompHeaderAccessor accessor =
                        MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
                if (accessor == null) {
                    return message;
                }

                // 只在 CONNECT 帧做认证，认证结果写入会话属性
                if (StompCommand.CONNECT.equals(accessor.getCommand())) {
                    String bearer = accessor.getFirstNativeHeader("Authorization");
                    if (bearer == null || !bearer.startsWith("Bearer ")) {
                        throw new MessagingException("缺少 Authorization 头");
                    }
                    String token = bearer.substring(7);

                    Authentication auth = jwtService.parseAuthentication(token);
                    if (auth == null) {
                        throw new MessagingException("token 无效或已过期");
                    }

                    // 关键：把用户绑定到这个 STOMP 会话
                    accessor.setUser(auth);

                    // 也可以把业务信息放进会话属性，后续 @MessageMapping 里能取到
                    Map<String, Object> sessionAttrs = accessor.getSessionAttributes();
                    if (sessionAttrs != null) {
                        sessionAttrs.put("tenantId", jwtService.parseTenantId(token));
                    }
                }
                return message;
            }
        });
    }
}

这个拦截器做了三件事：

  1. 只在 CONNECT 帧做认证，因为认证只需要做一次；
  2. accessor.setUser(auth) 把认证结果绑定到 STOMP 会话上，之后同一个会话里的所有消息都会带上这个用户（这就是为什么 @MessageMapping 方法里能直接注入 Principal）；
  3. 认证失败抛 MessagingException，连接会被拒绝。

这里有个顺序问题要特别注意，自定义的认证拦截器必须排在 Spring Security 的前面，官方推荐的实现方式是把它放在一个单独的 WebSocketMessageBrokerConfigurer 实现里，并标上 @Order(Ordered.HIGHEST_PRECEDENCE + 99)：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
@Configuration
@Order(Ordered.HIGHEST_PRECEDENCE + 99)
public class StompAuthConfig implements WebSocketMessageBrokerConfigurer {

    private final JwtService jwtService;

    public StompAuthConfig(JwtService jwtService) {
        this.jwtService = jwtService;
    }

    @Override
    public void configureClientInboundChannel(ChannelRegistration registration) {
        registration.interceptors(new ChannelInterceptor() {
            @Override
            public Message<?> preSend(Message<?> message, MessageChannel channel) {
                StompHeaderAccessor accessor =
                        MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);
                if (accessor == null) {
                    return message;
                }

                // 只在 CONNECT 帧做认证，认证结果写入会话
                // Spring 会记住这个用户，并关联到同一会话后续所有 STOMP 消息
                if (StompCommand.CONNECT.equals(accessor.getCommand())) {
                    String bearer = accessor.getFirstNativeHeader("Authorization");
                    if (bearer == null || !bearer.startsWith("Bearer ")) {
                        throw new MessagingException("缺少 Authorization 头");
                    }
                    Authentication auth = jwtService.parseAuthentication(bearer.substring(7));
                    if (auth == null) {
                        throw new MessagingException("token 无效或已过期");
                    }
                    // 关键：把用户绑定到这个 STOMP 会话
                    accessor.setUser(auth);
                }
                return message;
            }
        });
    }
}

而 Spring Security 侧手动组装组件的官方代码里，注册顺序是这样的

1
2
3
4
5
6
7
8
@Override
public void configureClientInboundChannel(ChannelRegistration registration) {
    AuthorizationChannelInterceptor authz = new AuthorizationChannelInterceptor(authorizationManager);
    AuthorizationEventPublisher publisher = new SpringAuthorizationEventPublisher(applicationContext);
    authz.setAuthorizationEventPublisher(publisher);
    // 注意顺序：先填充安全上下文，再做授权
    registration.interceptors(new SecurityContextChannelInterceptor(), authz);
}

SecurityContextChannelInterceptor 必须在 AuthorizationChannelInterceptor 之前，加上前面的 @Order(HIGHEST_PRECEDENCE + 99)，整体的执行顺序才是，自定义认证 → 填充安全上下文 → 消息授权。

CSRF：为什么 SockJS 特别麻烦

默认情况下，Spring Security 要求所有 CONNECT 消息都带 CSRF token。这确保了只有能够访问该 CSRF token 的站点才能建立连接。而由于只有同源站点才能访问到这个 CSRF token，外部域名就不被允许建立连接。

只有同源页面才能拿到 CSRF token，所以要求携带 CSRF token 就等价于强制同源。

但是 SockJS 不允许通过 HTTP 头或参数传 token，所以只能放在 STOMP 头里。

1
2
3
4
5
6
7
8
9
10
// 从 meta 标签或接口拿到 CSRF token
const headerName = document.querySelector('meta[name="_csrf_header"]').content;
const token = document.querySelector('meta[name="_csrf"]').content;

const headers = {};
headers[headerName] = token;

stompClient.connect(headers, () => {
  // 连接成功
});

对于纯静态 HTML（没有服务端渲染注入 token）的情况，文档建议暴露一个接口：

1
2
3
4
5
6
7
8
@RestController
public class CsrfController {

    @RequestMapping("/csrf")
    public CsrfToken csrf(CsrfToken token) {
        return token;
    }
}

前端先调 /csrf 拿到 token，再放进 STOMP 头。

如果想关掉 CSRF 校验，但是@EnableWebSocketSecurity 目前无法关闭 CSRF，要关只能手动组装组件（就是前面那段 AuthorizationChannelInterceptor + SecurityContextChannelInterceptor 的代码）。

SockJS 与 Spring Security 的额外配置

如果用了 SockJS，还有两处必须放宽，文档都给出了具体做法。

  • 帧选项。

    SockJS 的某些传输会使用 iframe，而 Spring Security 默认用 X-Frame-Options: DENY 阻止页面被嵌入（防点击劫持）。所以要允许同源嵌入

    1
    2
    http.headers(headers -> headers
            .frameOptions(HeadersConfigurer.FrameOptionsConfig::sameOrigin));

  • CSRF 的豁免范围

    SockJS 的 HTTP 传输会对 CONNECT 用 POST，需要在 Web 层对 STOMP 端点关闭 CSRF 保护，但只对这一个端点

    1
    2
    3
    http.csrf(csrf -> csrf
            // 只豁免 STOMP 端点，它是通过 STOMP 头保护的
            .ignoringRequestMatchers("/chat/**"));

SSE 的安全考量

SSE 的安全模型和普通 HTTP 接口基本一致，所以反而简单。

  • SSE 受 CORS 约束。

    因为 SSE 就是一个普通 GET 请求，浏览器的 CORS 机制正常生效。跨域时需要服务端返回正确的 Access-Control-Allow-Origin。EventSource 的 withCredentials: true 会让请求携带 Cookie，此时服务端不能用 Access-Control-Allow-Origin: *（CORS 规范禁止）。

  • token 不要放在 URL 里。

    这是 SSE 最常见的错误。因为EventSource 的构造函数只接受 URL，没法加自定义头（和原生 WebSocket 一样的问题）；于是很多人把 token 拼在 query string 里；但 URL 会被记录到 nginx access log、浏览器历史、Referer 头、APM 系统里，token 就泄漏了。

    正确做法有三种，按推荐顺序：

    1. 用 Cookie 认证（如果架构允许）：浏览器自动携带，URL 干净；

    2. 用一个短期的、一次性的 stream ticket：先通过正常接口（带 Authorization 头）换一个只能用一次的 ticket，然后用 ticket 建立 SSE 连接。ticket 有效期设得很短（比如 30 秒），即使泄漏了危害也有限；

      1
      2
      3
      4
      5
      6
      7
      8
      9
      10
      11
      12
      13
      14
      15
      16
      17
      18
      19
      /** 第一步：用标准认证换一次性 ticket */
      @PostMapping("/api/stream/ticket")
      public TicketVO issueTicket(Principal principal) {
          // 生成一个随机 ticket，绑定 userId，60 秒过期，存 Redis
          String ticket = UUID.randomUUID().toString();
          redis.opsForValue().set("stream:ticket:" + ticket, principal.getName(), Duration.ofSeconds(60));
          return new TicketVO(ticket);
      }

      /** 第二步：用 ticket 建立 SSE 连接，ticket 用完立即删除 */
      @GetMapping(path = "/api/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
      public SseEmitter stream(@RequestParam String ticket) {
          // 原子地取出并删除，保证只能用一次
          String username = redis.opsForValue().getAndDelete("stream:ticket:" + ticket);
          if (username == null) {
              throw new UnauthorizedException("ticket 无效或已过期");
          }
          return ssePushService.subscribe("user:" + username, null);
      }

    3. 如果非要把 token 放 URL，至少确保访问日志做了脱敏。

  • SSE 连接要有认证续期机制。

    SSE 连接可能保持几十分钟，而 token 可能 15 分钟就过期了。由于 SSE 连接建立后不再有”重新鉴权”的机会，你需要：

    • 让服务端定期检查会话有效性（比如每 5 分钟检查一次），失效就 complete() 掉连接；
    • 或者让前端在 token 即将过期时主动重连（关闭旧连接、用新 token 建新连接）。

  • 注意 Last-Event-ID 越权。

    如果你的 SSE 事件 ID 是全局递增的，那么客户端可以伪造一个很小的 Last-Event-ID 来要求服务端补发历史消息。必须校验这个用户是否有权访问这些历史事件。 正确的设计是，Last-Event-ID 只在”该用户自己的消息流”里做游标，服务端补发时要用 userId 作为过滤条件，而不是只按 ID 范围查。

安全清单

最后把这一部分的要点整理成一份可直接对照检查的清单：

检查项 WebSocket SSE
Origin 校验 必须，默认同源，不要设 * 由 CORS 机制保证
握手认证 必需（Cookie 或握手参数 token） 必需（Cookie 或 ticket）
消息级授权 必需（AuthorizationManager） 不适用（每条消息是独立请求）
区分 SUBSCRIBE / MESSAGE 必需，禁止客户端往 broker 前缀发消息 不适用
禁止订阅转换后的真实目的地 必需（禁 /queue/** 订阅） 不适用
CSRF 需要（默认 CONNECT 校验） 由 Spring Security 的 CSRF 机制覆盖
token 不能放 URL 无法避免（原生 API 限制），要走 STOMP 头 可以用 ticket 方案避免
连接数限制 必需（防单用户开大量连接） 必需（否则触发每域名 6 连接限制）
消息体大小限制 必需（messageSizeLimit / bufferSize） 由业务控制
慢客户端保护 必需（sendBufferSizeLimit） 必需（超时 + 心跳 + 失败清理）
历史数据越权 不适用 必需（Last-Event-ID 要按 userId 过滤）

企业级落地挑战

前面把”怎么实现”讲完了。这一部分讲”怎么上线”——这部分内容在教程里基本看不到，但它是决定项目成败的地方。

网关配置

本地开发时一切正常，部署到带 nginx 的环境后就各种奇怪现象，这几乎是每个实时推送项目的必经之路。

WebSocket：必须显式转发两个逐跳头部

nginx 官方文档专门有一页讲 WebSocket proxying，大意是：Upgrade 是逐跳（hop-by-hop）头部，nginx 默认不会把它从客户端传给后端。 同样被默认丢弃的还有 Connection。如果这两个头传不过去，后端就看不到 客户端想升级协议 这个意图，握手会直接失败或者退化成普通 HTTP 请求。

从 1.3.13 版本起，如果被代理的服务器返回了 101（Switching Protocols）状态码，并且客户端在请求里通过 Upgrade 头要求了协议切换，那么 nginx 会启用一种特殊的运行模式，在客户端和被代理服务器之间建立一条隧道。

nginx 官方给的配置是这样的：

1
2
3
4
5
6
location /chat/ {
    proxy_pass http://backend;
    # proxy_http_version 1.1; # before version 1.29.7
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";
}
  • proxy_http_version ：它的默认值从 1.29.7 起变成了 1.1；在 1.29.7 之前，默认值用的是 1.0。

也就是说：

  • nginx 1.29.7 及以上：默认已经是 HTTP/1.1，proxy_http_version 1.1; 可以不写；
  • nginx 1.29.7 以下：默认是 HTTP/1.0，必须显式写 proxy_http_version 1.1;，因为 HTTP/1.0 不支持协议升级语义。

建议是不管什么版本都写上这一行。 它在新版本上没有副作用，却能让配置在任意版本上都正确

官方还给了一个更讲究的写法，用 map 让 Connection 的值根据是否有 Upgrade 头动态决定：

1
2
3
4
5
6
7
8
9
10
11
12
13
map $http_upgrade $connection_upgrade {
    default upgrade;
    ''      close;
}

server {
    location /chat/ {
        proxy_pass http://backend;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection $connection_upgrade;
    }
}

这样同一个 location 既能处理 WebSocket 升级，有 Upgrade 头时 Connection: upgrade，也能处理普通 HTTP 请求，没有时 Connection: close。

nginx 默认 60 秒内没有任何数据传输就关闭连接。这就是为什么”连接总是一分钟左右断一次”成了 WebSocket 部署的经典问题。官方给了两个解决方向：调大 proxy_read_timeout，或者让服务端定期发 Ping 帧来重置超时。

一般是两个都做：把 proxy_read_timeout 调到 300-600 秒，同时服务端按 10-30 秒间隔发心跳。只调大超时不解决”连接已经死了但双方都不知道”的问题；只发心跳则在网络真正卡住时会被提前掐掉。

SSE 的问题在缓冲，不在升级

SSE 不需要转发任何头部，它就是一个普通 GET，所以配置反而更简单，但有一个独特的坑：缓冲。

proxy_buffering 的默认值是 on。

开启缓冲时，nginx 会 尽快把响应从后端读出来 并攒在缓冲区里；关闭时才会 收到什么立刻转发什么。

你辛辛苦苦写的事件流，被 nginx 攒在缓冲区里，攒够一批才发给客户端。用户看到的现象是”消息延迟几秒到几十秒才出现”或者”几条消息同时弹出来”。

两种关闭方式：

  • 改 nginx 配置

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    location /api/sse/ {
        proxy_pass http://backend;
        proxy_buffering off;
        proxy_cache off;
        proxy_http_version 1.1;
        # 读超时只作用于"两次连续读操作之间"，所以只要心跳够频繁就不会超时
        proxy_read_timeout 3600s;
        proxy_send_timeout 3600s;
        # 确保响应不被压缩（压缩会引入缓冲）
        gzip off;
    }

  • 应用层返回响应头（推荐）

    缓冲也可以通过 X-Accel-Buffering 响应头字段传入 yes 或 no 来开关；这个能力可以用 proxy_ignore_headers 指令禁用。所以服务端只要在 SSE 响应里加一个头就行：

    1
    response.setHeader("X-Accel-Buffering", "no");

方式二的优点是不用动运维配置，而且粒度精确，只有 SSE 接口关缓冲，其他接口保持不变。对于”公司网关统一管控、改配置要走审批”的场景，这是唯一可行的方案。

关于 proxy_read_timeout，它定义的是从被代理服务器读取响应的超时。这个超时只设置在两次连续读操作之间，而不是整个响应的传输过程。如果被代理服务器在这段时间内没有传输任何东西，连接就会被关闭。这个超时只作用在”两次连续读操作之间”，而不是整个响应的总时长。

这句话非常关键，它说明了为什么 SSE 只要心跳足够频繁，就不需要把超时设得特别大。每一次心跳都是一次读操作，都会重置计时器。默认 60 秒配合 15 秒心跳已经完全够用了。

反过来说，如果你的 SSE 没有心跳，那不管 proxy_read_timeout 设得多大，连接都撑不住，因为 60 秒（或你设的值）内没有任何数据传输，连接就会被关闭。这和 WebSocket 的情形完全一样。

一个完整的对照配置

把 WebSocket 和 SSE 的网关配置并排放在一起，方便对照：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
# 针对 WebSocket 升级的 Connection 值动态映射
map $http_upgrade $connection_upgrade {
    default upgrade;
    ''      close;
}

server {
    listen 443 ssl;
    http2 on;
    server_name app.example.com;

    # ---------- WebSocket ----------
    location /ws/ {
        proxy_pass http://backend;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection $connection_upgrade;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        # 连接级超时：服务端有心跳，这里给足余量
        proxy_read_timeout 600s;
        proxy_send_timeout 600s;
        # WebSocket 不需要缓冲
        proxy_buffering off;
    }

    # ---------- SSE ----------
    location /api/sse/ {
        proxy_pass http://backend;
        proxy_http_version 1.1;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        # SSE 的核心：关闭缓冲
        proxy_buffering off;
        proxy_cache off;
        # 心跳间隔 15s，60s 的超时绰绰有余；这里给足余量
        proxy_read_timeout 3600s;
        # 应用层返回 X-Accel-Buffering: no 也可以达到同样效果
    }
}

注意两个补充建议。

  • proxy_set_header Host $host 不要漏。 官方说明 proxy_set_header 的默认值是 Host $proxy_host 和 Connection close，而且原始请求的 Host 和 Connection 头默认不会传给被代理服务器。如果你的后端按 Host 做路由或签名校验，就需要显式传递。
  • proxy_set_header 的继承规则是”只有当前层没有任何 proxy_set_header 时才会继承上层”，这些指令只有在当前层级没有定义任何 proxy_set_header 指令时，才会从上一层配置继承。所以如果你在 http 层配了通用的 X-Forwarded-For，在 location 层又写了一条 proxy_set_header，那 http 层的那条就不会生效了。这是一个非常隐蔽的配置陷阱。

多实例部署：会话粘不住怎么办

单机部署时一切正常，一旦水平扩容到多实例，实时推送就会出一个诡异的现象：消息只推到了恰好连在同一台机器上的那部分用户。

原因很直接，消息的生产者（比如处理支付回调的实例 A）和消费者（比如连在实例 B 上的用户）不在同一台机器上，而实例 A 的 SimpMessagingTemplate 只能找到自己内存里维护的订阅关系。

Spring 官方认为。简单 broker 很适合入门，但它只支持 STOMP 命令的一个子集（它不支持 ack、receipt 以及其他一些特性），依赖一个简单的消息发送循环，而且不适合做集群。

另一处是：simple broker 无法用于多实例。用上完整功能的消息代理后，每个应用实例都连到 broker 上，一个实例广播的消息就能通过 broker 传到其他实例的客户端。

有四种常见解法，我按推荐度排序。

方案一：STOMP broker relay（官方方案）

把 simple broker 换成外部 STOMP broker（RabbitMQ、ActiveMQ 等）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
@Override
public void configureMessageBroker(MessageBrokerRegistry registry) {
    registry.enableStompBrokerRelay("/topic", "/queue")
            .setRelayHost("rabbitmq.internal")
            .setRelayPort(61613)
            .setClientLogin("app")
            .setClientPasscode("******")
            .setSystemLogin("system")
            .setSystemPasscode("******")
            // 多实例场景下的关键配置：用户目的地广播
            .setUserDestinationBroadcast("/topic/unresolved.user.dest")
            .setUserRegistryBroadcast("/topic/registry.broadcast");

    registry.setApplicationDestinationPrefixes("/app");
    registry.setUserDestinationPrefix("/user");
}

STOMP broker relay 就是一个 MessageHandler，它通过把消息转发给外部消息 broker 来处理消息；为了做到这一点，它会与 broker 建立 TCP 连接，把所有消息转发给它，然后再把从它那里收到的所有消息，通过各客户端的 WebSocket 会话转发下去。本质上，它扮演的是一个双向转发的”中继”角色。

broker relay 本身就是一个 MessageHandler，它和后端建立 TCP 连接，把所有消息转发给 broker，再把从 broker 收到的消息通过 WebSocket 会话转发给客户端。本质上它就是一个双向转发的”中继器”。

关键是两个 broadcast 配置。在多应用服务器的场景下，一个用户目的地可能无法被解析，因为这个用户连在另一台服务器上。在这种情况下，你可以配置一个目的地，把无法解析的消息广播出去，让其他服务器有机会尝试解析。这可以通过 Java 配置里的 userDestinationBroadcast 属性、或者 XML 里 message-broker 元素的 user-destination-broadcast 属性来设置。配置 userDestinationBroadcast 后，无法解析的消息会被广播出去，让其他实例有机会处理。

关于 userRegistryBroadcast，把本地用户注册表的内容广播出去，并监听来自其他服务器的这类广播。在多应用服务器场景下，这使得每个服务器的用户注册表都能感知到连接在其他服务器上的用户。

这两个属性都挂在 StompBrokerRelayRegistration 上，而且默认都是不设置的。其中 userDestinationBroadcast 在 用户私有消息（/user/**）+ 多实例 这个组合场景下是必须配置的，否则会出现有时能收到私信、有时收不到的随机故障。排查这类问题极其痛苦，因为它是概率性的。

broker relay 需要 reactor-netty 和 netty-all 来做 TCP 连接管理，这两个依赖容易漏。为了做 TCP 连接管理，需要在你的项目里加上 io.projectreactor.netty:reactor-netty 和 io.netty:netty-all 这两个依赖。

关于消息大小还有一个细节值得注意：STOMP broker relay 走的是 TCP，不受 WebSocket 帧大小的限制，所以大消息的问题会小一些。但 broker 本身可能有自己的大小限制，要去查对应 broker 的文档。

最后提醒一下外部 broker 的队列清理问题。用户会话结束后，那些为每个用户动态创建的队列必须被清理掉，否则队列会无限增长。这是接外部 broker 时最容易忽略的运维问题。

方案二：Redis Pub/Sub 或 Stream 做广播

如果不想引入 STOMP broker（比如团队已经有一套 Redis 基础设施），可以用 Redis 做跨实例的消息中转：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
@Service
public class ClusterBroadcastService {

    private final StringRedisTemplate redisTemplate;
    private final SimpMessagingTemplate messagingTemplate;

    /** 本实例要推送时，先发到 Redis 频道，让所有实例都能收到 */
    public void broadcast(String topic, Object payload) {
        String json = JsonUtils.toJson(payload);
        redisTemplate.convertAndSend("ws:broadcast:" + topic, json);
    }

    /** 所有实例都订阅 Redis 频道；收到后判断本地有没有订阅者，有就推 */
    @Bean
    public RedisMessageListenerContainer listenerContainer(
            RedisConnectionFactory factory, SimpMessagingTemplate template) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(factory);
        container.addMessageListener((message, pattern) -> {
            String channel = new String(message.getChannel(), StandardCharsets.UTF_8);
            String topic = channel.substring("ws:broadcast:".length());
            String payload = new String(message.getBody(), StandardCharsets.UTF_8);
            // 推给本实例上订阅了该 topic 的客户端
            template.convertAndSend("/topic/" + topic, payload);
        }, new PatternTopic("ws:broadcast:*"));
        return container;
    }
}

要特别注意 Pub/Sub 的丢消息问题：如果实例 B 在重启，恰好这期间有消息发到 Redis，B 就永远收不到了。所以这个方案适合”消息可丢、重连后能补齐”的场景（通知、状态刷新），不适合”每条消息都必须送达”的场景（订单状态变更）。

方案三：网关会话粘滞

用一致性哈希或 IP 哈希把同一个用户的连接都路由到同一个实例：

1
2
3
4
5
upstream ws_backend {
    ip_hash;                       # 按客户端 IP 哈希
    server 10.0.0.1:8080;
    server 10.0.0.2:8080;
}

但我不推荐这个方案，原因有三个：

  1. ip_hash 在客户端 IP 变化（移动网络切换、多级代理）时就会失效，用户连接被路由到别的实例；
  2. 扩容/缩容时哈希环变化，大量用户会被重新分配，导致连接风暴；
  3. 它只解决了 同一用户连在同一实例，没解决 业务事件在另一实例产生。支付回调可能落在实例 A，而这个用户连在实例 B，消息照样推不过去。

粘滞只能作为辅助手段，不能作为多实例推送的主方案。 真正的问题不在连接路由，而在消息如何跨实例传递。

方案四：消息队列 / 事件总线

如果系统本来就有 Kafka、RocketMQ 之类的消息队列，直接用它们做跨实例广播是最自然的。所有实例订阅同一个 topic，收到消息后各自推给自己本地的 WebSocket 会话。

这个方案的优点是基础设施已经存在，不增加新组件；缺点是需要自己实现”消息到会话”的路由逻辑，而且要注意消费组的选择，必须是”广播模式”（每个实例都收到），而不是”集群模式”（只有一个实例收到）。

在 Kafka 里对应的概念是每个实例用不同的 consumer group；在 RocketMQ 里是 BROADCASTING 模式。这个区别很关键，配错了就会变成”只有一个实例收到消息”，表现和没配一样。

四种方案对比

方案 跨实例投递 可靠性 引入组件 实现复杂度 适用场景
STOMP broker relay 支持 高（broker 保证） RabbitMQ/ActiveMQ 低（Spring 内建） 企业级首选
Redis Pub/Sub 支持 低（会丢） 无（已有 Redis） 中 消息可丢的场景
Redis Stream 支持 中高（可 ACK） 无（已有 Redis） 中高 需要可靠又不引新组件
网关会话粘滞 部分支持 — 无 低 仅作辅助
MQ 广播模式 支持 高 Kafka/RocketMQ 中 已有 MQ 基础设施

监控与可观测性

实时推送的监控，难点在于常规的 HTTP 监控指标在这里是失效的：一个保持了 30 分钟的 WebSocket 连接，在 APM 里可能只显示为一个耗时 30 分钟的请求，看不出任何有用信息。

WebSocket 侧要采集的指标

STOMP 路线的最大优势之一，就是 Spring 已经内建了统计能力。前面讲过 WebSocketMessageBrokerStats，这里给出落地成 Micrometer 指标的写法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
@Component
public class WebSocketMetricsExporter {

    private final WebSocketMessageBrokerStats stats;
    private final MeterRegistry meterRegistry;

    public WebSocketMetricsExporter(WebSocketMessageBrokerStats stats, MeterRegistry meterRegistry) {
        this.stats = stats;
        this.meterRegistry = meterRegistry;
        // 把 stats 的关键值注册成 Gauge，Micrometer 会定期抓取
        Gauge.builder("ws.sessions.current", stats, s -> parseCurrentSessions(s.getWebSocketSessionStatsInfo()))
                .description("当前活跃的 WebSocket 会话数")
                .register(meterRegistry);
    }

    /** 也可以自己维护计数器，粒度更细、更可靠 */
    private final AtomicInteger activeConnections = new AtomicInteger();

    public void onConnect() {
        meterRegistry.counter("ws.connect.total").increment();
        meterRegistry.gauge("ws.connections.active", activeConnections.incrementAndGet());
    }

    public void onClose(int closeCode) {
        // 按关闭码分类打点是排查断连的第一手段
        meterRegistry.counter("ws.disconnect.total", "code", String.valueOf(closeCode)).increment();
        activeConnections.decrementAndGet();
    }

    public void onMessageIn() {
        meterRegistry.counter("ws.message.inbound").increment();
    }

    public void onMessageOut() {
        meterRegistry.counter("ws.message.outbound").increment();
    }
}

五个指标：

指标 含义 异常时的排查方向
ws.connections.active 当前活跃连接数 突然归零说明大面积断连；持续上涨要查是否是连接泄漏
ws.disconnect.total{code="1006"} 异常断开次数 突增说明网络或代理在掐连接
ws.disconnect.total{code="1011"} 服务端错误断开 突增说明业务代码在抛异常
Connect Failures 会话建立后 60 秒内没收到任何消息就关闭 突增说明代理或网关配置有问题
Send Limit Exceeded 超过发送超时或缓冲上限被关闭 突增说明有大量慢客户端（弱网用户）

后两个可以直接从 WebSocketMessageBrokerStats 里取。Spring 文档对它们的定义，大意是：

  • Connect Failures：那些已经建立、但在 60 秒内没有收到任何消息就被关闭的会话。这通常是代理或网络问题的征兆。
  • Send Limit Exceeded：那些因为超过配置的发送超时或发送缓冲区限制而被关闭的会话，这种情况可能发生在慢客户端身上。

这两个计数器是 Spring 帮我们做好的诊断，直接接出来告警就能省掉大量排查时间。60 秒这个数字也正好对应 timeToFirstMessage 的默认值，所以它是可调的，如果你的客户端建立连接后确实要等很久才发第一条消息，记得把 timeToFirstMessage 调大，否则会误杀。

SSE 侧要采集的指标

SSE 没有内建统计，需要自己维护：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
@Component
public class SseMetrics {

    private final AtomicInteger activeEmitters = new AtomicInteger();
    private final AtomicLong totalEvents = new AtomicLong();

    private final MeterRegistry meterRegistry;

    public SseMetrics(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
        // 关键：活跃 emitter 数量直接反映内存占用风险
        Gauge.builder("sse.emitters.active", activeEmitters, AtomicInteger::get)
                .description("当前活跃的 SSE emitter 数量")
                .register(meterRegistry);
        Gauge.builder("sse.events.total", totalEvents, AtomicLong::get)
                .description("累计推送事件数")
                .register(meterRegistry);
    }

    public void onSubscribe() {
        activeEmitters.incrementAndGet();
    }

    public void onComplete() {
        activeEmitters.decrementAndGet();
    }

    public void onEventSent() {
        totalEvents.incrementAndGet();
    }
}

sse.emitters.active 是 SSE 最重要的指标，因为：

  • 它直接反映服务端的内存占用（每个 emitter 都持有连接和缓冲区）；
  • 如果它只涨不跌，说明连接泄漏了，这是 SSE 最典型的事故形态；
  • 把它和预期的在线用户数对比，如果明显偏高，说明有大量僵尸连接没被清理。

心跳是最好的健康探针。可以在每次心跳推送时，统计成功发送数 / 尝试发送数：

  • 如果这个比率下降，说明有越来越多的连接处于”看起来还在、实际已经死了”的状态；
  • 这也是发现客户端静默断网（比如用户直接拔网线）的唯一手段。

限流与容量规划

实时推送的连接数和消息量都容易失控，必须提前规划。

连接数维度的限流，三个层次都要限：

1
2
3
4
5
6
7
8
9
10
11
12
@Configuration
public class WebSocketLimitConfig {

    /** 单用户最大连接数（多标签页 + 多设备） */
    private static final int MAX_PER_USER = 5;

    /** 单 IP 最大连接数（防单机脚本刷连接） */
    private static final int MAX_PER_IP = 50;

    /** 单实例最大连接数（保护自身） */
    private static final int MAX_PER_INSTANCE = 20_000;
}

三个层次的作用不同：

  • 单用户限制防的是”正常用户的异常行为”（开了 20 个标签页、或者前端代码 bug 导致重复建连）；
  • 单 IP 限制防的是”单机攻击”（一个脚本从一台机器发起上万连接）。注意这个值不能设太小，因为企业出口 NAT 后面可能有几百个真实用户共享一个 IP；
  • 单实例限制是最重要的保护，防止实例被打满导致所有用户都受影响。达到上限后应该拒绝新连接而不是让内存耗尽。

而且对 服务端往客户端推 这个方向也要限流，因为客户端可以故意不读数据来制造积压：

  • 单连接发送速率限制：比如每连接每秒最多 50 条消息，超出的丢弃或降采样；
  • 单连接缓冲上限：STOMP 用 sendBufferSizeLimit，SSE 要自己统计未确认的字节数；
  • 消息大小上限：STOMP 用 setMessageSizeLimit，SSE 要在序列化后检查长度。

一个实用的做法是降采样而不是丢弃，对于高频的行情类推送，超过阈值时把多条消息合并成一条 批量更新 再发，客户端体验反而更好。

容量方面，我给出在项目里用的估算表，供参考：

资源 单个 STOMP 连接 单个 SSE 连接
会话对象 + 缓冲区 约 20-50 KB 约 10-30 KB
写入线程 不独占（线程池复用） MVC 下写入时占用线程
心跳开销 每 10s 一个空行（约 2 字节） 每 15s 一个注释行（约 15 字节）
1 万连接内存 约 200-500 MB 约 100-300 MB

这几个数字的量级比很多人想的要大。因为这样 1 万连接不是”1 万个轻量对象”，而是几百 MB 的堆内存。 所以：

  • 单实例 2 万连接已经需要认真调 JVM 堆大小和 GC 参数了；
  • 如果要支撑 10 万级以上连接，必须用 WebFlux 那种非阻塞模型，或者横向扩到多实例再配合 broker。

技术选型总结

把整篇文章的结论收敛成一张决策表：

业务场景 推荐方案 理由
站内通知、公告、系统消息 SSE 单向、低频、需要可靠重连与断点续传
订单/任务状态跟踪 SSE 状态变更次数少，SSE 的自动重连正好匹配
大屏看板、监控指标 SSE（HTTP/2 下）或 WebSocket 高频时 WebSocket 帧开销更低
实时聊天、IM WebSocket + STOMP 双向、需要用户寻址与订阅管理
协同编辑、白板 WebSocket + STOMP 高频双向，延迟敏感
在线游戏、实时互动 WebSocket（原生） 需要完全自定义协议与最低延迟
行情推送（高频） WebSocket + STOMP 高频大流量，需二进制与压缩支持
进度条、长任务反馈 SSE 单向、可断点续传、实现简单
需要向特定用户精准推送 WebSocket + STOMP /user/** 目的地是内建能力
团队网关改不动、只允许标准 HTTP SSE 就是一个 GET，无需改网关升级配置

如果只能记一句话，我会这么说：

先问自己”客户端要不要频繁上行”。不要，就优先 SSE，因为它便宜、可靠、几乎不需要改基础设施；要，就上 WebSocket + STOMP，并且从第一天就把安全、监控和多实例广播这三件事规划进去，因为它们都不是可以事后补的东西。

参考资料

这篇文章的协议层结论和 Spring 侧 API 全部来自下面这些一手来源，我在文中也逐处标注了引用位置，方便你核对原文。

协议规范

Spring

容器与网关

浏览器实现

关于版本基线：本文以 Spring Boot 4.x（Spring Framework 7.x、Spring Security 7.x、Jakarta EE 11 / Servlet 6.1、Tomcat 11）为准。若你的项目还在 Spring Boot 2.x/3.x 上，需要留意的差异主要是：包名 javax.* 与 jakarta.*、Spring Security 的 WebSocket 授权 API（旧版用 AbstractSecurityWebSocketMessageBrokerConfigurer，已在 Spring Security 7 中移除）、以及 setAllowedOrigins 与 setAllowedOriginPatterns 的推荐用法。