Spring 5 Reactive WebSockets：推荐使用答案

【问题标题】：Spring 5 Reactive WebSockets: recommended useSpring 5 Reactive WebSockets：推荐使用
【发布时间】：2018-03-17 14:02:03
【问题描述】：

我一直在学习一些关于 Spring 5 WebFlux、反应式编程和 websockets 的知识。我看过 Josh Long 的 Spring Tips: Reactive WebSockets with Spring Framework 5。通过 WebSocket 连接将数据从服务器发送到客户端的代码使用 Spring Integration IntegrationFlow 发布到 PublishSubcribeChannel，PublishSubcribeChannel 订阅了自定义 MessageHandler 接收消息，将其转换为对象，然后转换为 Json 并从提供给 Flux.create() 的回调中发送到 FluxSink，该回调用于发送到 WebSocketConnection。

我想知道是否推荐使用 IntegrationFlow 和 PublishSubscribeChannel 将事件从后台进程推送到客户端，或者在这个特定示例中这是否更方便（监控文件系统）。我想如果你可以控制后台进程，你可以让它直接发送到FluxSink？

我正在考虑类似于以下的用例：

一个机器学习过程，其进度受到监控
发送给玩家的游戏世界状态更新
聊天室/团队协作软件
...

【问题讨论】：

对这个问题的答案也很感兴趣。 Reactor 的 Flux 不是消息总线，所以也许这就是 Channel 更合适的原因。
对答案也很感兴趣

标签： spring websocket reactive-programming spring-webflux

【解决方案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/

【讨论】：

【解决方案2】：

从这个问题开始，Spring 引入了RSocket support - 您可能会认为它就像 Spring MVC 中现有的 WebSocket STOMP 支持一样，但更强大和高效，支持背压和协议级别的高级通信模式。

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

【讨论】：

RSocket 非常棒，因为它将网络作为一等公民引入了响应式范式。但是，RSocket 和 Websockets 并不相互排斥。 RSocket 是在 TCP 或 Websockets 上运行的协议。 WebSocket 协议像 HTTP/2 一样运行在 OSI 模型的第 7 层，而 RSocket 运行在 OSI 第 5/6 层。
不幸的是，用于 javascript 的 RSocket websocket 客户端目前只有 0.0.19 并且已经 10 个月大。见npmjs.com/package/rsocket-websocket-client
Rsocket 本身只有 0.2