java

3.1 WebSocket 服务端

在 Reactor Netty 上搭建 WebSocket 服务端：upgrade 握手、websocket() 回调、文本与二进制帧收发、inbound/outbound 双流管道、帧类型判断与 ping/pong 心跳处理。附可编译运行示例。

Leeting Yan 5 分钟阅读 2240 字

HTTP Upgrade 到 WebSocket

WebSocket 不是独立的协议，而是 HTTP 的升级。客户端先发送带 Upgrade: websocket 头的 HTTP 请求，服务端返回 101 Switching Protocols，之后连接转为全双工帧传输。

Reactor Netty 把这个过程封装在 .websocket() 方法中：

import reactor.netty.http.server.HttpServer;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Flux;

public class WsServer {
    public static void main(String[] args) {
        HttpServer.create().port(8080)
            .route(routes -> routes
                .ws("/ws/echo", (wsInbound, wsOutbound) -> {
                    // 把收到的所有文本帧原样返回
                    return wsOutbound.sendString(
                        wsInbound.receive().asString()
                    );
                })
            )
            .bindNow();
        System.out.println("WebSocket: ws://localhost:8080/ws/echo");
        try { System.in.read(); } catch (Exception ignored) {}
    }
}

.route(routes -> ...) 是比 .handle() 更正式的路由 API，支持按路径和方法匹配。.ws("/ws/echo", handler) 自动处理 HTTP upgrade 握手，把匹配路径的连接交给 WebSocket handler。

WebSocketFrame 与帧类型

WebSocket 消息被分割为帧。WebsocketInbound.receiveFrames() 返回 Flux<WebSocketFrame>；常用的 receive() 返回的是 ByteBufFlux，只有消息 payload，没有帧类型信息。要区分文本帧、二进制帧和控制帧，必须用 receiveFrames()：

帧类型含义处理
TextWebSocketFrame文本数据.text() 读取字符串
BinaryWebSocketFrame二进制数据.content() 读取 ByteBuf
CloseWebSocketFrame关闭连接发回 Close 帧，释放资源
PingWebSocketFrame心跳探测自动回复 Pong（Reactor Netty 默认）
PongWebSocketFrame心跳响应通常无需处理

区分帧类型

.ws("/ws/typed", (in, out) -> {
    return out.sendString(
        in.receiveFrames()
            .map(frame -> {
                // 帧对象在本回调返回后由 Reactor Netty 自动释放：这里只读内容，不要手动 release
                if (frame instanceof TextWebSocketFrame text) {
                    return "TEXT:" + text.text();
                }
                if (frame instanceof BinaryWebSocketFrame binary) {
                    return "BINARY:" + binary.content().readableBytes() + " bytes";
                }
                if (frame instanceof CloseWebSocketFrame) {
                    return "CLOSE";
                }
                return "OTHER:" + frame.getClass().getSimpleName();
            })
    );
})

引用计数只有一条规则：receive() 和 receiveFrames() 发出的对象都归 Reactor Netty 所有，它在 onNext 返回后自动释放。因此在本回调里同步读取内容（asString()、text()、content().readableBytes()）既不需要 retain()，也不能手动 release()——手动释放会让随后的自动释放抛出 IllegalReferenceCountException: refCnt: 0, decrement: 1，当前连接直接断开。只有当对象需要跨出当前回调继续存活时才 retain()，例如交给另一条连接异步发送（client.send(req.receive().retain())）或放进缓存稍后读取；这时释放责任转移到接收方，接收方消费完必须自己释放。反过来，receive().retain().asString() 是常见误区：retain() 多留了一个引用，asString() 读完并不释放，结果是每帧泄漏一块直接内存。

服务端主动推送

WebSocket handler 的返回类型是 Publisher<Void>（实际几乎总是用 Mono<Void> 或 NettyOutbound），它的终止表示 handler 的生命周期结束。如果你想在连接期间持续推送（如定时心跳、行情广播），需要用 Flux.merge 组合多个流：

import java.time.Duration;

.ws("/ws/push", (in, out) -> {
    // 流 1：每 5 秒推送一次心跳
    Flux<String> heartbeat = Flux.interval(Duration.ofSeconds(5))
        .map(i -> "{\"type\":\"heartbeat\",\"seq\":" + i + "}");

    // 流 2：把客户端消息加上时间戳回显
    Flux<String> echo = in.receive().asString()
        .map(msg -> "{\"type\":\"echo\",\"msg\":\"" + msg + "\",\"time\":" + System.currentTimeMillis() + "}");

    // 合并两个流输出
    return out.sendString(Flux.merge(heartbeat, echo));
})

Flux.merge 把多个流交错合并。心跳流和客户端消息流同时输出到 wsOutbound。

连接生命周期

WebSocket handler 返回的 Mono<Void> 完成时，连接关闭。要保持连接直到客户端主动断开，需要让 Mono 永远不完成：

.ws("/ws/persistent", (in, out) -> {
    // 1. 接收客户端消息（消费 inbound，否则 receive() 不请求）
    Mono<Void> receive = in.receive()
        .asString()
        .doOnNext(msg -> System.out.println("Received: " + msg))
        .then();  // 转成 Mono<Void>，但只在连接关闭时完成

    // 2. 持续推送
    Mono<Void> send = out.sendString(
        Flux.interval(Duration.ofSeconds(1))
            .map(i -> "tick-" + i)
    ).then();

    // 3. 任一端终止（完成或出错）时结束 handler，连接随之关闭
    return Mono.zip(receive, send).then();
})

关键点：

  • in.receive() 必须被订阅，否则 Reactor Netty 不读取客户端发来的数据，TCP 窗口会阻塞
  • out.sendString() 必须被调用，否则没有数据可写
  • Mono.zip(receive, send).then() 在 Mono<Void> 上不是等两端都完成：两个流都不发射值，任何一方完成或出错，zip 就立即终止并取消另一方。这里正是要这个语义——客户端断开让 receive 完成，或发送流结束/出错，都会结束 handler 并关闭连接
  • 如果确实需要等两个流都自然结束，用 Mono.when(receive, send)；但对 WebSocket 持久连接，任一端终止就关闭更符合预期，也能避免对端已断开后继续向已关闭的通道写数据

Ping/Pong 心跳

虽然 Reactor Netty 自动回复 Ping 帧，但服务端主动发送 Ping 可以更快地检测断开的客户端：

.ws("/ws/heartbeat", (in, out) -> {
    // 每 10 秒发送一次 Ping
    Flux<WebSocketFrame> pings = Flux.interval(Duration.ofSeconds(10))
        .map(i -> new PingWebSocketFrame());

    Mono<Void> sendPings = out.sendObject(pings).then();

    // 30 秒收不到任何帧（正常客户端会自动回 Pong）就超时并关闭连接
    Mono<Void> receive = in.receiveFrames()
        .timeout(Duration.ofSeconds(30))
        .doOnNext(frame -> {
            // 帧由 Reactor Netty 自动释放，这里只判断类型，不要 release
            if (frame instanceof PongWebSocketFrame) {
                System.out.println("Pong received at " + System.currentTimeMillis());
            }
        })
        .then();

    return Mono.zip(sendPings, receive).then();
})

注意：PingWebSocketFrame 在 Netty 的 io.netty.handler.codec.http.websocketx 包中，不要在 handler 返回流中发送纯字符串 ping——浏览器/客户端不会识别。

receive 上的 timeout 触发后会发出 onError，Mono.zip 立即把错误传给 handler，连接随即关闭，这正是心跳超时想要的效果；sendPings 则被取消。

可运行完整示例

import reactor.netty.http.server.HttpServer;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.time.Duration;

public class EchoWsServer {
    public static void main(String[] args) {
        HttpServer.create().port(8080)
            .route(routes -> routes
                .ws("/ws", (in, out) -> {
                    Flux<String> ticks = Flux.interval(Duration.ofSeconds(5))
                        .map(i -> "[server] tick " + i);

                    Flux<String> responses = in.receive().asString()
                        .map(msg -> "[echo] " + msg);

                    return out.sendString(Flux.merge(ticks, responses));
                })
            )
            .bindNow();

        System.out.println("ws://localhost:8080/ws");
        System.out.println("Test: wscat -c ws://localhost:8080/ws");
        try { System.in.read(); } catch (Exception ignored) {}
    }
}

编译运行后用 wscat 测试：

# 安装 wscat
npm install -g wscat
# 连接测试
wscat -c ws://localhost:8080/ws
> hello
< [echo] hello
< [server] tick 0
< [server] tick 1

与 HTTP 路由共存

同一个 HttpServer 可以同时处理 HTTP 和 WebSocket：

HttpServer.create().port(8080)
    .route(routes -> routes
        .get("/health", (req, res) ->
            res.sendString(Mono.just("ok")))
        .get("/api/status", (req, res) ->
            res.sendString(Mono.just("{\"status\":\"up\"}")))
        .ws("/ws", (in, out) ->
            out.sendString(in.receive().asString()))
    )
    .bindNow();

route() API 支持 .get()、.post()、.put()、.delete() 以及 .ws()，每个路由独立匹配。

小结

能力API注意
开启 WebSocket.route(r -> r.ws("/path", handler))自动处理 HTTP upgrade
接收消息receive()（payload）/ receiveFrames()（带帧类型）必须订阅，否则 TCP 阻塞；对象由 Reactor Netty 自动释放，不要手动 release
发送消息wsOutbound.sendString() / sendObject()sendObject 需正确帧类型
持久连接Mono.zip(receive, send).then()任一端终止就结束 handler（关闭连接）；Mono.when 才是等双方都完成
帧类型instanceof Text/Binary/Close/Ping/PongClose 帧只需识别，不用手动释放
心跳Flux.interval + PingWebSocketFrame客户端也应回复 Pong
引用计数.retain()只在对象跨出 onNext 异步转发时使用

下一节 3.2 WebSocket 会话管理与广播 ：多 session 管理、mapConnect、广播与订阅模式、背压在广播场景的特殊处理。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「java」更多文章

  1. 15.3 全书总结与学习路径回顾
  2. 15.2 响应式流处理的演进趋势
  3. 15.1 引擎选型指南