Flux并行环境下批量处理元素及buffer算子报错解决求助

问题原因 #

你的代码中IDE报错是因为调用parallel()后得到的是ParallelFlux类型，而buffer(int)是Flux类的专属算子，ParallelFlux并没有直接提供该方法，因此无法直接调用。

解决方案 #

要实现「每个并行处理轨道按指定大小批处理元素」的需求，需要对ParallelFlux中的每个子Flux单独应用buffer算子。可以通过transform方法完成这一操作，该方法允许对每个并行分支的Flux应用自定义的流处理逻辑。

修正后的代码 #

import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

public class BufferAndRunOnExample {
    public static void main(String[] args) {
        Flux.range(1, 10)
                .parallel()
                .runOn(Schedulers.parallel())
                // 对每个并行分支的Flux应用buffer，按3个一组批量处理
                .transform(flux -> flux.buffer(3))
                .doOnNext(batch -> 
                    System.out.printf("处理批次: %s，线程: %s%n", 
                        batch, Thread.currentThread().getName())
                )
                .sequential()
                .blockLast();
    }
}

代码说明 #

1. parallel()：将原Flux拆分为多个并行处理轨道（默认数量等于CPU核心数）。
2. runOn(Schedulers.parallel())：指定并行轨道使用的调度器，确保处理逻辑在多线程环境中执行。
3. transform(flux -> flux.buffer(3))：对每个并行分支的子Flux单独应用buffer(3)，让每个轨道自行积累元素，凑满3个后形成一个批次。
4. doOnNext：打印批次内容及处理线程，验证并行批量处理的效果。
5. sequential()：将多个并行处理的结果流合并回一个顺序流，方便后续统一处理。
6. blockLast()：阻塞主线程，等待整个流处理完成。

补充说明 #

如果你的需求是「先全局按3个一组批量，再将每个批次分发到并行轨道处理」，则需要调整算子顺序，先执行buffer再执行parallel，示例代码如下：

Flux.range(1, 10)
        .buffer(3)
        .parallel()
        .runOn(Schedulers.parallel())
        .doOnNext(batch -> 
            System.out.printf("处理批次: %s，线程: %s%n", 
                batch, Thread.currentThread().getName())
        )
        .sequential()
        .blockLast();

这种方式下，先将全局元素按3个一组打包，再把每个批次分配到不同线程并行处理，适合需要先批量再并行的场景。

内容的提问来源于stack exchange，提问作者sesha sai srivatsav