Spring反应器在不突变的情况下对输入进行配料

Question

我正在尝试批量记录不断发出从一个流源（Kafka）和调用我的服务在一批100。

作为输入，我得到的是一个单一的记录。我正在尝试什么是最好的方式来实现它的反应方式使用Spring反应堆，而不需要有一个突变和锁定在管道外。

 Mono.just(input)
                .subscribe(i -> {
                     batches.add(input);
                     if(batches.size() >= 100) {
                         // Invoke another reactive pipeline.
                         // Clear the batch (requires locking in order to be thread safe).
                     }
                });

什么是最好的方法，以实现分批在一个流式源使用反应器。

Answer 1

使用flux.buffer或flux.bufferTimeout可以将固定数量的元素收集到列表中

StepVerifier.create(
      Flux.range(0, 1000)
          .buffer(100)
   )
   .expectNextCount(10)
   .expectComplete()
   .verify()

在情况下，当输入是一个单个值时，假设like调用带有参数的方法：

public void invokeMe(String element);

您可以采用unicastprocessor技术，将所有数据传送到该处理器，这样它就可以处理批处理

class Batcher {

   final UnicastProcessor processor = UnicastProcessor.create();

   public void invokeMe(String element) {
       processor.sink().next(element);
       // or Mono.just(element).subscribe(processor);
   }


   public Flux<List<String>> listen() {
       return processor.bufferTimeout(100, Duration.ofSeconds(5));
   }
}

Batcher batcher = new Batcher();

StepVerifier.create(
      batcher.listen()
   )
   .then(() -> Flux.range(0, 1000)
                   .subscribe(i -> batcher.invokeMe("" + i)))
   .expectNextCount(10)
   .thenCancel()
   .verify()

请注意，unicastporcessor只允许一个订阅者，因此当批处理结果中有一个相关方和许多数据生产者时，它将对模型有用。如果您的订阅服务器和生产者一样多，您可能希望使用下一个处理器-->DirectProcessor、TopicProcessor、WorkerQueueProcessor。要了解更多关于反应堆处理器的信息，请访问以下链接