Webflux，使用Websocket如何防止订阅两次反应式redis消息操作答案

【问题标题】：Webflux, with Websocket how to prevent subscribing twice of reactive redis messaging operationWebflux，使用Websocket如何防止订阅两次反应式redis消息操作
【发布时间】：2021-12-31 04:31:50
【问题描述】：

我在 webflux 上有一个使用 redis 消息传递操作的 websocket 实现。它的作用是监听主题并通过 websocket 端点返回值。

我遇到的问题是每次用户通过 websocket 向端点发送消息时，似乎都进行了全新的 redis 订阅，导致 redis 消息主题上的订阅者积累，并且 websocket 响应随着数量的增加而增加redis 主题消息订阅次数也是如此（示例用户发送 3 条消息，redis 主题订阅增加到 3 次，websocket 连接响应 3 次）。

想知道是否有办法重用消息主题的相同订阅，这样可以防止多个 redis 主题订阅。

我使用的代码如下：

Websocket 处理程序

public class SendingMessageHandler implements WebSocketHandler {
  private final Gson gson = new Gson();

  private final MessagingService messagingService;

  public SendingMessageHandler(MessagingService messagingService) {
      this.messagingService = messagingService;
  }

  @Override
  public Mono<Void> handle(WebSocketSession session) {
      Flux<WebSocketMessage> stringFlux = session.receive()
              .map(WebSocketMessage::getPayloadAsText)
              .flatMap(inputData ->
                      messagingService.playGame(inputData)
                              .map(data ->
                                      session.textMessage(gson.toJson(data))
                              )
              );

      return session.send(stringFlux);
  }
 }

消息处理服务

公共类消息服务{ private final ReactiveRedisOperations reactiveRedisOperations;

  public MessagingService(ReactiveRedisOperations<String, GamePubSub> reactiveRedisOperations) {
      this.reactiveRedisOperations = reactiveRedisOperations;
  }


  public Flux<Object> playGame(UserInput userInput){
      return reactiveRedisOperations.listenTo("TOPIC_NAME");
  }

}

提前谢谢你。

【问题讨论】：

标签： redis spring-webflux spring-websocket

【解决方案1】：

而不是使用ReactiveRedisOperations，MessageListener 是这里的方式。您可以注册一次监听器，并将以下内容用作监听器。

data -> session.textMessage(gson.toJson(data))

注册应该只在连接开始时发生一次。您可以覆盖 SendingMessageHandler 的 void afterConnectionEstablished(WebSocketSession session) 来完成此操作。这样，每个新的 Websocket 连接、每条消息都会创建一个新订阅。

另外，不要忘记覆盖afterConnectionClosed，并取消订阅redis主题，并清理其中的监听器。

Instructions on how to use MessageListener.

【讨论】：