Session 注册
每个 WebSocket 连接是一个独立的会话。要管理多个客户端，需要在连接建立时注册、断开时注销：
import reactor.netty.http.server.HttpServer;
import reactor.netty.http.websocket.WebsocketInbound;
import reactor.netty.http.websocket.WebsocketOutbound;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
public class SessionServer {
// sessionId → outbound sink（用于向该 session 推送）
static final ConcurrentHashMap<String, Sinks.Many<String>> sessions = new ConcurrentHashMap<>();
public static void main(String[] args) {
HttpServer.create().port(8080)
.route(routes -> routes
.ws("/ws/chat", (in, out) -> {
String sessionId = java.util.UUID.randomUUID().toString();
System.out.println("Connected: " + sessionId);
// 创建该 session 的 sink：显式给有界队列，慢客户端不会无限占内存
Sinks.Many<String> sink = Sinks.many().unicast()
.onBackpressureBuffer(new ArrayBlockingQueue<>(256));
sessions.put(sessionId, sink);
// 接收客户端消息
Mono<Void> receive = in.receive()
.asString()
.doOnNext(msg -> handleMessage(sessionId, msg))
.doOnTerminate(() -> {
System.out.println("Disconnected: " + sessionId);
sessions.remove(sessionId);
})
.then();
// 向该 session 发送 sink 中的消息
Mono<Void> send = out.sendString(sink.asFlux()).then();
// 两个 Mono<Void> 都没有值：任一端终止 zip 就结束并取消另一端，
// 因此客户端断开或 session 被关闭都会结束 handler
return Mono.zip(receive, send).then();
})
)
.bindNow();
System.out.println("Chat server: ws://localhost:8080/ws/chat");
try { System.in.read(); } catch (Exception ignored) {}
}
static void handleMessage(String from, String msg) {
System.out.println("[" + from + "] " + msg);
// 广播给所有 session
sessions.forEach((id, sink) -> {
if (!id.equals(from)) {
sink.tryEmitNext("[" + from + "]: " + msg);
}
});
}
}
这个实现用了 Sinks.Many 作为每个 session 的独立消息管道。unicast() 保证每个 sink 只有一个订阅者（对应一条 WebSocket 连接）。要注意无参的 onBackpressureBuffer() 用的是无界队列（Queues.unbounded()），慢客户端只会让内存一直涨，永远不会触发 FAIL_OVERFLOW；上面显式传入 ArrayBlockingQueue(256) 才有界：队列满了 tryEmitNext 返回 FAIL_OVERFLOW，把慢客户端限制在它自己的缓冲里，不拖累其他 session。
Sinks 的选择
不同场景需要不同类型的 sink：
| Sink 类型 | 用途 | 背压策略 |
|---|---|---|
unicast().onBackpressureBuffer() | 单订阅者 | 默认无界（Queues.unbounded()），不会溢出；要背压保护需传有界队列，满时 tryEmitNext 返回 FAIL_OVERFLOW |
multicast().onBackpressureBuffer() | 多订阅者共享 | 默认 256 条共享队列（Queues.SMALL_BUFFER_SIZE）；慢订阅者让队列积压，满时所有订阅者一起受影响。要按订阅者丢弃用 directBestEffort()，要严格全发用 directAllOrNothing() |
replay().all() | 新订阅者收到历史 | 内存随历史增长 |
replay().limit(n) | 新订阅者收到最近 n 条 | 固定内存 |
对于聊天场景，unicast 最合适——每个 WebSocket 连接只有一个消费方（客户端）。
按用户标识路由
上面的实现用 UUID 作为 session ID，但实际业务通常需要按用户标识路由。WebsocketInbound 没有 nettyRequest()，拿不到原始请求 URI；它保留的是握手阶段的 HTTP 头，可以从这里读取客户端带上来的身份：
.ws("/ws/chat", (in, out) -> {
String userId = in.headers().get("X-User-Id");
if (userId == null || userId.isBlank()) {
return out.sendClose(1008, "missing user id"); // 1008 = policy violation
}
Sinks.Many<String> sink = Sinks.many().unicast().onBackpressureBuffer();
sessions.put(userId, sink); // 用 userId 替代 UUID
// ...
})
生产环境不要让客户端随便填 X-User-Id：这个头必须由网关在认证后注入，或者从 Authorization 里的 JWT 解析。如果一定要用查询参数，得改用 ws(Predicate<HttpServerRequest>, handler, spec) 重载，在 predicate 里从 HttpServerRequest 取参数，再通过 Channel attribute 传给 handler，比放 header 复杂得多。
广播的背压问题
广播场景中最大的陷阱是慢客户端拖垮整个系统。如果 100 个客户端中有 1 个网络极慢，逐个同步发送会导致其他 99 个也被阻塞。
解决：对每个 session 独立发送，并检查 tryEmitNext 的返回值：
static void broadcast(String msg) {
sessions.forEach((id, sink) -> {
Sinks.EmitResult result = sink.tryEmitNext(msg);
if (result.isFailure()) {
// 背压溢出：sink 缓冲区满（session 消费太慢）
System.err.println("Backpressure drop for session: " + id);
// 可选：关闭慢连接
// sink.tryEmitComplete();
}
});
}
tryEmitNext 是非阻塞的发送尝试。如果背压策略不允许（如缓冲区满），立即返回 EmitResult.FAIL_OVERFLOW 而不是一直等待。
订阅模式：按主题过滤
不是所有客户端都想接收所有消息。实现主题订阅：
static final ConcurrentHashMap<String, Sinks.Many<String>> sessions = new ConcurrentHashMap<>();
static final ConcurrentHashMap<String, java.util.Set<String>> subscriptions = new ConcurrentHashMap<>();
static void subscribe(String sessionId, String topic) {
subscriptions.computeIfAbsent(sessionId, k -> ConcurrentHashMap.newKeySet()).add(topic);
}
static void publish(String topic, String msg) {
subscriptions.forEach((sessionId, topics) -> {
if (topics.contains(topic)) {
var sink = sessions.get(sessionId);
if (sink != null) {
sink.tryEmitNext(msg);
}
}
});
}
客户端发送 SUBSCRIBE:orders 订阅订单主题，只接收该主题的推送。
连接数限制
生产环境需要限制最大并发连接数，防止资源耗尽：
HttpServer.create()
.port(8080)
.option(io.netty.channel.ChannelOption.SO_BACKLOG, 128) // server socket 的 accept 队列
.route(routes -> routes
.ws("/ws/chat", (in, out) -> {
if (sessions.size() >= 10_000) {
return out.sendClose(1013, "server busy"); // 1013 = Try Again Later
}
// ... 正常注册 session
return Mono.empty();
})
)
注意 SO_BACKLOG 控制的是 accept 队列（已完成握手、等待应用 accept）大小，不是总连接数。真正的连接数限制要在 ws 处理器入口做（上面的 sessions.size() 检查），因为 HTTP 的 503 响应救不了已经完成升级的 WebSocket 连接——只能用 Close 帧（1013）告诉客户端稍后再试。
优雅的 Session 清理
应用关闭时，需要通知所有客户端并关闭连接：
static void shutdown() {
sessions.forEach((id, sink) -> {
sink.tryEmitNext("{\"type\":\"shutdown\"}");
sink.tryEmitComplete(); // 发送 onComplete，客户端收到 Close 帧
});
sessions.clear();
}
在实际部署中，配合 Kubernetes 的 preStop 钩子和 SIGTERM 信号处理器调用 shutdown()。
eventd v2 的 WebSocket 设计
eventd v2 的实时推送层使用 WebSocket 广播行情事件：
┌─→ session A (用户 alice)
EventSource ─┼─→ session B (用户 bob) ← 每个 session 独立 sink
└─→ session C (用户 charlie)
↓
慢消费者 → 缓冲区满 → tryEmitNext 返回 FAIL_OVERFLOW
↓
关闭慢连接，释放资源
广播前检查每个 session 的订阅列表（用户可能只关心特定 symbol 的行情），避免无效推送。
小结
| 能力 | 实现 | 注意 |
|---|---|---|
| Session 注册 | ConcurrentHashMap<String, Sink> | 连接断开时清理 |
| 按用户路由 | 从 query/header 读取 userId | 生产用认证 token |
| 广播 | sessions.forEach((id, sink) -> sink.tryEmitNext(...)) | 用 tryEmitNext 防阻塞 |
| 背压处理 | EmitResult.isFailure() | 慢客户端可断开 |
| 主题订阅 | ConcurrentHashMap<sessionId, Set<topic>> | 过滤避免无效推送 |
| 连接限制 | 应用层 sessions.size() 检查 | 已升级的连接只能用 Close 1013，HTTP 503 发不出去 |
| 优雅关闭 | sink.tryEmitComplete() | 客户端收到关闭信号 |
下一节 3.3 eventd v2 网关 ：网关模块结构、路由表设计、HTTP ↔ WebSocket 协议转换、与 API 层的衔接。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。