Spring Reactor：无突变实现输入数据批处理

嘿，你的需求正好踩中了Reactor响应式编程的强项——完全不需要手动维护外部集合或者加锁，用原生操作符就能优雅解决！

首先得指出你初步尝试的问题：用外部batches集合攒数据属于管道外的状态突变，在Reactor的异步多线程环境下会有并发安全问题，而且违背了响应式编程的无状态设计原则，这正是我们要避免的。

最优实现方案：用buffer()操作符做批次聚合 #

Reactor提供了buffer()系列操作符，专门用来将上游流的元素按指定规则（数量、时间等）聚合成批次，完全在管道内处理，天然线程安全，不需要任何锁。结合Spring Kafka的响应式API，代码可以写成这样：

// 假设你已经通过配置注入了ReactiveKafkaConsumerTemplate实例
@Autowired
private ReactiveKafkaConsumerTemplate<String, YourRecordType> reactiveKafkaConsumerTemplate;

@Autowired
private BusinessService businessService;

public void startBatchProcessing() {
    reactiveKafkaConsumerTemplate.receive()
        // 先提取Kafka ConsumerRecord中的实际业务数据
        .map(ConsumerRecord::value)
        // 核心：每积累100条记录就生成一个批次List
        .buffer(100)
        // 处理每个批次，调用业务服务（推荐业务服务返回响应式类型Mono/Flux）
        .flatMap(batch -> {
            // 调用你的业务服务，这里假设processBatch返回Mono<Void>表示处理完成
            return businessService.processBatch(batch)
                .doOnSuccess(v -> System.out.printf("成功处理批次，共%d条记录%n", batch.size()))
                .doOnError(e -> System.err.printf("批次处理失败：%s%n", e.getMessage()));
        })
        // 订阅流，启动整个处理流程
        .subscribe();
}

为什么这个方案更好？ #

• 无外部状态/锁：所有状态聚合都在Reactor管道内完成，buffer()操作符由Reactor维护内部状态，天然支持并发场景，不需要你手动处理线程安全
• 响应式友好：如果你的业务服务也是响应式实现（返回Mono/Flux），整个流会保持非阻塞特性，充分利用系统资源
• 灵活扩展：如果需要兼顾“数量+超时”的批次策略（比如没攒够100条但5秒过去了也要处理），只需要把buffer(100)改成bufferTimeout(100, Duration.ofSeconds(5))即可
• 异常处理可控：可以通过onErrorContinue、retry等操作符灵活处理批次处理失败的情况，比如重试某个失败批次，或者跳过错误批次继续处理后续数据

额外注意点 #

1. 确保你的businessService.processBatch()方法是线程安全的，或者如果它是阻塞的，记得用subscribeOn(Schedulers.boundedElastic())把阻塞操作放到专门的线程池里，避免阻塞Reactor的IO线程
2. 如果需要精确的一次性语义（Exactly-Once），可以结合Kafka的事务机制和Reactor的transactional()操作符来实现

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