承接的问题
12.1 和 12.2 讲的是任务已经进了队列之后怎么调度。但并行流的第一性问题不在调度，而在任务是怎么被拆出来的：拆成几份、每份多大、拆出来的子任务能不能均匀落到 worker 上。
parallelStream 看起来是一行代码，背后是 Spliterator.trySplit() 反复对半切，切到"目标叶大小"就停。这个目标由 AbstractTask.LEAF_TARGET = parallelism << 2 决定：并行度 9 的 commonPool，目标叶子数是 36，也就是大约每个核 4 个叶子。源码注释给的理由是"为了负载均衡而过度拆分"。
本节回答四个问题：Spliterator 的三个方法各自承诺了什么；AbstractTask 为什么用循环而不是递归拆分；拆分粒度失控会发生什么；以及为什么 parallelStream 里做 I/O 会把整个 JVM 的并行流一起拖住。
Spliterator 的三个契约
并行流只依赖 Spliterator 的三个方法：
public interface Spliterator<T> {
Spliterator<T> trySplit(); // 切一半出去；切不动返回 null
long estimateSize(); // 剩余元素数的估计
int characteristics(); // 有序性、大小是否精确、可否修改
}
trySplit() 的契约最容易写错：它返回的新 Spliterator 覆盖前半段，this 变成后半段，两段不重叠、合起来等于原来。返回值是 null 表示"切不动了"，调用方应当把它当叶子处理。注意它没有"切多少"的参数——粒度完全由实现决定。
estimateSize() 不是"精确大小"，是"估计"。只有声明了 SIZED 特征，它才必须精确；声明了 SUBSIZED，则每个 split 出来的子段也必须精确。AbstractTask 用 estimateSize() 算目标叶大小，如果估计偏差很大，拆出来的叶子会一大一小。
characteristics() 是一组位掩码，常见的几个：
| 特征 | 含义 | 对并行流的影响 |
|---|---|---|
SIZED | estimateSize() 精确 | 决定能否用目标大小拆分 |
SUBSIZED | 每个子段也精确 | 影响 AbstractTask 的目标计算 |
ORDERED | 有相遇顺序 | 决定 forEachOrdered/findFirst 的语义 |
SORTED | 按比较器有序 | 影响 sorted() 是否要重排 |
NONNULL | 不含 null | 影响某些短路操作的优化 |
IMMUTABLE / CONCURRENT | 源不可变 / 可并发修改 | 决定是否允许并发遍历 |
一个最小可用的自定义 Spliterator.OfInt，每次对半切：
import java.util.Spliterator;
import java.util.function.IntConsumer;
public final class HalfSplit implements Spliterator.OfInt {
private final int[] a;
private int lo, hi;
public HalfSplit(int[] a, int lo, int hi) {
this.a = a; this.lo = lo; this.hi = hi;
}
@Override
public Spliterator.OfInt trySplit() {
int mid = (lo + hi) >>> 1;
if (mid <= lo) return null; // 切不动了
int oldLo = lo;
lo = mid; // this 变成后半段
return new HalfSplit(a, oldLo, mid);
}
@Override public long estimateSize() { return hi - lo; }
@Override public int characteristics() {
return SIZED | SUBSIZED | IMMUTABLE | NONNULL | ORDERED;
}
@Override public boolean tryAdvance(IntConsumer action) {
if (lo >= hi) return false;
action.accept(a[lo++]);
return true;
}
@Override public void forEachRemaining(IntConsumer action) {
for (int i = lo; i < hi; i++) action.accept(a[i]);
lo = hi;
}
}
这段代码有一个隐含约束：trySplit() 修改了 this.lo，所以它只能在单线程下被调用，或者由调用方保证同一个实例不被并发访问。并行流框架满足这个约束——每个 AbstractTask 独占一个 Spliterator。但如果你把这个对象传给别的线程池自己用，就会出错。
forEachRemaining 一定要实现，而且要和 tryAdvance 语义一致。并行流的叶子节点会调 forEachRemaining，如果它漏掉元素或者重复遍历，求和结果就错了，而且不会有任何异常。
AbstractTask：用循环拆，不用递归
parallelStream 的拆分逻辑在 java.util.stream.AbstractTask 里，它继承 CountedCompleter：
// JDK 25 源码摘录，已删减
@Override
public void compute() {
Spliterator<P_IN> rs = spliterator, ls; // right, left spliterators
long sizeEstimate = rs.estimateSize();
long sizeThreshold = getTargetSize(sizeEstimate);
boolean forkRight = false;
K task = (K) this;
while (sizeEstimate > sizeThreshold && (ls = rs.trySplit()) != null) {
K leftChild, rightChild, taskToFork;
task.leftChild = leftChild = task.makeChild(ls);
task.rightChild = rightChild = task.makeChild(rs);
task.setPendingCount(1);
if (forkRight) {
forkRight = false;
rs = ls;
task = leftChild;
taskToFork = rightChild;
} else {
forkRight = true;
task = rightChild;
taskToFork = leftChild;
}
taskToFork.fork();
sizeEstimate = rs.estimateSize();
}
task.setLocalResult(task.doLeaf());
task.tryComplete();
}
三个设计点：
用 while 而不是递归。 拆一次就 fork 一个子任务、自己继续处理另一半，循环继续。递归深度是 O(1)，不会因为拆分次数多而爆栈。这一点在 12.3 后面会看到一个反例：手写的 RecursiveTask 如果每层都递归调用 compute()，拆得细就会 StackOverflowError。
forkRight 交替。 如果 Spliterator 总是左边大右边小（或者反过来），固定的"fork 左、算右"会让某个线程一直拿大的那半。交替 fork 方向把偏差摊平。
setPendingCount(1) + tryComplete()。 这是 CountedCompleter 的完成计数：父任务挂起时记 1，子任务完成时 tryComplete() 递减，减到 0 才触发父任务的 onCompletion。和 RecursiveTask.join() 的区别是不需要显式 join，也不需要在父任务里阻塞等待。
目标叶大小由两个方法决定：
// JDK 25 源码摘录，已删减
private static final int LEAF_TARGET = ForkJoinPool.getCommonPoolParallelism() << 2;
public static int getLeafTarget() {
Thread t = Thread.currentThread();
if (t instanceof ForkJoinWorkerThread) {
return ((ForkJoinWorkerThread) t).getPool().getParallelism() << 2;
}
else {
return LEAF_TARGET;
}
}
public static long suggestTargetSize(long sizeEstimate) {
long est = sizeEstimate / getLeafTarget();
return est > 0L ? est : 1L;
}
<< 2 就是乘 4：每个核 4 个叶子，注释说是为了"在叶子不均匀或某些核被占用时，让别的线程有活可帮"。注意 getLeafTarget() 会先看当前线程是不是 ForkJoinWorkerThread——如果是，用那个池的并行度；否则退回 commonPool 的并行度。所以同一个 Spliterator 在专用池里跑和在 commonPool 里跑，目标叶大小可能不同。
本机拆了多少次
用 IntStream.range(0, n).spliterator() 包一层计数 Spliterator，再交给并行流，就能数出 JDK 自己拆了几次：
commonPool.parallelism=9 leafTarget=36
n=2^12 splits=63 leaves=64 sum=8386560
n=2^16 splits=63 leaves=64 sum=2147450880
n=2^20 splits=63 leaves=64 sum=549755289600
n=2^24 splits=39 leaves=40 sum=140737479966720
（本机 Temurin 25.0.5+8，10 核，load average 82.6；单次运行、随调度波动。）
n 从 4096 涨到 1048576，拆分次数稳定在 63、叶子稳定在 64。算一下：目标叶大小是 n / 36，对半切只能切出 2 的幂，1048576 切 6 层是 64 片、每片 16384，比 29127 小；切 5 层是 32 片、每片 32768，比目标大。所以停在 64。2^12 时目标 113，切 6 层每片 64，也停在 64。
n=2^24 时反而只拆了 39 次。这一条我没有稳定的解释，只能说：拆分次数不是"n 越大越多"，它由目标叶大小和 trySplit 的切法共同决定，具体到这个数据点还受并行流内部短路和负载影响。把它当"63 左右"的量级读，不要当公式。
拆分粒度失控会怎样
把 trySplit 改成"每次只切 8 个元素出去"，同样 100 万个元素，两种写法结果完全不同：
A HalfSplit+RecursiveTask sum=549755289600 splits=15 leaves=16 ms=24
B TinySplit+RecursiveTask StackOverflowError splits=10434 leaves=7625
C HalfSplit+parallelStream sum=549755289600 splits=63 ms=16
D IntStream.range.parallel sum=549755289600 ms=19
E TinySplit+parallelStream sum=549755289600 splits=127432 ms=40
（本机 Temurin 25.0.5+8；单次运行，ms 随调度波动，只看结构不看绝对值。）
A 和 C 是同一个 HalfSplit，两种驱动方式：手写 RecursiveTask 拆了 15 次（16 个叶子），并行流拆了 63 次（64 个叶子）——因为并行流的目标叶大小是框架算的，手写版的目标是 n / 16。同一个 Spliterator，驱动方不同，拆分次数差 4 倍。
B 是反面教材：手写 RecursiveTask 每层递归调用 compute()，TinySplit 每次只切 8 个，递归深度约 13 万层，在 10434 次拆分后 StackOverflowError。注意它不是在 trySplit 里崩的，是 compute 的调用栈崩的。
E 是同一个 TinySplit 交给并行流：拆了 127432 次，没有崩，结果正确。原因是 AbstractTask.compute() 用 while 循环，拆分次数再多也只占一层栈。同一个糟糕的 Spliterator，在并行流里只是慢，在手写递归里会崩——这就是 AbstractTask 坚持循环拆分的原因。
E 的 127432 次拆分也说明了粒度失控的代价：每个叶子只有 8 个元素，框架要为每个叶子分配一个 CountedCompleter、做一次 fork/join 簿记。本机 40ms 对 16ms，差的两倍多基本都花在簿记上。
commonPool 的默认并行度是硬上限
ForkJoinPool.commonPool() 的并行度默认是"核数 - 1"，本机 10 核是 9。这个值可以用 -Djava.util.concurrent.ForkJoinPool.common.parallelism=N 改，但改的是全 JVM 共享的那一个池。
用 18 个阻塞任务（Thread.sleep(400) 模拟 I/O）测一下：
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ForkJoinPool;
import java.util.stream.IntStream;
public final class CommonPoolLimit {
static final Set<String> THREADS = ConcurrentHashMap.newKeySet();
static final int TASKS = 18;
static final int SLEEP_MS = 400;
static void blockingTask() {
THREADS.add(Thread.currentThread().getName());
try {
Thread.sleep(SLEEP_MS);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
public static void main(String[] args) {
int par = ForkJoinPool.getCommonPoolParallelism();
System.out.println("availableProcessors=" + Runtime.getRuntime().availableProcessors()
+ " commonPool.parallelism=" + par);
THREADS.clear();
long t0 = System.nanoTime();
IntStream.range(0, TASKS).parallel().forEach(i -> blockingTask());
System.out.printf("commonPool 耗时=%dms 线程数=%d%n",
(System.nanoTime() - t0) / 1_000_000, THREADS.size());
THREADS.clear();
ForkJoinPool big = new ForkJoinPool(TASKS);
t0 = System.nanoTime();
big.submit(() -> IntStream.range(0, TASKS).parallel().forEach(i -> blockingTask())).join();
System.out.printf("专用池(%d) 耗时=%dms 线程数=%d%n", TASKS,
(System.nanoTime() - t0) / 1_000_000, THREADS.size());
big.shutdown();
}
}
本机输出：
availableProcessors=10 commonPool.parallelism=9
commonPool 18 个阻塞任务 耗时=816ms 线程数=10
专用池(18) 18 个阻塞任务 耗时=411ms 线程数=18
（本机 Temurin 25.0.5+8；单次运行，耗时随调度波动，不作基准。）
18 个 400ms 的任务在 commonPool 里跑了两轮（816ms），因为只有 9 个 worker；换成 18 线程的专用池，一轮跑完（411ms）。线程数 10 而不是 9，是因为调用 parallelStream 的主线程自己也参与执行——这也是 12.1 提到的"外部提交会被帮助执行"。
风险不在这 816ms，而在共享：commonPool 是全 JVM 一个。你的报表任务占满 9 个 worker 的 400ms 里，同一个 JVM 里任何别的 parallelStream、任何 CompletableFuture 的默认异步执行，都在这条队列后面排着。如果一个请求路径里的 parallelStream 做的是数据库查询，那么它排队的对象是整台机器上所有用默认池的代码。
在 libraryd-conc 里落地
库存盘点天然是"数组 + 纯函数"，正适合并行流。但要用对三件事：源必须是 SIZED、叶子必须是纯内存计算、池子必须是专用的。
package com.plume.conc.core;
import java.util.Spliterator;
import java.util.concurrent.ForkJoinPool;
import java.util.function.IntConsumer;
/** 盘点批次的区间视图：不复制底层数组，只移动 lo/hi。 */
public final class BatchSpliterator implements Spliterator.OfInt {
private final int[] slotIds;
private int lo, hi;
public BatchSpliterator(int[] slotIds, int lo, int hi) {
this.slotIds = slotIds; this.lo = lo; this.hi = hi;
}
@Override
public Spliterator.OfInt trySplit() {
int mid = (lo + hi) >>> 1;
if (mid - lo < 1024) return null; // 叶子下限：小于 1024 不再拆
int oldLo = lo;
lo = mid;
return new BatchSpliterator(slotIds, oldLo, mid);
}
@Override public long estimateSize() { return hi - lo; }
@Override public int characteristics() {
return SIZED | SUBSIZED | IMMUTABLE | NONNULL | ORDERED;
}
@Override public boolean tryAdvance(IntConsumer action) {
if (lo >= hi) return false;
action.accept(slotIds[lo++]);
return true;
}
@Override public void forEachRemaining(IntConsumer action) {
for (int i = lo; i < hi; i++) action.accept(slotIds[i]);
lo = hi;
}
/** 盘点：每个槽位独立计算，结果按分类累加，全程不碰 I/O。 */
public static long countReserved(int[] slotIds, ForkJoinPool pool) {
return pool.submit(() ->
java.util.stream.StreamSupport.intStream(
new BatchSpliterator(slotIds, 0, slotIds.length), true)
.mapToLong(id -> id % 7 == 0 ? 1 : 0)
.sum()).join();
}
}
三个决策：
第一，trySplit 加了下限 mid - lo < 1024。不加下限时，框架会一路切到单元素，拆分次数和叶子对象数按元素数线性增长——上面的 E 组实验就是这个下场。加了下限，叶子数被压在"元素数 / 1024"的量级。
第二，characteristics 声明了 SIZED | SUBSIZED，因为 estimateSize() 是精确的 hi - lo；如果不声明 SIZED，AbstractTask 的目标叶计算会退回默认值，拆分行为变得不可预测。
第三，countReserved 用专用池 pool.submit(...)，而不是直接 parallel()。这样盘点不会占用 commonPool，也不会被别人的 parallelStream 拖住。调用方负责把 slotIds 读进内存——读库是 I/O，必须在提交前完成。
什么时候不要用并行流
源是链表或迭代器。 LinkedList、BufferedReader.lines()、Files.lines() 的 trySplit 要么切不动，要么切出来的段大小不可知；estimateSize 不准，目标叶大小就失真。数据量小时，拆分开销比计算还大。
叶子要做 I/O。 上面 816ms 的实验已经说明：commonPool 的 9 个 worker 被阻塞任务占满时，全 JVM 的并行流一起排队。数据库查询、HTTP 调用、文件读写都应该放到虚拟线程或专门的 I/O 池，算完再回到并行流做纯计算。
需要顺序或状态。 forEach 不保证顺序，findFirst 在并行下必须尊重相遇顺序，短路收益有限；有共享可变状态时，parallelStream 里的 += 一定会丢更新。要么用 collect 的合并器，要么老老实实串行。
任务本身只有几微秒。 目标叶大小是 n / (4 × 并行度)，如果每个叶子的计算只有几微秒，CountedCompleter 的分配和 fork/join 的簿记会占大头。本机 A 组（16 个叶子，24ms）和 C 组（64 个叶子，16ms）的差距就来自拆分策略，不是计算本身。
常见错误与边界
trySplit() 返回 this。 契约要求返回一个覆盖前半段的新对象，this 变成后半段。返回 this 会让框架重复处理同一段数据，结果是重复计数，且不会抛异常。
estimateSize() 撒谎。 声明了 SIZED 却返回估计值，AbstractTask 会按错误的目标拆分，叶子大小严重不均，某个 worker 拖到最后。
在 forEachRemaining 里漏掉元素。 tryAdvance 和 forEachRemaining 必须覆盖同一段数据。框架在叶子上只调一次 forEachRemaining，漏了就少算，不会报错。
用单次冷启动的耗时判断并行流快不快。 同一个 100 万元素求和，D 组单次运行 19ms；预热 5 轮后取最好成绩，本机是串行 0.691ms、并行 0.175ms。19ms 里绝大部分是 JIT 编译和框架初始化，不是拆分成本。数据量小于一万时拆分才真正占主导，这时并行确实可能比 for 慢。
在 commonPool 里嵌套 parallelStream。 外层占满 worker 后，内层流没有线程可用，会退化成串行甚至死等。这是 ForkJoinPool 的经典陷阱，和 12.2 的"外部帮助"不是一回事。
小结
并行流的性能由拆分决定，拆分的三个契约是 trySplit 切一半、estimateSize 给估计、characteristics 声明语义。AbstractTask 用 while 循环而不是递归来拆，fork 方向交替以摊平偏斜，目标叶大小是 并行度 × 4，本机 commonPool 并行度 9、目标 36，实际拆出 64 个叶子。粒度失控时，同一个 Spliterator 在并行流里拆 12 万次还能跑完，在手写递归里一万次就 StackOverflowError——这是 CountedCompleter 循环拆分的价值。最后，commonPool 是全 JVM 共享的 9 个线程，18 个阻塞任务要跑两轮（816ms），而 18 线程的专用池只要一轮（411ms）；把 I/O 放进并行流，代价由整台机器上所有用默认池的代码一起付。
三节合起来是一条完整的链：WorkQueue 提供两端队列（12.1），poll/phase 处理两端相撞（12.2），Spliterator 与 AbstractTask 决定往队列里放多少任务（12.3）。下一章换到另一个方向：任务之间不再靠拆分并行，而是靠依赖关系串起来——CompletableFuture 的 Completion 栈。
阅读导航：上一节 12.2 双向队列与 steal 竞争 ｜ 下一节 13.1 CompletableFuture 的依赖栈
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。