本文详解为何 publishOn() 无法实现真正的并行映射，以及如何通过 parallel() + runOn() 正确启用多线程并发执行 map 操作，确保 CPU 密集型任务高效利用多核资源。
本文详解为何 `publishon()` 无法实现真正的并行映射，以及如何通过 `parallel()` + `runon()` 正确启用多线程并发执行 `map` 操作，确保 cpu 密集型任务高效利用多核资源。
在使用 Project Reactor 处理批量数据时，一个常见误区是认为调用 .publishOn(Schedulers.parallel()) 就能自动让后续的 .map() 并行执行。但事实并非如此：publishOn 仅切换下游操作符的执行线程上下文，它不会改变运算的串行本质——即每个 map 仍按顺序在一个线程（如 parallel-1）上依次执行，只是这个“顺序流”被迁移到了并行调度器的线程池中。
要真正实现多元素并发处理（例如对每个 buffer(2) 生成的子列表独立、同时执行耗时逻辑），必须启用 Reactor 的并行 Flux 模式。该模式将原始序列划分为多个并行子流（默认线程数 = CPU 核心数），每个子流在独立线程上执行其专属的 map 逻辑。
正确写法如下：
Orderly Sdk React Hooks
下载
Orderly React SDK 钩子使用参考指南，包括 useOrderEntry、usePositionStream、useOrderbookStream、useCollateral 等。
@Test
public void testParallelBufferedProcessing() throws InterruptedException {
Flux.just("a", "b", "c", "d", "e", "f", "g")
.buffer(2) // → ["a","b"], ["c","d"], ["e","f"], ["g"]
.parallel() // 启用并行模式：拆分为 N 个并行子流
.runOn(Schedulers.parallel()) // 为每个子流指定执行线程池（关键！）
.map(this::doTimeintensiveStuff) // ✅ 每个 buffer 在不同 parallel-N 线程上并发执行
.sequential() // 合并结果回有序序列（保持原始 buffer 顺序）
.doOnNext(val -> {
log.info("[" + Thread.currentThread().getName() + "] " + val);
})
.blockLast();
}
private String doTimeintensiveStuff(List<String> input) {
try {
TimeUnit.MILLISECONDS.sleep(100); // 模拟真实耗时操作
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return String.join(", ", input);
}✅ 预期日志输出（体现真正并行）：
[parallel-2] a, b [parallel-3] c, d [parallel-1] e, f [parallel-4] g
⚠️ 关键注意事项：
-
parallel()必须在buffer()之后、map()之前调用，否则并行化的是原始元素而非缓冲块； -
runOn()是触发实际线程切换的必需步骤；仅parallel()不指定调度器，默认使用Schedulers.immediate()（即主线程），毫无并发效果； -
sequential()用于保障最终结果顺序与输入 buffer 顺序一致（若顺序不重要，可省略）； - 对于 I/O 密集型任务，建议改用
Schedulers.boundedElastic()；CPU 密集型才推荐Schedulers.parallel()； - 并行度可通过
.parallel(n)显式指定（如.parallel(4)），避免过度线程竞争。
总结：publishOn 控制「在哪执行」，parallel().runOn() 才决定「是否并发执行」。理解这一根本区别，是写出高性能响应式数据处理逻辑的关键前提。
