如何通过Reactor-Kafka接收器获取消息列表？

解答 #

1. collectToList()完全不可行

collectToList()会等待整个Flux流终止后才发射收集到的列表，但Reactor-Kafka的receive()返回的是无限流（只要Kafka主题存在且消费者运行，就会持续接收消息），这意味着该操作会一直将消息堆积在内存中，直到发生内存溢出（OOM），完全不符合批量处理的需求。

2. MAX_POLL_RECORDS_CONFIG的作用

MAX_POLL_RECORDS_CONFIG=500是Kafka客户端的配置，限制的是每次调用poll()方法时最多拉取的消息条数，但Reactor-Kafka会将poll到的每条消息作为独立元素发射到Flux中，默认情况下Flux会逐个推送单条消息，不会自动形成500条的列表，该配置无法直接控制Flux发射列表的大小。

3. 正确的批量处理实现

要实现类似同步场景的批量处理，应该使用Reactor的buffer()或bufferTimeout()操作符：

• buffer(500)：每收集到500条消息就发射一次列表
• bufferTimeout(500, Duration.ofSeconds(1))：满足“收集到500条”或“等待1秒”任一条件时发射列表，避免长时间无消息时一直等待

示例代码：

@Bean
public ApplicationRunner runner() {
    return args -> {
        ReceiverOptions<String, String> ro = ReceiverOptions.<String, String>create(
                Map.of(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092",
                       // 其他配置
                       ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500))
        .withKeyDeserializer(new StringDeserializer())
        .withValueDeserializer(new StringDeserializer())
        .subscription(Collections.singletonList("topic"));
        
        KafkaReceiver.create(ro)
                .receive()
                .buffer(500) // 或使用bufferTimeout(500, Duration.ofSeconds(1))
                .doOnNext(messages -> {
                    // 提取消息值并批量处理
                    List<String> messageValues = messages.stream()
                                                        .map(ReceiverRecord::value)
                                                        .toList();
                    processorService.process(messageValues);
                    
                    // 批量确认偏移量
                    messages.forEach(ReceiverRecord::acknowledge);
                })
                .subscribe();
    };
}

补充说明

• buffer()的大小建议和MAX_POLL_RECORDS_CONFIG保持一致，匹配Kafka客户端的拉取批次，减少不必要的内存累积
• 必须手动调用acknowledge()确认消息偏移量（如果使用手动提交模式），避免重复消费

内容的提问来源于stack exchange，提问作者Mikhail Geyer