Spring Reactive WebSocketClient使用Sinks出现阻塞超时异常问题

问题根因 #

你遇到的阻塞本质是两个未完成的流互相等待导致的：

• 首先你的Sinks.Many<Message>配置了replay缓存2条数据，但你没有在收到全部预期消息后主动调用emitComplete发射完成信号。而测试中StepVerifier.create(sinks.asFlux()).expectComplete().verify()会一直等待Flux的完成信号，在信号到达前会无限阻塞。
• 其次你客户端的WebSocketHandler返回值是sendMono.then(receiveMono)，而session.receive()流只有在WebSocket连接主动关闭时才会终止，你测试逻辑中没有主动断开连接的代码，因此整个client.execute返回的Mono永远不会完成，无论是subscribe还是加超时的block都会因为等不到结束信号而卡住或抛超时异常。

至于Spring Boot 2.4版本能正常运行的原因：Spring Boot 2.4对应的Reactor版本中ReplayProcessor默认存在废弃的自动终止逻辑，同时旧版Spring WebSocket的客户端实现会在测试上下文销毁时主动断开连接，两个因素叠加才让你的旧代码能正常运行，属于非预期的兼容表现。
而你换成List后逻辑不需要等待流完成，只要消息存入列表断言就可以直接结束，自然不会阻塞。

解决方案 #

可以任选以下一种方式修复：

• 给Sink添加完成触发逻辑：在客户端接收消息的doOnNext中添加计数逻辑，收到2条预期消息后主动调用sinks.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST)，同时主动调用session.close()关闭WebSocket连接。
• 调整StepVerifier逻辑：不需要等待流完成，将expectComplete()替换为expectNextCount(2).thenCancel()，验证到2条消息后直接取消订阅，不需要等待完成信号。
• 测试末尾添加主动关闭逻辑：StepVerifier验证通过后，主动调用客户端的关闭方法终止WebSocket连接，释放阻塞的流。

内容的提问来源于stack exchange，提问作者Hantsy