小贝子编程

对使用springboot webflux创建的多个反应性WebSocketSession的广播不起作用



以下是场景：

  1. 我创建了一个反应堆kafka接收器
  2. kafka接收器消耗的数据发布到WebSocketHanlder
  3. WebSocketHanlder使用SimpleUrlHandlerMapping映射到URL
  4. URL模式是api/v1/ws/｛ID｝，我希望根据URI中使用的不同ID创建多个WebSocketSession，这些ID由单个WebSocketHanlder管理，这实际上正在发生
  5. 但是，当发布来自kafka接收器的数据时，只有第一个创建的WebSocketSession会接收它，而所有其他WebSocketSessions都不会接收数据
  6. 我使用的是带有启动器tomcat的弹簧靴2.6.3

如何将数据发布到创建的所有WebSocketSession我的代码：

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中的handle(…(方法定义更改为以下内容，但来自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通量：

  1. reactor.Flux.Map.42-onSubscribe(FluxMapSubscriber
  2. reactor.Flux.Map.46-onSubscribe(FluxMap.MapSubscriber(

但当从kafka主题读取数据时，只有一个订户接收到数据：

  • reactor.Flux.Map.42-onNext({"问候"："你好"，"名字"："Prashant〃}(

这是Webflux API本身的bug吗？

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

问题我在WebSocketHandler handle((方法中使用Flux(从KafkaReceiver获得(的方式不正确。对于从多个客户端请求创建的每个websocket会话，将调用handle方法。因此，为KafkaReceiver.create((.rereceive((创建了多个Flux对象。其中一个Flux从KafkaReceiver读取数据，但其他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();
}

解决方案确保仅为一种方法是在WebSocketHandler(或KAfkaOnumer类(的构造函数中创建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;
}

相关内容

  • 没有找到相关文章

最新更新


热门标签：

javascript python java c# php android html jquery c++ css ios sql mysql arrays asp.net json python-3.x ruby-on-rails .net sql-server django objective-c excel regex ruby linux ajax iphone xml vba spring asp.net-mvc database wordpress string postgresql wpf windows xcode bash git oracle list vb.net multithreading eclipse algorithm macos powershell visual-studio image forms numpy scala function api selenium