广播到多个使用springboot web通创建的响应式WebSocketSsession不起作用

提问者：小点点

广播到多个使用springboot web通创建的响应式WebSocketSsession不起作用

以下是场景：

我创建一个Reactorkafka接收器
从接收器kafka数据被发布到WebSocketHanlder
WebSocketHanlder使用SimpleUrlHandlerMaps映射到URL
URL模式是api/v1/ws/{ID}，我希望基于URI中使用的不同ID创建多个WebSocketSession，这些ID由单个WebSocketHanlder管理，这实际上正在发生
但是当来自接收者kafka数据被发布时，只有第一个创建的WebSocketSession接收它，所有其他WebSocketSession都不会接收数据
我正在使用spring-boot 2.6.3和starter-tomcat

如何将数据发布到创建My Code的所有WebSocketSsession：

Web套接字处理程序的配置


@Configuration
@Slf4j
public class OneSecPollingWebSocketConfig
{
   private OneSecPollingWebSocketHandler oneSecPollingHandler;

   @Autowired
   public OneSecPollingWebSocketConfig(OneSecPollingWebSocketHandler oneSecPollingHandler)
   {
      this.oneSecPollingHandler = oneSecPollingHandler;
   }

   @Bean
   public HandlerMapping webSocketHandlerMapping()
   {
      log.info("onesecpolling websocket configured");
      Map<String, WebSocketHandler> handlerMap = new HashMap<>();
      handlerMap.put(WEB_SOCKET_ENDPOINT, oneSecPollingHandler);
      SimpleUrlHandlerMapping mapping = new SimpleUrlHandlerMapping();
      mapping.setUrlMap(handlerMap);
      mapping.setOrder(1);
      return mapping;
   }
}

WebSocket HAndler代码


@Component
@Slf4j
public class OneSecPollingWebSocketHandler implements WebSocketHandler
{
   private ObjectMapper objectMapper;
   private OneSecPollingKafkaConsumerService oneSecPollingKafkaConsumerService;
   private Map<String, WebSocketSession> wsSessionsByUserSessionId = new HashMap<>();

   @Autowired
   public OneSecPollingWebSocketHandler(ObjectMapper objectMapper, OneSecPollingKafkaConsumerService oneSecPollingKafkaConsumerService)
   {
      this.objectMapper = objectMapper;
      this.oneSecPollingKafkaConsumerService = oneSecPollingKafkaConsumerService;
   }

   @Override
   public Mono<Void> handle(WebSocketSession webSocketSession)
   {
      Many<String> sink = Sinks.many().multicast().onBackpressureBuffer(Queues.SMALL_BUFFER_SIZE, false);
      wsSessionsByUserSessionId.put(getUserPollingSessionId(webSocketSession), webSocketSession);
      sinkSubscription(webSocketSession, sink);
      Mono<Void> output = webSocketSession.send(sink.asFlux().map(webSocketSession::textMessage)).doOnSubscribe(subscription ->
      {
      });
      return Mono.zip(webSocketSession.receive().then(), output).then();
   }

   public void sinkSubscription(WebSocketSession webSocketSession, Many<String> sink)
   {
      log.info("number of sessions; {}", wsSessionsByUserSessionId.size());
      oneSecPollingKafkaConsumerService.getTestTopicFlux().doOnNext(record ->
      {
         //log.info("record: {}", record);
         sink.tryEmitNext(record.value());
         record.receiverOffset().acknowledge();
      }).subscribe();
   }

   public String getOneSecPollingTopicRecord(ReceiverRecord<Integer, String> record, WebSocketSession webSocketSession)
   {
      String lastRecord = record.value();
      log.info("record to send: {} : webSocketSession: {}", record.value(), webSocketSession.getId());
      record.receiverOffset().acknowledge();
      return lastRecord;     
   }

   public String getUserPollingSessionId(WebSocketSession webSocketSession)
   {
      UriTemplate template = new UriTemplate(WEB_SOCKET_ENDPOINT);
      URI uri = webSocketSession.getHandshakeInfo().getUri();
      Map<String, String> parameters = template.match(uri.getPath());
      String userPollingSessionId = parameters.get("userPollingSessionId");
      return userPollingSessionId;
   }
}

Kafka接收机

@Service
@Slf4j
public class OneSecPollingKafkaConsumerService
{
   private String bootStrapServers;

   @Autowired
   public OneSecPollingKafkaConsumerService(@Value("${bootstrap.servers}") String bootStrapServers)
   {
      this.bootStrapServers = bootStrapServers;
   }

   private ReceiverOptions<Integer, String> getRecceiverOPtions()
   {
      Map<String, Object> consumerProps = new HashMap<>();
      consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootStrapServers);
      //consumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, "reactive-consumer");
      consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "onesecpolling-group");
      consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
      consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
      //consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
      //consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

      ReceiverOptions<Integer, String> receiverOptions = ReceiverOptions
         .<Integer, String> create(consumerProps)
         .subscription(Collections.singleton("HighFrequencyPollingKPIsComputedValues"));

      return receiverOptions;
   }

   public Flux<ReceiverRecord<Integer, String>> getTestTopicFlux()
   {
      return createTopicCache();
   }

   private Flux<ReceiverRecord<Integer, String>> createTopicCache()
   {
      Flux<ReceiverRecord<Integer, String>> oneSecPollingMessagesFlux = KafkaReceiver.create(getRecceiverOPtions())
         .receive()
         .delayElements(Duration.ofMillis(500));
      return oneSecPollingMessagesFlux;
   }
}

POM依赖关系

<dependencies>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-devtools</artifactId>
    </dependency>
    <!-- 
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-security</artifactId>
    </dependency> 
    -->
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-webflux</artifactId>
    </dependency>
    <dependency>
      <groupId>io.projectreactor.kafka</groupId>
      <artifactId>reactor-kafka</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.cloud</groupId>
      <artifactId>spring-cloud-stream</artifactId>
    </dependency>
    <dependency>
      <groupId>org.projectlombok</groupId>
      <artifactId>lombok</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-actuator</artifactId>
    </dependency>
    <!-- This is breaking WebFlux 
      <dependency>
        <groupId>org.springdoc</groupId>
        <artifactId>springdoc-openapi-ui</artifactId>
        <version>${springdoc.version}</version>
      </dependency>
      -->
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-aop</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-tomcat</artifactId>
      <scope>provided</scope>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-test</artifactId>
      <scope>test</scope>
    </dependency>
    <dependency>
      <groupId>io.projectreactor</groupId>
      <artifactId>reactor-test</artifactId>
      <scope>test</scope>
    </dependency>
    <dependency>
      <groupId>org.springframework.cloud</groupId>
      <artifactId>spring-cloud-stream</artifactId>
      <classifier>test-binder</classifier>
      <type>test-jar</type>
      <scope>test</scope>
    </dependency>
    <!-- <dependency>
      <groupId>org.springframework.security</groupId>
      <artifactId>spring-security-test</artifactId>
      <scope>test</scope>
    </dependency> -->
  </dependencies>

我还尝试将WebSocketHanlder中的句柄（…）方法定义更改为以下，但kafka中的数据仍然只推送到一个websocket会话：

@Override
   public Mono<Void> handle(WebSocketSession webSocketSession)
   {
      Mono<Void> input = webSocketSession.receive().then();
      Mono<Void> output = webSocketSession.send(oneSecPollingKafkaConsumerService.getTestTopicFlux().map(ReceiverRecord::value).map(webSocketSession::textMessage));
      return Mono.zip(input, output).then();
   }

此外，我尝试了以下内容：

public Mono<Void> handle(WebSocketSession webSocketSession)
   {      
      Mono<Void> input = webSocketSession.receive()
         .doOnSubscribe(subscribe -> log.info("sesseion created sessionId:{}:userId:{};sessionhash:{}",
            webSocketSession.getId(),
            getUserPollingSessionId(webSocketSession),
            webSocketSession.hashCode()))
         .then();
      Flux<String> source = oneSecPollingKafkaConsumerService.getTestTopicFlux().map(record -> getOneSecPollingTopicRecord(record, webSocketSession)).log();
      Mono<Void> output = webSocketSession.send(source.map(webSocketSession::textMessage)).log();
      return Mono.zip(input, output).then().log();
   }

我启用了log（）并得到以下输出：

20:09:22.652 [http-nio-8080-exec-9] INFO  c.m.e.w.p.i.w.v.OneSecPollingWebSocketHandler - sesseion created sessionId:a:userId:124;sessionhash:1974799413
20:09:22.652 [http-nio-8080-exec-9] INFO  reactor.Flux.RefCount.41 - | onSubscribe([Fuseable] FluxRefCount.RefCountInner)
20:09:22.652 [http-nio-8080-exec-9] INFO  reactor.Flux.Map.42 - onSubscribe(FluxMap.MapSubscriber)
20:09:22.652 [http-nio-8080-exec-9] INFO  reactor.Flux.Map.42 - request(1)
20:09:22.652 [http-nio-8080-exec-9] INFO  reactor.Flux.RefCount.41 - | request(32)
20:09:22.659 [http-nio-8080-exec-9] INFO  reactor.Mono.FromPublisher.43 - onSubscribe(MonoNext.NextSubscriber)
20:09:22.659 [http-nio-8080-exec-9] INFO  reactor.Mono.FromPublisher.43 - request(unbounded)
20:09:25.942 [http-nio-8080-exec-10] INFO  reactor.Mono.IgnorePublisher.48 - onSubscribe(MonoIgnoreElements.IgnoreElementsSubscriber)
20:09:25.942 [http-nio-8080-exec-10] INFO  reactor.Mono.IgnorePublisher.48 - request(unbounded)
20:09:25.942 [http-nio-8080-exec-10] INFO  c.m.e.w.p.i.w.v.OneSecPollingWebSocketHandler - sesseion created sessionId:b:userId:123;sessionhash:1582184236
20:09:25.942 [http-nio-8080-exec-10] INFO  reactor.Flux.RefCount.45 - | onSubscribe([Fuseable] FluxRefCount.RefCountInner)
20:09:25.942 [http-nio-8080-exec-10] INFO  reactor.Flux.Map.46 - onSubscribe(FluxMap.MapSubscriber)
20:09:25.942 [http-nio-8080-exec-10] INFO  reactor.Flux.Map.46 - request(1)
20:09:25.942 [http-nio-8080-exec-10] INFO  reactor.Flux.RefCount.45 - | request(32)
20:09:25.947 [http-nio-8080-exec-10] INFO  reactor.Mono.FromPublisher.47 - onSubscribe(MonoNext.NextSubscriber)
20:09:25.949 [http-nio-8080-exec-10] INFO  reactor.Mono.FromPublisher.47 - request(unbounded)
20:10:00.880 [reactive-kafka-onesecpolling-group-11] INFO  reactor.Flux.RefCount.41 - | onNext(ConsumerRecord(topic = HighFrequencyPollingKPIsComputedValues, partition = 0, leaderEpoch = null, offset = 474, CreateTime = 1644071999871, serialized key size = -1, serialized value size = 43, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = {"greeting" : "Hello", "name" : "Prashant"}))
20:10:01.387 [parallel-5] INFO  reactor.Flux.Map.42 - onNext({"greeting" : "Hello", "name" : "Prashant"})
20:10:01.389 [parallel-5] INFO  reactor.Flux.Map.42 - request(1)

在这里，我们可以看到我们有2个订阅者kafka通量：

reactor.Flux.Map.42-onSubscribe（FlexMap. MapSubscriber
reactor.Flux.Map.46-onSubscribe（FlexMap. MapSubscriber）

但是当数据从kafka主题中读取时，它只被一个订阅者接收：

reactor.Flux.Map.42-onNext（{"问候"："你好"，"姓名"："Prashant"}）

它是否是WebFlowAPI本身的bug？

共1个答案

匿名用户

我已经找到了问题和解决方案。

问题我在WebSocketHandler句柄（）方法中使用Flux（从KafkaReccher获得）的方式不正确。对于从多个客户端请求创建的每个websocket会话，句柄方法都会被调用。因此，创建了KafkaReceiver. create（）.接收（）的多个Flux对象。其中一个Flux从KafkaReccher读取数据，但其他Flux对象未能做到这一点。

public Mono<Void> handle(WebSocketSession webSocketSession)
   {      
      Mono<Void> input = webSocketSession.receive()
         .doOnSubscribe(subscribe -> log.info("sesseion created sessionId:{}:userId:{};sessionhash:{}",
            webSocketSession.getId(),
            getUserPollingSessionId(webSocketSession),
            webSocketSession.hashCode()))
         .then();
      **Flux<String> source = oneSecPollingKafkaConsumerService.getTestTopicFlux()**.map(record -> getOneSecPollingTopicRecord(record, webSocketSession)).log();
      Mono<Void> output = webSocketSession.send(source.map(webSocketSession::textMessage)).log();
      return Mono.zip(input, output).then().log();
   }

解决方案确保只为KafkaReceiver. create（）.接收（）创建一个Flux。这样做的一种方法是在WebSocketHandler（或KAfkaCOnsumer类）的构造函数中创建Flux

private final Flux<String> source;

   @Autowired
   public OneSecPollingWebSocketHandler(OneSecPollingKafkaConsumerService oneSecPollingKafkaConsumerService)
   {
      source = oneSecPollingKafkaConsumerService.getOneSecPollingTopicFlux().map(r -> getOneSecPollingTopicRecord(r));
   }

   @Override
   public Mono<Void> handle(WebSocketSession webSocketSession)
   {
      // add usersession id as session attribute
      Mono<Void> input = getInputMessageMono(webSocketSession);
      Mono<Void> output = getOutputMessageMono(webSocketSession);
      return Mono.zip(input, output).then().log();
   }

   private Mono<Void> getOutputMessageMono(WebSocketSession webSocketSession)
   {
      Mono<Void> output = webSocketSession.send(source.map(webSocketSession::textMessage)).doOnError(err -> log.error(err.getMessage())).doOnTerminate(() ->
      {
         log.info("onesecpolling session terminated;{}", webSocketSession.getId());
      }).log();
      return output;
   }

   private Mono<Void> getInputMessageMono(WebSocketSession webSocketSession)
   {
      Mono<Void> input = webSocketSession.receive().doOnSubscribe(subscribe ->
      {
         log.info("onesecpolling session created sessionId:{}:userId:{}", webSocketSession.getId(), getUserPollingSessionId(webSocketSession));
      }).then();
      return input;
   }

   private String getOneSecPollingTopicRecord(ReceiverRecord<Integer, String> record)
   {
      String lastRecord = record.value();
      record.receiverOffset().acknowledge();
      return lastRecord;
   }

   private String getUserPollingSessionId(WebSocketSession webSocketSession)
   {
      UriTemplate template = new UriTemplate(WEB_SOCKET_ENDPOINT);
      URI uri = webSocketSession.getHandshakeInfo().getUri();
      Map<String, String> parameters = template.match(uri.getPath());
      String userPollingSessionId = parameters.get(WEB_SOCKET_ENDPOINT_USERID_SUBPATH);
      return userPollingSessionId;
   }