本节目标：把每条连接的回显升级成“谁在线、给谁发、谁已经死了”三件事，并让广播在慢客户端面前不拖垮整个进程。
适用版本：JDK 25 + Netty 4.2.19.Final（netty-transport、netty-handler、netty-codec-http模块）。
承接的问题
10.2 的 /ws/echo 是“来一条回一条”，连接之间互不知道对方存在。/ws/broadcast 需要的是另一套结构：
- 新连接完成握手后要登记，断开后要自动消失，否则广播会一直往已关闭的
Channel写，日志里全是ClosedChannelException。 - 一条消息要发给所有在线连接，但某个客户端网络很差、出站缓冲已经堆满时，不能让它把广播线程卡住。
- 客户端拔网线、NAT 表项过期时，TCP 可能几分钟都不通知服务端；连接对象和堆外缓冲一直占着，必须有应用层心跳。
- 广播需要知道“是谁发的”“发给哪个房间”，至少要把
userId和Channel绑定起来。
这一节用 ChannelGroup 做会话表，用 AttributeKey 存用户标识，用 IdleStateHandler 加 ping/pong 做存活检测，最后给出 /ws/broadcast 的完整 WebSocketHub 和广播边界。
核心机制一：ChannelGroup 就是会话表
io.netty.channel.group.ChannelGroup 是一个 Set<Channel>，额外提供 write、writeAndFlush、close 这类面向所有成员的批量操作。最常用的实现是：
private final ChannelGroup group =
new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
DefaultChannelGroup 的关键行为是自动移除：每个 Channel 加入时，它会在这个 channel 的 close future 上挂一个监听器，连接关闭后把 channel 从集合里删掉。业务代码不需要在 channelInactive 里手动 remove，也不会因为忘记删除而让集合无限增长。
GlobalEventExecutor 是 io.netty.util.concurrent 里的全局单线程执行器，专门用来跑这类轻量的通知任务。它不属于任何 I/O EventLoop，所以从任意业务线程调用 group.add、group.writeAndFlush 都是线程安全的；监听器也不会阻塞某个 worker 线程。
登记点仍然只能是 HANDSHAKE_COMPLETE，不能是 channelActive：
if (evt == WebSocketServerProtocolHandler.ServerHandshakeStateEvent.HANDSHAKE_COMPLETE) {
group.add(ctx.channel());
}
用户标识放在 AttributeKey 上，而不是另开一个 Map<ChannelId, String>：
public static final AttributeKey<String> USER_ID = AttributeKey.valueOf("linkd.userId");
AttributeKey 挂在 Channel 上，连接关闭时随 channel 一起回收，不存在“会话表删了、用户表忘了删”的第二份状态。valueOf 同一个名字返回同一个 key 实例，跨 handler 读取时不需要传对象引用。
userId 在握手前从查询串取出来。10.1 讲过默认路径匹配会把查询串算进去，所以 /ws/broadcast 的协议 handler 要开 checkStartsWith；参数解析放在 WebSocketServerProtocolHandler 之前：
package com.plume.linkd.v2.server;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.QueryStringDecoder;
import io.netty.util.AttributeKey;
import java.util.List;
public final class UserIdExtractor extends ChannelInboundHandlerAdapter {
public static final AttributeKey<String> USER_ID = AttributeKey.valueOf("linkd.userId");
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
if (msg instanceof FullHttpRequest request) {
List<String> ids = new QueryStringDecoder(request.uri()).parameters().get("userId");
if (ids != null && !ids.isEmpty()) {
ctx.channel().attr(USER_ID).set(ids.get(0));
}
}
ctx.fireChannelRead(msg);
}
}
这个 handler 只读 uri()，不消费消息，也不改变引用计数：ChannelInboundHandlerAdapter 不自动释放入站对象，ctx.fireChannelRead(msg) 把所有权原样交给下一个 handler。放在 HttpObjectAggregator 之后，才能拿到 FullHttpRequest；放在 WebSocketServerProtocolHandler 之前，才能在 HTTP 请求被升级消费前读到查询串。
如果业务需要按 userId 反查 channel，再维护一个 ConcurrentHashMap<String, Channel>，但删除时要用 remove(key, value)，避免误删同名的新连接。ChannelGroup 负责“所有连接”，map 负责“按用户找连接”，两者职责分开；不要指望一个 map 同时承担广播和自动清理。
核心机制二：广播与慢客户端
最简单的广播只有一行：
group.writeAndFlush(new TextWebSocketFrame(text));
ChannelGroup 会遍历所有成员，为每个 channel 准备一份 retained duplicate 再写出。DefaultChannelGroup.write 内部对 ByteBuf 或 ByteBufHolder 走 retainedDuplicate()，最后释放调用方传入的原始消息；所以把帧交给 writeAndFlush 之后，不要再 release 它，也不要拿同一个对象去做别的事。
这一行没有 isWritable 过滤。某个客户端出站缓冲超过高水位线时，writeAndFlush 仍然会把消息排进它的队列，队列继续涨，最后要么触发 OOM，要么在写失败时关闭连接。/ws/broadcast 用的是带过滤的版本：
public void broadcast(Channel sender, String text) {
int slow = 0;
for (Channel channel : group) {
if (!channel.isActive()) {
continue;
}
if (!channel.isWritable()) {
slow++;
continue;
}
channel.writeAndFlush(new TextWebSocketFrame(text));
}
if (slow > 0) {
LOG.log(System.Logger.Level.WARNING,
"broadcast skipped slow clients=" + slow + " sender=" + sender.remoteAddress());
}
}
这里的 isWritable() 是 9.1 讲的水位线判断：出站缓冲低于低水位时为 true，高于高水位时为 false。广播跳过 false 的连接，语义是“这条消息你不收了”，而不是“服务端替你排队”。如果某个连接持续不可写，可以选择用 1013 关闭它，或者让它自己重连；继续为它累积消息只会把内存压力从一条连接扩散到整个进程。
ChannelGroup 的迭代是弱一致的：遍历期间加入的连接可能收不到这一条，刚断开的连接可能还会被访问一次（isActive 会挡住）。广播本来就是尽力而为，不承诺每个连接都收到；要可靠投递，需要消息 id、确认和重放，那是应用层协议，不是 ChannelGroup 的职责。
本机用两个 JDK java.net.http.WebSocket 客户端连 /ws/broadcast?userId=alice 和 ?userId=bob，由 alice 发一条消息，输出：
alice got=hello all
bob got=hello all
发送者自己也收到广播，因为循环没有排除 sender。要排除发送者，把第一行改成 if (channel == sender) continue;，但这样发送者就不知道广播是否成功，需要额外的确认消息。
核心机制三：心跳与死连接清理
WebSocket 的 ping/pong 是控制帧，WebSocketServerProtocolHandler 已经自动处理：收到 ping 回 pong，收到 pong 直接丢弃。业务 handler 看不到它们（10.2 已说明）。因此心跳检测要放在协议 handler 之前：
ch.pipeline().addLast(new IdleStateHandler(90, 30, 0, TimeUnit.SECONDS));
ch.pipeline().addLast(new WebSocketHeartbeatHandler());
ch.pipeline().addLast(new WebSocketServerProtocolHandler(...));
三个参数依次是读空闲、写空闲、全空闲，单位秒。这里的配置是“90 秒没读到任何东西就认为对端死了，30 秒没写东西就发一个 ping，全空闲不启用”。
为什么不启用全空闲？因为服务端自己每 30 秒发一次 ping，写操作会重置全空闲计时器；即使客户端一个字节都不回，全空闲也永远不会触发。真正能证明对端还活着的是入站数据：pong、文本、二进制都算。读空闲才是死亡信号。
心跳 handler 把两类事件分开处理：
package com.plume.linkd.v2.server;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import io.netty.handler.timeout.IdleStateEvent;
public final class WebSocketHeartbeatHandler extends ChannelDuplexHandler {
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
if (evt instanceof IdleStateEvent) {
boolean writerIdle = evt == IdleStateEvent.WRITER_IDLE_STATE_EVENT
|| evt == IdleStateEvent.FIRST_WRITER_IDLE_STATE_EVENT;
if (writerIdle) {
ctx.channel().writeAndFlush(new PingWebSocketFrame());
} else {
ctx.channel().writeAndFlush(new CloseWebSocketFrame(1001, "heartbeat timeout"))
.addListener(ChannelFutureListener.CLOSE);
}
return;
}
ctx.fireUserEventTriggered(evt);
}
}
两个细节。第一，用 ctx.channel().writeAndFlush 而不是 ctx.writeAndFlush：从 channel 的 tail 开始走完整出站链，close 帧会经过 WebSocketServerProtocolHandler，由它记录关闭状态并启动关闭流程；从当前 handler 的 ctx 写出会跳过协议 handler 的这段逻辑。第二，ping 不带载荷，控制帧最多 125 字节；如果要用载荷算 RTT，放 8 字节时间戳就够了，不要把业务 JSON 塞进 ping。
本机把空闲时间临时改成 3 秒读空闲、1 秒写空闲，用一个只完成握手、之后不回任何帧的原始客户端观察，输出：
t=1.00s opcode=0x9 len=0 payload=
t=2.00s opcode=0x9 len=0 payload=
t=3.01s opcode=0x8 len=19 payload=03e96865617274626561742074696d656f7574
前三行是每 1 秒一个 ping（opcode 0x9）；3.01 秒时服务端发 close（opcode 0x8），载荷 03e9 是状态码 1001，后面的 UTF-8 字节是 heartbeat timeout。生产配置用 30 秒 ping、90 秒超时，这两个值不是协议规定，要按 RTT 和故障发现时间调；macOS 的 TCP keepalive 默认空闲两小时才开始探测，不能替代应用层心跳。
在 linkd-v2 里落地：/ws/broadcast 的完整 WebSocketHub
把前面的片段合成一个类。它只接受文本帧，二进制和聚合错误由 10.2 的 handler 处理；ChannelGroup 负责会话表，AttributeKey 负责用户标识：
package com.plume.linkd.v2.server;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.TooLongFrameException;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.util.concurrent.GlobalEventExecutor;
@ChannelHandler.Sharable
public final class WebSocketHub extends SimpleChannelInboundHandler<TextWebSocketFrame> {
private static final System.Logger LOG = System.getLogger(WebSocketHub.class.getName());
private final ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
if (evt == WebSocketServerProtocolHandler.ServerHandshakeStateEvent.HANDSHAKE_COMPLETE) {
String userId = ctx.channel().attr(UserIdExtractor.USER_ID).get();
if (userId == null) {
userId = "anonymous";
ctx.channel().attr(UserIdExtractor.USER_ID).set(userId);
}
group.add(ctx.channel());
LOG.log(System.Logger.Level.INFO,
"ws join user=" + userId + " size=" + group.size());
return;
}
ctx.fireUserEventTriggered(evt);
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame frame) {
broadcast(ctx.channel(), frame.text());
}
public void broadcast(Channel sender, String text) {
int slow = 0;
for (Channel channel : group) {
if (!channel.isActive()) {
continue;
}
if (!channel.isWritable()) {
slow++;
continue;
}
channel.writeAndFlush(new TextWebSocketFrame(text));
}
if (slow > 0) {
LOG.log(System.Logger.Level.WARNING,
"broadcast skipped slow clients=" + slow + " sender=" + sender.remoteAddress());
}
}
public int size() {
return group.size();
}
public void closeAll(int code, String reason) {
for (Channel channel : group) {
channel.writeAndFlush(new CloseWebSocketFrame(code, reason))
.addListener(ChannelFutureListener.CLOSE);
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
if (cause instanceof TooLongFrameException) {
ctx.writeAndFlush(new CloseWebSocketFrame(1009, "message too big"))
.addListener(ChannelFutureListener.CLOSE);
return;
}
LOG.log(System.Logger.Level.WARNING,
"ws error remote=" + ctx.channel().remoteAddress(), cause);
ctx.close();
}
}
WebSocketHub 本身标 @Sharable，因为它的全部状态是 ChannelGroup，没有按连接变化的字段。SimpleChannelInboundHandler<TextWebSocketFrame> 仍然只接受文本；二进制回显、聚合器、会话表三者可以共存，靠类型过滤分流。
/ws/broadcast 的 pipeline 装配如下，注意前四行的顺序：
ch.pipeline().addLast(new HttpServerCodec());
ch.pipeline().addLast(new HttpObjectAggregator(64 * 1024));
ch.pipeline().addLast(new UserIdExtractor());
ch.pipeline().addLast(new IdleStateHandler(90, 30, 0, TimeUnit.SECONDS));
ch.pipeline().addLast(new WebSocketHeartbeatHandler());
ch.pipeline().addLast(new WebSocketServerProtocolHandler(
"/ws/broadcast", null, true, 64 * 1024, false, true));
ch.pipeline().addLast(new WebSocketFrameAggregator(64 * 1024));
ch.pipeline().addLast(hub);
UserIdExtractor 在协议 handler 之前，才能读到查询串；IdleStateHandler 和心跳 handler 也在协议 handler 之前，才能看到 pong 和原始入站字节；WebSocketFrameAggregator 在协议 handler 之后，才能消费已经解出来的帧；hub 在最后，负责业务。六参数构造器的最后一个 true 是 checkStartsWith，没有它，/ws/broadcast?userId=alice 不会匹配。
关闭服务器时先关会话，再关监听，最后停 EventLoop：
public void close() {
hub.closeAll(1001, "server shutdown");
if (serverChannel != null) {
serverChannel.close().syncUninterruptibly();
}
boss.shutdownGracefully();
worker.shutdownGracefully();
}
顺序反了会让 closeAll 往已经被 EventLoop 丢弃的连接写 close 帧；serverChannel.close() 先执行能挡住新连接，但已有连接还在 worker 上，shutdownGracefully 会等它们把关闭流程走完。
广播风暴与按房间分组
/ws/broadcast 的广播是 O(N) 写，但 N 个客户端各发一条就是 O(N²)。1000 个连接同时发言，每人一条，服务端要写 100 万次；isWritable 只能保护单条连接，保护不了总 CPU 和总内存。三个缓解方向：
- 按房间分组：
Map<String, ChannelGroup>，每个房间一个 group，广播只遍历本房间。房间数多时，map 本身要清理，空 group 不能无限留着。 - 限流与去重：同一用户 1 秒内只允许发一条，或者对相同内容做合并；这属于业务层，
ChannelGroup不提供。 - 换成发布订阅：进程内广播用 group 足够；跨节点广播需要 Redis pub/sub、NATS 或消息队列，每个节点只向本机连接扇出。
按房间分组的最小结构是：
private final Map<String, ChannelGroup> rooms = new ConcurrentHashMap<>();
ChannelGroup room(String name) {
return rooms.computeIfAbsent(name, key ->
new DefaultChannelGroup(GlobalEventExecutor.INSTANCE));
}
DefaultChannelGroup 会在连接关闭时自动移除成员，但不会把空 group 从 rooms 里删掉；房间很多且生命周期短时，需要在成员数归零后清理，或者定期扫描 isEmpty()。并发下 size()==0 之后可能又有新连接加入，清理要用 rooms.remove(name, group) 的带值删除，避免误删刚被复用的房间。
常见错误与边界
错误一：在 channelInactive 里手动 group.remove，又依赖自动移除。 两个动作本身不冲突，但手动删除后再遍历可能触发弱一致迭代的边界情况；更重要的是，如果只手动删而忘了处理异常关闭路径，就会漏。交给 DefaultChannelGroup 的 close future 监听器，业务只负责 add。
错误二：广播时对每个连接 writeAndFlush 却不看 isWritable。 慢客户端会把出站队列越堆越长，最后拖垮整个进程。广播要么跳过不可写连接，要么用 1013 关闭它；不要用无界队列“先存着”。
错误三：把心跳定时器放在业务线程里直接写 channel。 定时线程和 EventLoop 线程同时写同一条连接，帧可能交错。心跳只应该发一个 PingWebSocketFrame，由所属 EventLoop 串行写出；Netty 的 writeAndFlush 本身是线程安全的，但不要在定时线程里做阻塞等待。
错误四：只启用 allIdle。 服务端自己发 ping 会重置全空闲计时器，客户端死了也检测不到。用读空闲判断死亡，写空闲触发 ping。
边界：广播没有投递保证。 遍历期间加入的收不到，跳过的慢客户端收不到，写失败的也收不到；没有消息 id、没有重放。需要可靠广播时，在应用层做序列号和补发。
边界：ChannelGroup 不是分布式会话表。 它只包含本 JVM 的 channel。多节点部署时，每个节点各有一份 group，跨节点广播要经过外部消息系统；不要假设 group.size() 是集群在线数。
小结
DefaultChannelGroup(GlobalEventExecutor.INSTANCE)在连接关闭时自动移除成员，登记点放在HANDSHAKE_COMPLETE。- 用户标识用
AttributeKey挂在Channel上，随连接回收；需要反查时另建 map，并用remove(key, value)删除。 - 广播优先用
group.writeAndFlush；需要过滤慢客户端时遍历 group，按isActive()和isWritable()跳过。 - 心跳把
IdleStateHandler放在协议 handler 之前，读空闲关闭、写空闲发 ping；全空闲会被自己的 ping 重置，不能用来判死。 - close 帧的状态码用 1001、1009、1013 这类线上值，1005、1006、1015 不能发送。
- 广播是 O(N)，N 个客户端同时发言就是 O(N²)；按房间分组、限流和外部 pub/sub 是三个扩展方向。
WebSocket 这条线到这里闭环：握手、帧、会话、心跳、广播。下一节进入 HTTP/2，看 Http2FrameCodec 如何把一条 TCP 连接拆成多条独立的多路复用流。
阅读导航：上一节：10.2 帧处理、分片与聚合 · 下一节：11.1 Http2FrameCodec 与多路复用 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。