Reactor Kafka基于分区排序的并发处理未达预期

问题根源 #

• 消费者并发数未配置：你的ReceiverOptions未设置concurrency属性，默认值为1，意味着仅1个消费者线程拉取所有5个分区的消息。即便后续通过groupBy分流不同分区的消息，源头消费是串行的，导致整体无法并行处理多分区数据。
• 偏移提交方式不当：sample(Duration.ofMillis(5000))会每隔5秒仅保留流中最后一个偏移并提交，这不仅会导致大量处理后的记录偏移未提交（存在重复消费风险），还会人为延缓处理速度，阻碍并发执行。

解决方案 #

1. 配置消费者并发数 #

在ReceiverOptions中设置concurrency为分区数（5），Reactive Kafka会创建对应数量的消费者线程，每个线程独立负责一个分区的消息拉取，从源头实现分区并行消费。

2. 调整偏移提交策略 #

替换sample为批量提交逻辑，比如用bufferTimeout收集一定数量或一定时间内的偏移再批量提交，既保证分区内有序，又能高效提交偏移。

3. 确保调度器线程充足 #

使用具备足够线程数的调度器（如Schedulers.parallel()或自定义线程池），保证每个分区的处理线程能真正并行运行。

修改后的示例代码 #

@Bean
Map<String, Object> kafkaConsumerConfiguration() {
    Map<String, Object> configuration = new HashMap<>();
    configuration.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    configuration.put(ConsumerConfig.GROUP_ID_CONFIG, "sampleGroupId");
    configuration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
    configuration.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    configuration.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    return configuration;
}

@Bean
ReceiverOptions kafkaReceiverOptions(@Value("${kafka.topic.in}") String inTopicName) {
    ReceiverOptions<String, String> options = ReceiverOptions.create(kafkaConsumerConfiguration());
    return options.addAssignListener(assignments -> log.info("Assigned: " + assignments))
            .subscription(Collections.singletonList(inTopicName))
            .concurrency(5); // 设置并发数与分区数一致
}

@Bean
KafkaReceiver<String, String> reactiveKafkaReceiver(ReceiverOptions<String, String> kafkaReceiverOptions) {
    return KafkaReceiver.create(kafkaReceiverOptions);
}

// 定义线程充足的调度器
private final Scheduler scheduler = Schedulers.parallel();

@EventListener(ApplicationStartedEvent.class)
public void onMessage() {
    reactiveKafkaReceiver
            .receive()
            .groupBy(m -> m.receiverOffset().topicPartition())
            .flatMap(partitionFlux ->
                    partitionFlux.publishOn(scheduler)
                            .map(r -> {
                                // 执行记录处理逻辑
                                processRecord(partitionFlux.key(), r);
                                return r.receiverOffset();
                            })
                            // 批量提交：每100条或5秒提交一次，取先触发的条件
                            .bufferTimeout(100, Duration.ofMillis(5000))
                            .concatMap(offsets -> {
                                if (!offsets.isEmpty()) {
                                    // 提交该批次最后一个偏移（保证分区内顺序）
                                    return offsets.get(offsets.size() - 1).commit();
                                }
                                return Mono.empty();
                            })
            )
            .subscribe();
}

// 业务处理方法示例
private void processRecord(TopicPartition partition, ReceiverRecord<String, String> record) {
    log.info("Processing partition: {}, record: {}", partition, record.value());
    // 此处添加你的具体业务逻辑
}

额外说明 #

• concurrency取值建议等于分区数，让每个消费者线程对应一个分区，实现最优并行消费效率。
• 批量提交的参数（100条/5秒）可根据业务需求调整，平衡提交频率与性能。
• 如果processRecord是耗时操作，可自定义线程池调度器，确保线程数足够支撑并发处理。

内容的提问来源于stack exchange，提问作者Kokil Jain