Spring 5反应式WebSockets：推荐使用

Question

我最近学习了一些关于 Spring 5 WebFlux、反应式编程和websocket的内容。我看了Josh Long的Spring Tips: Reactive WebSockets with Spring Framework 5。发送从服务器通过 WebSocket 连接到客户端的数据的代码使用了 Spring Integration中的IntegrationFlow，它发布到一个具有自定义MessageHandler订阅者的PublishSubcribeChannel，该处理程序将消息转换为对象，然后将其转换为Json并发出到FluxSink中, 由传递给Flux.create()的回调函数使用，这用于发送至WebSocketConnection。

我想知道在后台进程向客户端推送事件时，是否建议采用IntegrationFlow和PublishSubscribeChannel，或者这仅仅是在这个特定的例子中更加方便（监视文件系统）。如果您可以控制后台进程，我认为直接将其发射到FluxSink中可能更好？

我正在考虑类似以下的使用情况：

• 监视机器学习过程的进度
• 将游戏世界状态更新发送给玩家
• 聊天室/团队协作软件
• ...

- herman

---

非常感兴趣了解这个问题的答案。Reactor的Flux不是消息总线，所以可能这就是为什么Channel更合适的原因。 - Gorazd Rebolj

非常感兴趣，希望能得到答案。 - USS-Montana

我并不认为没有理由不直接将后台事件发布到FluxSink中。从实现的角度来看，这似乎要简单得多。 - Shy Albatross

Answer 1

过去我做的一件行之有效的事情是创建一个实现 WebSocketHandler 接口的 Spring 组件：

@Component
public class ReactiveWebSocketHandler implements WebSocketHandler {

然后在handle方法中，Spring注入WebSocketSession对象

@Override
public Mono<Void> handle(WebSocketSession session) {

然后创建一个或多个Flux响应式发布者，用于向客户端发送消息(WebSocketMessage)。

    final var output = session.send(Flux.merge(flux1, flux2));

接下来，您可以将传入和传出的Flux对象压缩到一个Mono中，然后Spring会处理它。

        return Mono.zip(incomingWebsocketMsgResponse.getWebSocketMsgFlux().then(),
                outputWithErrorMsgs)
                .then();

示例：https://howtodoinjava.com/spring-webflux/reactive-websockets/

该示例介绍了如何使用Spring WebFlux实现反应式WebSocket。它解释了如何创建一个WebSocket服务器和客户端，并演示了如何使用Java代码实现基本的发送和接收操作。此外，该示例还涵盖了如何处理错误以及如何实现心跳检测功能。

- anataliocs

Answer 2

自从这个问题提出来后，Spring引入了RSocket支持 - 你可以将其视为Spring MVC中WebSocket STOMP支持的升级版，但更加强大和高效，在协议层面支持背压和先进的通信模式。

对于你提到的用例，我建议使用RSocket，因为你将获得一个强大的编程模型，其中包括@MessageMapping和Spring中所有预期的支持（JSON和CBOR的编解码器、安全性等）。

- Brian Clozel

---

RSocket非常棒，因为它将网络作为响应式编程范式的一等公民引入其中。然而，RSocket和Websockets并不是互斥的。RSocket是在TCP或Websockets上运行的协议。WebSocket协议在OSI模型的第7层上运行，就像HTTP/2一样，而RSocket则在OSI模型的第5/6层上运行。 - anataliocs

不幸的是，RSocket websocket客户端适用于javascript目前仅为0.0.19版本，而且已经有10个月没有更新了。请参见https://www.npmjs.com/package/rsocket-websocket-client。 - herman

Rsocket本身只处于0.2版本。 - anataliocs