本节目标：让 WebSocket 连接可检测、可关闭、可广播，并让多线程调用下的帧不会交错。
适用版本：JDK 25（Temurin 25.0.5+8）；只使用java.nio、java.util.concurrent与 JDK 标准库。
承接的问题
11.2 的 WebSocketFrameCodec 已经能把字节流切成帧，/ws/echo 也能把一条消息原样回给发送者。但一个只能回显的连接还缺三样东西：
- 客户端拔网线、进程被杀、NAT 表项过期时，TCP 不会立刻通知服务端；连接处于半开状态，文件描述符和会话对象一直占着。
- 客户端主动关闭时，如果服务端不回 close 帧，浏览器会一直等；反过来，服务端关闭时也要让客户端知道原因。
- 1000 个连接、多个业务线程同时发消息时，两个线程各自往同一个
SocketChannel写半帧，客户端收到的就是交错字节。
本节按「活着的检测、体面的结束、并发的分发」三件事，把 WebSocketSessionImpl 和 WebSocketHub 补齐。
核心机制一：ping/pong 与空闲检测
WebSocket 有两个控制帧专门做心跳：ping（opcode 9）和 pong（opcode 10）。收到 ping 必须回一个 pong，并把 ping 的载荷原样带回；pong 不需要再确认。控制帧载荷不超过 125 字节，所以 ping 里塞不下太多业务数据。
服务端通常这样用：
- 每隔
pingInterval发一个 ping，收到 pong 或任何入站帧就更新活跃时间； - 如果超过
idleTimeout没有任何入站帧，就以 1001 关闭会话。
关键点是更新「活跃时间」的时机。只在 pong 分支更新会误杀一种客户端：它持续发文本消息，但从不主动回 pong（有些手写客户端会漏）。更稳妥的做法是任何入站帧都算活跃，ping/pong 只是用来在没有业务流量时制造入站帧。
void handle(WebSocketFrame frame) {
if (state.get() == State.CLOSED) {
return;
}
lastSeenNanos = System.nanoTime();
switch (frame.opcode()) {
case WebSocketFrame.OP_PING -> enqueueControl(
new WebSocketFrame(true, WebSocketFrame.OP_PONG, frame.payload()));
case WebSocketFrame.OP_PONG -> {
}
case WebSocketFrame.OP_CLOSE -> handleClose(frame);
default -> {
WebSocketFrame message = assembler.accept(frame);
if (message != null && onMessage != null) {
onMessage.accept(message);
}
}
}
}
本机把一个带 hb 载荷的 ping 喂给会话，抓到的回包是：
ping -> opcode=0xa payload=hb
0xa 就是 pong，载荷与 ping 完全一致。回显载荷不是可选项：RFC 6455 要求 pong 的 application data 与 ping 相同，客户端可能用载荷里的序号匹配请求和响应。
心跳定时器可以放在 ScheduledExecutorService，也可以放在事件循环的定时任务里。无论放哪里，定时线程都只能调用 sendPing() 这类「入队」方法，不能直接 channel.write。否则定时线程和事件循环线程会同时写同一条连接，帧边界立刻乱掉。sendPing() 只做一次状态检查和入队：
void sendPing() {
if (state.get() == State.OPEN) {
enqueueControl(new WebSocketFrame(true, WebSocketFrame.OP_PING, new byte[0]));
}
}
TCP 自带的 keepalive 不能替代应用层心跳。macOS 上默认要空闲两小时才开始探测，而且它只能证明内核还愿意回 ACK，不能证明 WebSocket 应用还在处理消息。心跳间隔要大于正常 RTT，又小于业务能容忍的发现故障时间；linkd 的示例值是 30 秒 ping、90 秒超时，这两个值不是协议规定，要按部署环境调。
核心机制二：关闭码与关闭握手
close 帧（opcode 8）的载荷是「2 字节关闭码 + UTF-8 原因」，总长不超过 125 字节，所以原因最多 123 字节。载荷可以为空，表示没有状态码；长度为 1 是非法的，因为放不下完整的 2 字节码。
常用关闭码：
| 码 | 含义 | 能否上线 |
|---|---|---|
| 1000 | 正常关闭 | 能 |
| 1001 | 端点离开，如服务重启 | 能 |
| 1002 | 协议错误 | 能 |
| 1003 | 收到不支持的数据类型 | 能 |
| 1007 | 载荷不是合法 UTF-8 | 能 |
| 1008 | 违反策略 | 能 |
| 1009 | 消息过大 | 能 |
| 1011 | 服务端内部错误 | 能 |
| 1012–1014 | 服务重启、过载、网关错误 | 能 |
| 1005/1006/1015 | 本地状态，不能出现在线上 | 不能 |
| 3000–4999 | 应用或私有用途 | 能 |
1005、1006、1015 是本地状态，不是线上值。尤其 1006：它表示「没有收到 close 帧就断了」，由实现自己记录，绝不能出现在 close 帧里。把 1006 发出去，对端会以 1002 关闭。
关闭握手分主动和被动两种：主动方发 close，进入 CLOSING，不再发数据帧，收到对端 close 后关闭 TCP；被动方收到 close，回一个 close（通常回同样的码），然后关闭 TCP。服务端可以在发完 close 后立即关 TCP，客户端通常等服务端先关，避免自己的 close 还没发完连接就没了。
WebSocketSessionImpl 用 AtomicReference<State> 让关闭只执行一次：
@Override
public void close(int code, String reason) {
beginClose(closePayload(code, reason));
}
private void beginClose(byte[] payload) {
if (!state.compareAndSet(State.OPEN, State.CLOSING)) {
return;
}
enqueueControl(new WebSocketFrame(true, WebSocketFrame.OP_CLOSE, payload));
}
private void handleClose(WebSocketFrame frame) {
if (frame.payload().length == 0) {
beginClose(new byte[0]);
return;
}
int code = ((frame.payload()[0] & 0xFF) << 8) | (frame.payload()[1] & 0xFF);
beginClose(closePayload(code, ""));
}
beginClose 的 CAS 让并发的两个 close() 只有一个能入队 close 帧；第二个直接返回，不会出现两个 close 帧。被动关闭时回显对端的码，但原因留空，这是允许的：原因只是给人看的调试信息，不参与状态机。
关闭码和原因长度都要在发送前校验，不能等到对端断开才发现。本机验证三组边界：
static byte[] closePayload(int code, String reason) {
if (!validCloseCode(code)) {
throw new IllegalArgumentException("close code not allowed on the wire: " + code);
}
byte[] text = reason.getBytes(StandardCharsets.UTF_8);
if (text.length > 123) {
throw new IllegalArgumentException("close reason too long: " + text.length);
}
ByteBuffer payload = ByteBuffer.allocate(2 + text.length);
payload.putShort((short) code).put(text);
return payload.array();
}
close code=1000 reason=bye tcpClosed=true open=false
code 1006: close code not allowed on the wire: 1006
long reason: close reason too long: 124
第一组是正常关闭：close 帧的码是 1000、原因是 bye，随后 TCP 关闭，会话不再是 open。第二组说明 1006 被拦在发送前。第三组用 124 字节原因，加上 2 字节码就是 126 字节，超过 125 上限，所以必须在入队前拒绝，而不是让编码器写一个超长控制帧。
关闭之后不能再发数据帧。sendText 和 sendBinary 都会先检查状态，非 OPEN 直接丢弃；已经排队但还没写出的数据帧，在进入 CLOSING 后也没有意义，可以在溢出或关闭时清掉。协议不允许「关闭之后再补一条消息」，对端会以 1002 处理。
核心机制三：会话表与广播的线程安全
广播的朴素写法是维护一个 List<WebSocketSession>，收到消息就遍历调用 sendText。单线程下没问题，多线程下有两个坑：
- 帧交错：两个线程同时调用同一个会话的
sendText，各自把帧写进SocketChannel，内核可能把两次写拼成帧头A + 帧头B + 载荷A + 载荷B。接收方按长度切，第二帧的头就变成了载荷的一部分。 - 遍历时修改：广播线程遍历
HashMap，另一个线程加入或移除会话，直接抛ConcurrentModificationException；即使用ConcurrentHashMap，遍历期间被移除的会话也可能被访问到。
解决办法是把「谁写 channel」收窄到一个线程：每个会话有自己的有界发送队列，sendText 只入队，真正的写出由会话所属的 EventLoop 串行执行。会话表用 ConcurrentHashMap，广播用弱一致迭代器，删除用 remove(key, value) 避免误删同 id 的新会话。本机用两个假连接验证广播和关闭：
package com.plume.linkd.ws;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public final class WebSocketHub implements AutoCloseable {
private final Map<String, WebSocketSessionImpl> sessions = new ConcurrentHashMap<>();
private final ScheduledExecutorService heartbeat;
private final long idleTimeoutMillis;
public WebSocketHub(long pingIntervalMillis, long idleTimeoutMillis) {
this.idleTimeoutMillis = idleTimeoutMillis;
this.heartbeat = Executors.newSingleThreadScheduledExecutor(r -> {
Thread thread = new Thread(r, "linkd-ws-heartbeat");
thread.setDaemon(true);
return thread;
});
this.heartbeat.scheduleAtFixedRate(this::pingAll,
pingIntervalMillis, pingIntervalMillis, TimeUnit.MILLISECONDS);
}
public void add(WebSocketSessionImpl session) {
sessions.put(session.id(), session);
}
public void remove(String id) {
WebSocketSessionImpl session = sessions.remove(id);
if (session != null) {
session.close(1001, "session removed");
}
}
public void broadcastText(String text) {
for (WebSocketSessionImpl session : sessions.values()) {
if (!session.isOpen()) {
sessions.remove(session.id(), session);
continue;
}
session.sendText(text);
}
}
private void pingAll() {
long now = System.nanoTime();
for (WebSocketSessionImpl session : sessions.values()) {
if (!session.isOpen()) {
sessions.remove(session.id(), session);
continue;
}
if (session.idleMillis(now) > idleTimeoutMillis) {
session.close(1001, "heartbeat timeout");
sessions.remove(session.id(), session);
} else {
session.sendPing();
}
}
}
@Override
public void close() {
heartbeat.shutdownNow();
for (WebSocketSessionImpl session : sessions.values()) {
session.close(1001, "server shutdown");
}
sessions.clear();
}
}
broadcast size=2 a=all b=all
after hub.close a.open=false b.open=false tcpClosed=true/true
broadcastText 没有加锁，也没有在遍历时调用会阻塞的 write。sendText 对每个会话只是入队，慢会话不会拖住快会话；如果某个会话的队列满了，它自己会以 1013 关闭，广播循环继续。WebSocketHub.close() 先停心跳线程，再逐个发 1001，最后清表；顺序反了会让心跳线程在清表后又 ping 一批已关闭的会话。
ConcurrentHashMap 的迭代器是弱一致的：遍历时可能看到已经删除的会话，也可能看不到刚加入的会话。这对广播是可接受的语义——广播本来就是「尽力而为」，不承诺每个连接都收到。如果业务要求可靠投递，需要消息 id、确认和重试，那是另一个层次的问题，不能靠 ConcurrentHashMap 解决。
在 linkd 里落地：WebSocketSessionImpl 的发送队列
前面已经给了 ping/pong 和关闭方法，这里补会话的状态、队列和 flush。整个类的骨架是：
package com.plume.linkd.ws;
import com.plume.linkd.core.*;
import java.nio.ByteBuffer;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
import java.util.function.Consumer;
public final class WebSocketSessionImpl implements WebSocketSession {
private enum State {
OPEN, CLOSING, CLOSED
}
private final String id;
private final EventLoop loop;
private final Connection connection;
private final WebSocketFrameCodec codec;
private final WebSocketFrameCodec.Assembler assembler;
private final Consumer<WebSocketFrame> onMessage;
private final int maxQueuedFrames;
private final ConcurrentLinkedQueue<ByteBuffer> outbound = new ConcurrentLinkedQueue<>();
private final AtomicInteger queued = new AtomicInteger();
private final AtomicBoolean flushScheduled = new AtomicBoolean();
private final AtomicReference<State> state = new AtomicReference<>(State.OPEN);
private volatile long lastSeenNanos = System.nanoTime();
public WebSocketSessionImpl(String id, EventLoop loop, Connection connection,
WebSocketFrameCodec codec, Consumer<WebSocketFrame> onMessage,
int maxQueuedFrames) {
this.id = id;
this.loop = loop;
this.connection = connection;
this.codec = codec;
this.assembler = codec.newAssembler();
this.onMessage = onMessage;
this.maxQueuedFrames = maxQueuedFrames;
}
public void onReadable(ByteBuffer in) {
for (WebSocketFrame frame : codec.decode(in)) {
handle(frame);
}
}
private void enqueue(WebSocketFrame frame) {
if (state.get() != State.OPEN) {
return;
}
if (queued.incrementAndGet() > maxQueuedFrames) {
queued.decrementAndGet();
outbound.clear();
queued.set(0);
close(1013, "outbound queue full");
return;
}
outbound.add(codec.encode(frame));
scheduleFlush();
}
private void scheduleFlush() {
if (flushScheduled.compareAndSet(false, true)) {
loop.execute(this::flush);
}
}
private void flush() {
flushScheduled.set(false);
ByteBuffer frame;
while ((frame = outbound.poll()) != null) {
queued.decrementAndGet();
connection.enqueue(frame);
}
if (state.get() == State.CLOSING) {
state.set(State.CLOSED);
connection.close();
}
}
}
WebSocketSession 接口只保留共享类型要求的三个方法：
public interface WebSocketSession {
void sendText(String text);
void sendBinary(byte[] data);
void close(int code, String reason);
}
id()、isOpen()、idleMillis()、sendPing() 放在实现类上，因为只有 WebSocketHub 需要它们；sendText/sendBinary 只是把 UTF-8 文本或二进制载荷包成帧再调用 enqueue。
队列的并发设计有三层：outbound 用 ConcurrentLinkedQueue 让任意线程入队；queued 用 AtomicInteger 做有界背压，避免 size() 的 O(n) 开销；flushScheduled 用 CAS 让一次突发只投递一个 flush 任务。
flush() 在事件循环线程上执行，串行地调用 connection.enqueue，所以同一会话的帧不会交错。connection.enqueue 本身如果是非阻塞队列，flush 就不会卡住事件循环；如果它内部会阻塞，必须换成非阻塞实现，否则一个慢客户端会拖住整个 loop。
队列满时的处理是「丢数据、发 1013、关闭」。这不是唯一选择：也可以暂停读事件、给上游反向压力，或者把消息写到磁盘再重试。但 WebSocket 的广播场景里，慢客户端通常意味着网络或页面已经异常，继续为它排队只会让内存涨到 OOM。1013 的语义是「稍后重试」，客户端可以重连；比无声丢消息更容易排查。
linkd-app 只做两件事：new WebSocketHub(30_000, 90_000) 启心跳，router.add("GET", "/ws/echo", ...) 回 101。会话表和广播都不在 HTTP 路由里，/healthz 的延迟不受广播影响。
常见错误与边界
错误一：把 1006 发到线上。 1006 是「异常关闭」的本地记录，不是可发送的关闭码。对端收到 1006 会认为协议错误，可能再回一个 1002，日志里出现两个关不掉的连接。发送前用白名单校验，1005、1006、1015 直接拒绝。
错误二：收到 close 不回复。 主动方发完 close 会等对端的 close，等不到就一直占着连接。服务端必须回一个 close 再关 TCP；如果不回，浏览器端的关闭回调不会触发，用户看到的是「连接一直转圈」。
错误三：广播线程直接写 channel，或用无界队列。 两个业务线程同时 channel.write，帧会交错；即使加 synchronized，也会把网络 I/O 的延迟带进锁里。队列无界则会让慢客户端把内存拖到 OOM。正确做法是每会话有界队列 + 单一 flush 线程，超限时关闭 1013 或对上游施加背压。
错误四：只在 pong 分支更新活跃时间。 只发业务消息、不回 pong 的客户端会被误判为超时。任何入站帧都算活跃；pong 只是没有业务流量时的保底信号。
边界：广播是尽力而为。 遍历期间加入的会话可能收不到这一条，断开的会话可能已经被移除；没有消息 id、确认和重放。需要可靠广播时，要在应用层做序列号和补发，而不是假设 sendText 返回就代表对方收到了。
边界：关闭原因最多 123 字节。 这是 UTF-8 编码后的字节数，不是字符数。一个中文字符通常占 3 字节，40 个汉字就会超限；截断要在字符边界上做，不能直接切字节。
小结
- 心跳用 ping/pong，pong 必须回显 ping 的载荷；任何入站帧都应更新活跃时间。
- 心跳定时器只入队，不直接写 channel；写操作统一由会话所属的事件循环串行执行。
- close 帧是 2 字节码加 UTF-8 原因，总长不超过 125 字节；1005、1006、1015 不能上线。
- 主动关闭用 CAS 让关闭只发一个 close；被动关闭回显对端的码，然后关闭 TCP。
- 会话表用
ConcurrentHashMap，广播用弱一致迭代，删除用remove(key, value)。 - 每会话有界队列加单一 flush，既避免帧交错，也把慢客户端的背压变成 1013 关闭。
WebSocket 的连接生命周期到这里闭环。下一节回到 Reactor 主线：事件循环怎么组织任务队列，怎么把读、写、定时任务和跨线程提交放进同一个线程模型。
阅读导航：上一节：11.2 帧格式、掩码与分片 · 下一节：12.1 事件循环与任务队列 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。