Webflux，使用 Websocket 如何防止订阅两次 reactive redis 消息传递操作

Question

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

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

想知道是否有一种方法可以重复使用对消息传递主题的相同订阅，从而防止多个 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);
  }
 }

消息处理服务

public class 消息服务{ private final ReactiveRedisOperations reactiveRedisOperations;

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


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

}

提前致谢。

Answer 1

而不是使用 ReactiveRedisOperations，MessageListener 是这里的方法。可以注册一次监听器，使用下面的监听器。

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

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

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

Instructions on how to use MessageListener.

Webflux，使用 Websocket 如何防止订阅两次 reactive redis 消息传递操作

Webflux, with Websocket how to prevent subscribing twice of reactive redis messaging operation

redis

spring-webflux

spring-websocket