1. 从一次推送延迟说起为什么要关注响应式 WebSocket前阵子帮一个做实时协作工具的朋友排查问题场景很典型一个基于 Spring Boot 2.x 的在线白板应用用户量不大几十个人同时在线但偶尔会出现光标位置更新延迟一两秒的情况。服务端用的是传统的ServerEndpoint注解方式实现的 WebSocket线程池配置也不算小按理说不该卡。抓了几次线程栈之后发现问题出在消息广播环节——每次广播都要遍历所有会话逐个发送一旦某个客户端网络状况差sendMessage就会阻塞住当前线程后面的消息全排队。这就是典型的阻塞式 IO 在高并发推送场景下的软肋。解决思路有两条一是把广播逻辑丢到独立线程池做异步化二是干脆换成响应式栈用Reactive WebSocket从根上把阻塞问题消掉。前者改动小但治标不治本后者学习曲线陡一点但配合 Spring Boot 2.x 的 WebFlux 体系能拿到背压、非阻塞、函数式路由这一整套能力。这篇文章就围绕第二条路展开把Spring Boot 2.x 环境下 Reactive WebSocket 的落地过程完整拆一遍包括它和传统 WebSocket 的本质区别、依赖怎么选、Handler 怎么写、会话怎么管理、背压怎么处理以及我在实际调试中踩过的那些坑。适合谁看如果你已经写过基础的 Spring Boot WebSocket 示例但对WebFlux、Reactor这套东西还停留在“听说过”的阶段或者你正在做一个消息推送、实时行情、协同编辑类的功能被阻塞和内存泄漏折腾过那这篇内容应该能帮你少走点弯路。我会尽量把每个“为什么这么写”讲清楚而不是甩一段代码让你自己悟。2. 响应式 WebSocket 到底解决了什么问题2.1 传统 WebSocket 的阻塞模型瓶颈传统 Spring WebSocket 建立在 Servlet 容器之上底层是Tomcat或Jetty的线程模型。每个 WebSocket 连接在握手完成后会绑定到一个容器线程上消息的读取和写入都在这条线程或它的派生线程里完成。这种模型在连接数不多、消息量不大的时候完全够用但一旦出现下面几种情况就会暴露问题。第一种是慢消费者。广播时如果某个客户端接收慢session.sendMessage()会一直等到数据写进 TCP 缓冲区才返回期间线程被占着。连接数一多线程池很快被耗尽新消息只能排队。第二种是大量空闲连接。每个连接占一条线程几千个连接就是几千条线程线程上下文切换的开销非常可观内存占用也不低。第三种是IO 与业务耦合。读取消息、处理业务、写回响应都在同一条链路里任何一步耗时都会拖累整条链路。虽然可以用Async打散但线程模型本身没变只是把阻塞挪了个地方。2.2 Reactive WebSocket 的非阻塞本质Reactive WebSocket 跑在WebFlux之上底层默认是Reactor Netty用的是事件循环Event Loop模型。它不再为每个连接分配一条线程而是用少量线程处理大量连接IO 操作全部异步化。核心区别在于消息的读写不再是“调用一个方法然后等它返回”而是“订阅一个数据流数据到了自然会被推过来”。在 Spring 的响应式 WebSocket 里一个会话被抽象成两个Flux入站消息是一个FluxWebSocketMessage出站消息是一个接收FluxWebSocketMessage的Publisher。你写 Handler 的时候本质上是在描述“入站流经过怎样的变换后变成出站流”而不是在写命令式的收发逻辑。这种声明式的写法配合 Reactor 的操作符天然支持背压——下游消费不过来时上游会被通知减速而不是无脑堆积。2.3 背压机制为什么在推送场景里是刚需背压这个词听起来玄乎其实生活里到处都是。想象一个水龙头往水桶里放水桶满了你还继续放水就溢出来。背压就是桶满了之后告诉水龙头“先关小点”。在消息推送里如果服务端产生消息的速度超过客户端消费的速度没有背压的话消息要么在服务端内存里堆积直到 OOM要么被直接丢弃。传统 WebSocket 的sendMessage是 fire-and-forget 式的你很难感知下游到底消费了多少。而 Reactive WebSocket 里出站流是一个Publisher下游网络层会通过request(n)告诉上游“我还能处理 n 条”。你可以用onBackpressureBuffer、onBackpressureDrop等操作符明确指定堆积或丢弃策略行为完全可控。对于行情推送、日志流这类场景这个能力非常关键。3. 环境搭建与依赖选型的关键决策3.1 Spring Boot 版本与 WebFlux 的绑定关系标题里明确是 Spring Boot 2.x这个范围其实挺宽2.0 到 2.7 之间 WebFlux 的 API 有过一些调整。我实测下来2.3 及以上版本在响应式 WebSocket 的稳定性上明显更好尤其是会话关闭时的资源回收逻辑。如果你还在 2.0 或 2.1建议至少升到 2.3。本文的示例基于 2.7.x 编写向下兼容到 2.3 基本没问题。有一点必须强调Reactive WebSocket 和传统的 spring-boot-starter-websocket 不能混用。前者依赖的是spring-boot-starter-webflux后者依赖的是spring-boot-starter-web。如果你两个都引了Spring Boot 会优先启动 Servlet 容器WebFlux 的自动配置不会生效Handler 根本注册不上。这是新手最容易踩的坑之一。3.2 依赖清单与排除项一个干净的响应式 WebSocket 项目pom.xml里核心依赖就两个dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-webflux/artifactId /dependency dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-websocket/artifactId /dependency注意第二个spring-boot-starter-websocket在响应式场景下提供的是spring-websocket里的响应式支持类比如WebSocketHandler、WebSocketSession这些接口它本身不会引入 Servlet 容器。但如果你项目里还残留着spring-boot-starter-web一定要排除掉否则容器冲突。dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId exclusions exclusion groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-tomcat/artifactId /exclusion /exclusions /dependency提示判断当前项目跑的是哪个栈最直接的办法是看启动日志。如果看到Netty started on port说明是 WebFlux如果看到Tomcat started on port那就是 Servlet 栈响应式 WebSocket 不会生效。3.3 为什么默认选 Reactor Netty 而不是 UndertowWebFlux 支持三种运行时Reactor Netty、Undertow、TomcatServlet 3.1 异步。响应式 WebSocket 在 Reactor Netty 上的支持最完整背压语义也最准确。Undertow 虽然也能跑但在 WebSocket 的背压传递上有一些已知的限制某些操作符的行为会和 Netty 下不一致。Tomcat 的异步 WebSocket 本质上还是 Servlet 模型响应式只是套了层壳性能优势不明显。所以除非你有明确的理由必须用 Undertow否则默认用 Reactor Netty它是spring-boot-starter-webflux的默认依赖不用额外配置。4. Handler 与配置类的完整实现4.1 一个最小可用的 ReactiveWebSocketHandler先看 Handler 的核心结构。响应式 WebSocket 的 Handler 需要实现WebSocketHandler接口重写handle方法Component public class EchoWebSocketHandler implements WebSocketHandler { Override public MonoVoid handle(WebSocketSession session) { FluxWebSocketMessage output session.receive() .map(msg - session.textMessage(echo: msg.getPayloadAsText())); return session.send(output); } }这段代码干了什么session.receive()拿到入站消息流map把每条消息转成一条新的文本消息session.send(output)把出站流写回客户端。整个方法返回一个MonoVoid表示“这个会话的处理过程”当流结束或出错时这个 Mono 才会完成。注意这里没有一行阻塞代码也没有手动关闭会话。会话的生命周期由框架管理流结束客户端断开时Mono自然完成资源被回收。这就是声明式写法的好处——你描述的是数据流向而不是控制流程。4.2 用 HandlerMapping 注册路由Handler 写好了还得告诉框架“哪个路径用哪个 Handler”。响应式栈里用的是SimpleUrlHandlerMapping配置类长这样Configuration public class WebSocketConfig { Bean public HandlerMapping webSocketHandlerMapping(EchoWebSocketHandler handler) { MapString, WebSocketHandler map new HashMap(); map.put(/ws/echo, handler); SimpleUrlHandlerMapping mapping new SimpleUrlHandlerMapping(); mapping.setUrlMap(map); mapping.setOrder(-1); return mapping; } Bean public WebSocketHandlerAdapter webSocketHandlerAdapter() { return new WebSocketHandlerAdapter(); } }两个 Bean 缺一不可。HandlerMapping负责路径到 Handler 的映射WebSocketHandlerAdapter负责把 WebSocket 握手请求适配到 Handler 上。setOrder(-1)是为了让这个映射优先于其他普通 HTTP 映射避免路径被别的 Handler 抢走。注意WebSocketHandlerAdapter在较新的版本里可以省略框架会自动配置。但如果你发现握手一直返回 404先检查这个 Bean 在不在。4.3 会话管理与广播的正确姿势单机 echo 只是热身真实场景里更常见的是广播——一个客户端发消息所有连接的客户端都收到。传统写法里我们会维护一个SetWebSocketSession遍历发送。响应式里如果还这么干就退回了阻塞老路。正确的做法是用一个Sinks.Many作为消息总线Component public class BroadcastHandler implements WebSocketHandler { private final Sinks.ManyString sink Sinks.many() .multicast().onBackpressureBuffer(); Override public MonoVoid handle(WebSocketSession session) { FluxWebSocketMessage output sink.asFlux() .map(session::textMessage); MonoVoid input session.receive() .map(WebSocketMessage::getPayloadAsText) .doOnNext(sink::tryEmitNext) .then(); return Mono.zip(input, session.send(output)).then(); } }这里有几个关键点。Sinks.many().multicast().onBackpressureBuffer()创建的是一个多播的、带缓冲的 sink所有订阅者共享同一条消息流。每个会话的handle方法里出站流订阅这个 sink入站流把收到的消息 emit 进 sink。Mono.zip把入站处理和出站发送两个Mono合并任何一个结束都会触发整体结束。tryEmitNext是 Reactor 3.4 之后推荐的发射方式它会返回一个EmitResult你可以根据结果判断是成功、失败还是被丢弃。相比老的emitNext它不会在失败时抛异常打断流更适合广播这种“尽力而为”的场景。5. 背压策略与消息发射的实战细节5.1 三种背压策略的取舍Sinks.many().multicast()后面可以接不同的背压策略选哪个取决于你的业务对消息丢失的容忍度。策略行为适用场景风险onBackpressureBuffer缓冲所有消息直到下游消费消息不能丢消费速度波动大缓冲无上限时可能 OOMonBackpressureBuffer(int)缓冲指定条数超出丢弃消息较重要但有容量上限超出部分静默丢失onBackpressureDrop下游没请求就丢弃行情、日志等可容忍丢失丢失无感知onBackpressureLatest只保留最新一条状态类推送如光标位置中间状态丢失我那个白板项目最后选的是onBackpressureBuffer(1024)因为光标位置可以丢但白板上的图形绘制指令不能丢。缓冲 1024 条是个经验值按单条消息平均 200 字节算最多占 200KB 内存可控。如果你的消息体很大这个数字要相应调小。5.2 消息发射失败的处理tryEmitNext返回的EmitResult有几种可能OK、FAIL_OVERFLOW、FAIL_CANCELLED、FAIL_TERMINATED、FAIL_ZERO_SUBSCRIBER。广播场景里最常见的是FAIL_OVERFLOW缓冲满了和FAIL_ZERO_SUBSCRIBER还没有订阅者。Sinks.EmitResult result sink.tryEmitNext(payload); if (result.isFailure()) { log.warn(消息发射失败: {}, 原因: {}, payload, result); // 根据业务决定是否重试、降级或丢弃 }不要忽略这个返回值。我见过有项目直接sink.emitNext(...)不检查结果缓冲溢出时抛异常把整个流终止了所有客户端连接全断。用tryEmitNext加上失败处理至少能保证单条消息的问题不会扩散成全局故障。5.3 会话关闭时的资源清理响应式 WebSocket 的会话关闭是自动的但如果你在 Handler 里持有外部资源比如数据库连接、定时任务需要在流结束时清理。用doFinally操作符return Mono.zip(input, session.send(output)) .doFinally(signal - { log.info(会话关闭, 信号: {}, signal); // 清理该会话相关的资源 }) .then();signal的类型有ON_COMPLETE、ON_ERROR、ON_CANCEL三种。正常断开是ON_COMPLETE异常是ON_ERROR客户端强制取消是ON_CANCEL。根据信号类型可以做不同的清理逻辑比如异常断开时记录日志便于排查。6. 常见问题排查与避坑经验6.1 握手 404 或 426 的排查路径握手失败是最常见的问题表现是客户端连不上服务端日志里可能什么都没有。按下面顺序排查确认容器是 Netty。看启动日志有没有Netty started。如果是 Tomcat说明spring-boot-starter-web没排除干净。确认 HandlerMapping 的 order。如果 order 是默认值可能被其他映射拦截。设成 -1 或更小的值。确认路径匹配。SimpleUrlHandlerMapping的路径是精确匹配/ws/echo和/ws/echo/是两个不同的路径。确认 WebSocketHandlerAdapter 存在。缺了它握手请求不会被适配到 Handler。426 错误通常是 Upgrade 头没被正确处理多半是容器不对或者有过滤器拦截了请求。6.2 消息发送后客户端收不到的几种原因有时候握手成功客户端也能发消息但收不到服务端的推送。常见原因有三个。第一是出站流没有订阅。响应式是惰性的session.send(output)返回的Mono如果不被订阅什么都不会发生。确保它在Mono.zip或直接返回的链路里。第二是sink 没有订阅者。multicast类型的 sink 在没有订阅者时发射的消息会被丢弃返回FAIL_ZERO_SUBSCRIBER。如果客户端刚连上还没订阅出站流此时广播的消息就丢了。解决办法是用replay类型的 sink 或者加一个初始订阅。第三是背压策略导致消息被丢。如果用了onBackpressureDrop下游没及时请求的消息会被静默丢弃。调试时可以先换成onBackpressureBuffer确认是不是策略问题。6.3 内存泄漏的定位与预防响应式 WebSocket 的内存泄漏通常和 sink 的生命周期有关。如果 sink 是全局单例而订阅它的会话流没有正确结束订阅关系会一直挂着导致会话对象无法回收。预防手段有两个。一是确保每个会话的handle方法返回的Mono在会话断开时能完成这样订阅会自动取消。二是定期检查 sink 的订阅者数量如果发现只增不减说明有会话没被正确清理。可以用sink.asFlux().doOnSubscribe(...)和doOnCancel(...)打点统计。实操心得我在调试阶段会在doFinally里打印当前活跃会话数用一个AtomicInteger计数。正常运行时这个数字应该和实际连接数吻合如果持续上涨基本可以确定有泄漏。6.4 与前端配合的几个细节前端用原生WebSocketAPI 连接响应式 WebSocket 时有几个地方容易出问题。一是心跳响应式 WebSocket 默认没有内置心跳长时间空闲的连接可能被中间设备断开需要自己实现 ping/pong。二是重连客户端断开后要能自动重连重连时注意不要重复订阅导致消息重复。三是消息格式建议统一用 JSON文本消息和二进制消息分开处理避免解析歧义。7. 性能调优与生产环境注意事项7.1 线程模型与 Event Loop 配置Reactor Netty 默认的 Event Loop 线程数是CPU 核心数 * 2。对于 WebSocket 这种长连接场景这个数量通常够用因为每个连接不占线程。但如果你的 Handler 里有耗时的业务逻辑比如查数据库一定要把它切到独立的调度器上否则会阻塞 Event Loop影响所有连接。session.receive() .publishOn(Schedulers.boundedElastic()) .flatMap(msg - someBlockingService.process(msg)) .map(session::textMessage)Schedulers.boundedElastic()是专门为阻塞任务准备的调度器线程数有上限不会无限增长。注意publishOn和subscribeOn的区别publishOn影响下游操作符的执行线程subscribeOn影响上游的订阅行为。在 WebSocket 场景里通常用publishOn就够了。7.2 连接数上限与资源限制Reactor Netty 对单个连接的读写缓冲区有默认限制消息体过大时可能触发DataBufferLimitException。可以通过配置调整spring: codec: max-in-memory-size: 2MB这个值默认是 256KB如果你的消息体可能超过这个大小记得调大。但也不要无脑调成几百 MB内存占用会失控。合理做法是根据业务消息的最大尺寸设置留一倍余量。7.3 监控指标该看哪些生产环境里响应式 WebSocket 有几个关键指标值得监控活跃连接数、消息发射失败率、缓冲队列长度、Event Loop 的阻塞时间。Spring Boot Actuator 配合 Micrometer 可以拿到 Reactor Netty 的连接指标但消息层面的指标需要自己埋点。我一般会在 sink 的发射处和会话的doFinally处打点用 Micrometer 的Counter和Gauge记录接入 Prometheus 后能直观看到趋势。8. 写在最后的一点个人体会从传统 WebSocket 迁到响应式 WebSocket最大的门槛不是 API而是思维方式的转变。刚开始写的时候总想着“先拿到 session再遍历发送”写着写着就发现处处别扭。后来强迫自己用“流”的视角去看问题——入站是一个流出站是一个流业务逻辑是流之间的变换——代码一下子就顺了。另一个体会是不要为了响应式而响应式。如果你的连接数就几十个消息量也不大传统 WebSocket 完全够用硬上响应式只会增加维护成本。响应式的价值在连接数上千、消息推送频繁、对背压有要求的场景里才真正体现出来。选型的时候先想清楚自己的瓶颈在哪再决定要不要上这套东西。最后分享一个小技巧调试响应式 WebSocket 时在关键操作符后面加.log()Reactor 会把该节点收到的所有信号onNext、onComplete、onError、request、cancel打印出来。背压问题、订阅问题、流提前终止的问题看日志基本都能定位。这个习惯帮我省了大量猜测的时间。