使用 reactor Kafka 从主题读取并将消息批量写入 REST 端点答案

【问题标题】：Read from topic and write the messages in batches to a REST endpoint using reactor Kafka使用 reactor Kafka 从主题读取并将消息批量写入 REST 端点
【发布时间】：2021-09-10 00:23:49
【问题描述】：

我正在开发一个使用反应式 Kafka 的项目，该项目使用来自 Kafka 主题的消息并将消息批量发布到 REST 端点。我被困在批处理部分并将该批处理发送到端点。我需要从主题中读取 N 条消息（这里的 N 条是可配置的），然后将这 N 条消息发送到 REST 端点。如何使用 reactor Kafka 读取 N 条消息？我查看了https://projectreactor.io/docs/kafka/release/reference/#_overview 中的示例，但找不到与我的问题类似的示例。任何有关解决此问题的建议都会非常有帮助。

这是我目前阅读主题消费消息的代码

@Slf4j
@Service
public class Service implements CommandLineRunner {
    @Autowired
    @Qualifier("KafkaConsumerTemplate")
    public ReactiveKafkaConsumerTemplate<String, String> KafkaConsumerTemplate;


    public Flux<String> consume() {
        return KafkaConsumerTemplate.receiveAutoAck()
                .doOnNext(consumerRecord -> log.info("received key={}, value={} from topic={}, offset={}",
                        consumerRecord.key(),
                        consumerRecord.value(),
                        consumerRecord.topic(),
                        consumerRecord.offset())
                )
                .map(ConsumerRecord::value)
                .doOnNext(metric -> log.debug("successfully consumed {}={}", Metric[].class.getSimpleName(), Metric))
                .doOnError(throwable -> log.error("Error while consuming : {}", throwable.getMessage()));

    }


    @Override
    public void run(String... args) throws Exception {
        consume().subscribe();
    }
}

【问题讨论】：

查看take 或buffer 运营商

标签： apache-kafka spring-webflux spring-kafka project-reactor reactor-kafka

【解决方案1】：

所以你可以使用 #buffer(int) 操作来达到这些目的。

针对您的具体情况：

int bufferSize = 10;
return KafkaConsumerTemplate.receiveAutoAck()
          .doOnNext(consumerRecord -> log.info("received key={}, value={} from topic={}, offset={}",
              consumerRecord.key(),
              consumerRecord.value(),
              consumerRecord.topic(),
              consumerRecord.offset())
           ).map(ConsumerRecord::value)
            .doOnNext(metric -> log.debug("successfully consumed {}={}", Metric[].class.getSimpleName(), Metric))
            .buffer(bufferSize)
            .doOnError(throwable -> log.error("Error while consuming : {}", throwable.getMessage()));

@Override
public void run(String... args) throws Exception {
   consume().subscribe(it -> {
      //it is a List of batched entities, here you can do whatever you want with your data.
   });
}

【讨论】：

谢谢，这正是我要找的