《Java 并发编程·线程与内存模型》12.2 并行流的边界与 commonPool 陷阱

承接的问题

12.1 的 RecursiveTask 要自己写拆分、自己控制阈值、自己建池。parallelStream() 把这三件事都封装掉了：

long totalStock = books.parallelStream().mapToLong(Book::stock).sum();

代价是封装也把决策藏了起来。这段代码至少有四个问题没有答案：

• 它用的是哪个池？池里几个线程？和别人的并行流共享吗？
• 数据被切成几片？每片多大？切片成本算在谁头上？
• mapToLong 里的代码在哪个线程跑？能不能访问 ThreadLocal？能不能阻塞？
• 结果按什么顺序合并？如果操作不满足结合律，结果还对吗？

这一节不谈「并行流更快」，只谈它的边界：什么时候它值得用，什么时候它会把整个 JVM 拖慢。

parallel() 做了什么

一次 parallelStream() 的执行分三步：

1. 拆分：从数据源的 Spliterator 反复调用 trySplit()，把输入切成若干段。ArrayList、数组、IntStream.range 按下标对半切，成本 O(1)；LinkedList 的 trySplit 要顺序遍历找中点；Stream.iterate、BufferedReader.lines() 这类未知长度的源基本拆不动，并行度接近 1。
2. 提交：每个分段包装成一个 ForkJoinTask，提交到当前线程所属的池。普通业务线程没有所属池，用的就是 ForkJoinPool.commonPool()。
3. 合并：各分段的局部结果按树的形状合并。sum/count/collect 有专门的合并逻辑，顺序由实现决定，不是 encounter order。

有两个实现细节决定了后面的所有现象：

• 叶子粒度由框架决定，用户无法配置。AbstractTask 的目标是「每个 worker 分到 4 个左右叶子」，即叶子数约为 parallelism × 4。你没法像 RecursiveTask 那样指定阈值；能调的只有并行度（自建池，或启动时的 -Djava.util.concurrent.ForkJoinPool.common.parallelism），叶子数随之缩放。
• ForkJoinPool 只对 ManagedBlocker 做补偿。任务里调用 Thread.sleep、JDBC、HTTP 时，池并不知道线程被阻塞了，不会创建替补线程。并行度就这么被吃掉。

先展开第一点。JDK 的叶子目标是 parallelism << 2，即把输入切成约 parallelism × 4 份：10 核机器上 commonPool 并行度是 9，所以 100 万元素会被切成 36 份左右，每份约 2.8 万元素。这也解释了为什么并行流对「每个元素很快、但元素很多」的场景不敏感——每份仍然很大，而每份内部的循环是纯顺序的。

不过叶子数只是目标，实际参与计算的线程数还取决于任务长度。本机跑一条 2000 个元素的极短并行流，只观察到 5 个线程参与：

parallelStream sum = 1999000，参与线程 5 个
  [ForkJoinPool.commonPool-worker-1, ForkJoinPool.commonPool-worker-2, ForkJoinPool.commonPool-worker-3, ForkJoinPool.commonPool-worker-4, main]

任务太短时，框架还没来得及唤醒其它 worker，整条流就结束了。所以「并行度 9」不代表每次都有 9 个线程在干活。

第二点是大规模误用的根源。ForkJoinPool 的设计假设是「任务在算，不在等」；而业务代码里最常见的恰恰是等。

本机实测一：拆分成本与数据源

拆分和合并都是要花钱的。元素本身的工作量不够大时，并行只会更慢。测量方式：对 ArrayList<Integer> 做 mapToDouble(i -> sin(i) * cos(i)).sum()，每种规模重复到总元素数约 400 万，预热后计时。

List<Integer> data = new ArrayList<>(n);
for (int i = 0; i < n; i++) {
    data.add(i);
}
long t0 = System.nanoTime();
double acc = data.parallelStream().mapToDouble(i -> Math.sin(i) * Math.cos(i)).sum();
long us = (System.nanoTime() - t0) / 1000;

本机 JDK 25，load averages 约 110–125：

n=1000    repeats=4000  sequential    14.24 us/次  parallel    35.45 us/次  (par/seq=2.49x)
n=10000   repeats=400   sequential   137.54 us/次  parallel    73.85 us/次  (par/seq=0.54x)
n=100000  repeats=40    sequential  1348.71 us/次  parallel   338.37 us/次  (par/seq=0.25x)

盈亏平衡点在 1 万附近。1000 个元素时，并行流是顺序流的 2.49 倍慢——多出来的时间全花在拆任务、唤醒 worker、合并结果上。10 万元素时并行拿到 4 倍加速，仍然不到 9 倍，因为 sin/cos 的吞吐有限，合并也要时间。

数据源的影响同样明显。同样 100 万元素、同样的 sin/cos 负载：

ArrayList  n=1000000 sequential median= 16ms  parallel median=  2ms
LinkedList n=1000000 sequential median= 15ms  parallel median=  5ms

ArrayList 的并行加速是 8 倍，LinkedList 只有 3 倍——差的 3ms 就是 LinkedList.trySplit() 反复遍历找中点的钱。把每个元素的工作换成一次乘法（轻量运算），LinkedList 的并行甚至会比顺序更慢：本机另跑一组，100 万元素重复 20 次，顺序 3290ms，并行 4423ms。拆分的固定成本一旦超过计算本身，并行就是纯亏。

所以第一判据不是「数据大不大」，而是「元素数 × 单元素工作量 是否远大于拆分与合并的固定成本」。经验上，单元素工作量在微秒级、元素数过万，才值得考虑。

本机实测二：阻塞任务污染 commonPool

commonPool 是全 JVM 共享的。一段并行流里的阻塞任务，会让所有其它并行流一起排队。测量分两步：

1. 先在池空闲时跑一个「20 个元素、每个 sleep(10)」的流，记录耗时。
2. 另起一个线程，用「36 个元素、每个 sleep(300)」的流把 commonPool 占满；100ms 后再跑第 1 步那个小流。

CountDownLatch began = new CountDownLatch(1);
Thread hog = Thread.ofPlatform().name("hog").start(() -> {
    began.countDown();
    IntStream.range(0, 36).parallel().forEach(i -> sleep(300));
});
began.await();
Thread.sleep(100);
long polluted = timeMs(() -> IntStream.range(0, 20).parallel()
        .map(i -> { sleep(10); return i; }).asLongStream().sum());

实测：

20x10ms 的流：commonPool 空闲 31ms，被 36x300ms 占满时 276ms
36x300ms 的流在 commonPool 上：1231ms（9 个 worker + 调用线程，四轮）
同一个流放进 new ForkJoinPool(9)：1235ms
同一个流放进 new ForkJoinPool(16)：916ms，线程数 16

三个结论：

• 同一个流，空闲时 31ms，被占满时 276ms，慢 8.9 倍。这不是「并行流慢」，是「你排队等了别人的 300ms」。
• commonPool 的 1231ms 和 new ForkJoinPool(9) 的 1235ms 几乎一样：并行度相同，吞吐就相同，换池本身不加速。
• new ForkJoinPool(16) 是 916ms：36 个任务分三轮（16+16+4），理论下限是 900ms。这说明阻塞任务的吞吐只由线程数决定——每个线程都在睡觉，多一个线程就多一份并发。

注意 ForkJoinPool 不会为 Thread.sleep 做补偿。它只认 ManagedBlocker（ForkJoinPool.managedBlock），因为那是一个显式契约。业务代码里的 JDBC、HTTP 客户端不会告诉你「我要阻塞了」，池只能眼睁睁看着 worker 被占住。

绕开 commonPool 的正确姿势

官方没有提供「给某条流指定池」的 API，但有一个被广泛使用的技巧：先把任务提交到自己的池，再在池的线程里调用 parallelStream()。流任务会使用当前线程所属的池，而不是 commonPool。

package com.plume.conc.core;

import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;

public final class StockSnapshotService {
    private final ForkJoinPool pool = new ForkJoinPool(4);

    public long totalStock(List<Book> snapshot) {
        try {
            return pool.submit(() ->
                    snapshot.parallelStream().mapToLong(Book::stock).sum()).get();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException("interrupted", e);
        } catch (ExecutionException e) {
            throw new IllegalStateException("snapshot failed", e.getCause());
        }
    }

    public void close() {
        pool.shutdown();
    }
}

四个必须写清的边界：

• 这是实现行为，不是 API 契约。ForkJoinTask.fork() 在提交时检查当前线程是不是 ForkJoinWorkerThread：是就用它所属的池，不是才回退到 commonPool。JDK 多年如此，但规范没有承诺。升级 JDK 时要回归。
• 只能包住整条流水线。pool.submit(() -> stream.parallel()...).get() 里的 get() 会阻塞调用线程，所以调用方不能再是池里的线程，否则可能自锁。
• 嵌套并行流仍然是同一个池。外层流的分片里再开一条并行流，两者抢同一组 worker；如果内层还需要等外层让出线程，就可能卡死。不要嵌套。
• ThreadLocal 不会自动传进去。pool.submit 的 lambda 在调用线程创建，但真正执行在 worker 上，ThreadLocal 的值丢失。需要上下文就用 13.3 的装饰器显式传递。

如果一段逻辑里有阻塞，更彻底的做法是根本不用并行流：把阻塞任务交给虚拟线程或专门的 IO 池，并行流只留给纯 CPU 的聚合。

顺序依赖与副作用

并行流最容易出错的地方不在性能，在语义。

最后一条值得展开。下面这段代码看起来像是「批量保存」，实际是在 commonPool 的 9 个 worker 上做 9 路并发数据库写：

// 反例：阻塞 IO 跑在 commonPool 上
books.parallelStream().forEach(repo::save);

它有三个问题：占满 commonPool 影响全 JVM；并发度不可控（数据库连接池可能只有 8 个连接）；异常在半路抛出后，已经写了一半，没有回滚。正确做法是用固定大小的 ExecutorService 或虚拟线程 + Semaphore 限流，并且自己控制批次与事务边界。

异常语义也要注意：并行流里任何一个分片抛出未捕获异常，整条流水线以该异常结束，但其它分片不会被立即取消，它们会继续跑到自然结束。表现是「异常已经抛出来了，日志里还有后续的写入」。

本机实测（40 个元素，第 0 个立刻抛异常，其余各 sleep(200)，三次运行）：

caught IllegalStateException: java.lang.IllegalStateException: boom
异常抛出时已完成 36 个
500ms 后已完成 39 个

异常抛出时已经完成了 34–37 个，之后又补完了 2–5 个，最终 39/40 全部执行（失败的那个除外）。如果你的「后续写入」有副作用，半途失败就没有原子性可言——批次划分、事务边界、幂等键都得自己兜。

在 libraryd-conc 里落地

借阅系统每天要算一次「低库存书目」：库存小于等于 2 的书，用于生成补货清单。这是纯内存计算，适合并行；但它不能和业务代码里的其它并行流抢 commonPool。

package com.plume.conc.core;

import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.stream.Collectors;

public final class LowStockScanner {
    private static final int LOW_STOCK = 2;
    private final ForkJoinPool pool = new ForkJoinPool(4);

    public List<String> scan(List<Book> snapshot) {
        try {
            return pool.submit(() -> snapshot.parallelStream()
                            .filter(book -> book.stock() <= LOW_STOCK)
                            .map(Book::isbn)
                            .collect(Collectors.toList()))
                    .get();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException("scan interrupted", e);
        } catch (ExecutionException e) {
            throw new IllegalStateException("scan failed", e.getCause());
        }
    }

    public void close() {
        pool.shutdown();
    }
}

设计点：

• 池是类的私有字段，并行度 4。库存扫描是短任务，4 个 worker 足够；开太大只会和业务线程抢 CPU。
• collect(Collectors.toList()) 而不是 forEach(list::add)。collect 的合并逻辑是框架保证的，每个分片有自己的容器，最后合并，没有共享写。
• 返回值是 List<String>，不是 List<Book>。只把 ISBN 带出池，避免把领域对象暴露给后续可能修改它的代码。
• 异常在边界处转换。ExecutionException 的 getCause() 才是业务异常，直接 throw e 会丢掉原始堆栈。

同一个包里的 StockAudit（12.1）和这个类解决的是两个不同问题：前者是「一个大任务递归切分」，后者是「一批小任务并行过滤」。前者能控制阈值，后者不能；前者的池可以按任务调，后者的并行度只由数据量间接影响。

什么时候不要用

• 数据量小或单元素太便宜。本机 1000 个元素时并行慢 2.49 倍。判据是「元素数 × 单元素耗时」是否远超任务调度成本。
• 任务里有阻塞。JDBC、HTTP、文件、锁等待都会占住 worker，而且不会触发补偿。
• 需要顺序或单线程语义。forEach 不保序，findFirst 要额外协调，有状态 lambda 直接是错的。
• 操作不满足结合律。浮点求和、字符串拼接、先减后加这类操作，并行与顺序的结果可能不同。
• 数据源拆分很贵。LinkedList、Stream.iterate、IO 流的分片成本可能与计算本身同量级。
• 需要可观测性。并行流没有任务级的监控、超时、取消；出问题只能看线程栈。

常见错误与边界

小结

• parallelStream() 的三步是拆分、提交、合并。叶子粒度由框架决定（约 parallelism × 4 个），用户无法配置阈值。
• 拆分与合并有固定成本：本机 1000 个元素时并行比顺序慢 2.49 倍，1 万左右才是盈亏平衡点；LinkedList 的拆分成本让加速比从 8 倍掉到 3 倍。
• commonPool 是全 JVM 共享的，且只对 ManagedBlocker 做补偿。实测同一个 20×10ms 的流，空闲 31ms，被阻塞任务占满后 276ms。
• 想隔离就用「提交到自建池再在池内并行」的技巧，但那是实现行为而非契约；有阻塞就该换虚拟线程或 IO 池。
• 顺序、副作用、结合律是并行流的语义边界，比性能更容易出事。

12.1 和 12.2 都用了分治，但都绕开了一个数字：切到多小为止。下一节把阈值摊开，用归并排序量出「太小」和「太大」各自的代价，再补上异常传播与取消的语义。

阅读导航：上一节 12.1 工作窃取与 ForkJoinPool ｜ 下一节 12.3 分治拆分、阈值与异常