前段时间一个朋友拉我过去说他们的批量计算服务每隔几小时就会出现一次“线程数莫名翻倍然后整柜卡住”的怪象。日志里全是java.util.concurrent.ForkJoinPool的痕迹线程转储拉出来一看工作线程几乎全卡在join()调用上。聊到最后发现问题根子不在于join用得太多而在于他们对fork/join家族的线程控制理解得过于理想化——只看到了“分治并行”的甜头没搞清背后线程是怎么被创建、指派、入队、偷取和等待的。这篇文章就把fork/join这套东西的线程控制细节摊开讲一遍包括ForkJoinPool的并行度设计、fork/join的线程流转真相、工作窃取队列、以及我踩过的几种坑和实测数据。适合正在用或者准备用并行流、递归任务、CompletableFuture的开发者尤其是对“为什么线程池会莫名暴涨”“为什么任务偶发性饥饿”这类问题没头绪的朋友。1. 为什么ExecutorService在递归分治场景下会“裁跟头”Fork/Join模型改了什么先回到一个最朴素的问题既然ThreadPoolExecutor已经很成熟了为什么JDK还要在Java 7里引进ForkJoinPool1.1 一个典型的分治任务数组递归聚合假设你要对一个长度上千万的整数数组求和。最简单粗暴的并行方式是把它拆成两半每一半再往下拆直到任务小到值得顺序执行。伪代码大概是if (length THRESHOLD) { return directSum(arr, low, high); } else { int mid (low high) 1; SumTask left new SumTask(arr, low, mid); SumTask right new SumTask(arr, mid, high); // 到这里怎么并行 }如果你用ThreadPoolExecutor最直觉的写法是这样的FutureLong leftFuture pool.submit(left); FutureLong rightFuture pool.submit(right); return leftFuture.get() rightFuture.get();表面上看没错但问题很快就来了。1.2 固定线程池面对递归任务的三层崩溃第一层崩溃在任务数量的不可控。一个上千万的数组拆到阈值1万任务数量直接到千级甚至万级。ThreadPoolExecutor的任务是放阻塞队列的队列一长内存和出队开销都很难看。第二层崩溃在父任务占着线程等子任务。这是最致命的一点。父任务调future.get()时整个工作线程是挂起等待的。假设池里有8个线程每个线程执行到一个需要等待子任务的父任务这8个线程就全“挂”在等结果上子任务反而排不进队列执行形成典型的池饥饿。现实中你可能会看到线程池里的线程数长期不涨、任务堆积或者干脆超时。第三层崩溃是任务之间的“父子局部性”完全丢失。子任务是从队列里被任意线程捡走的父任务刚拆出来的数据可能还在CPU缓存里结果被别的线程拿去算缓存命中率直线下降。1.3 ForkJoinPool的三个关键改变ForkJoinPool针对这三层问题做了三个核心设计任务内联执行工作线程拆出子任务后不会傻等。它会自己动手执行其中一个子任务另一个放进队列等待时还能去“偷”别的任务来干。线程不会被空等浪费。工作窃取队列每个工作线程维护自己的双端队列本地任务按“后进先出”处理空闲线程按“先进先出”从别人队里偷任务既减少争抢又保持了父任务的数据局部性。协作式调度取代抢占式调度普通线程池是“任务排队、线程抢”ForkJoinPool是“任务分治、线程协作”任务不再是孤立单元而是有父子关系、可合并结果的计算树。所以ForkJoinPool不是简单地把固定线程池换了个名字而是把“线程控制”的粒度从外部提交下降到了内部任务协作。理解这一点后文的所有坑都可以顺着这条主线推出来。2. 并行度的数字游戏CommonPool默认值、自定义池、线程工厂与补偿机制“控制线程”第一件事就是控制线程数量。ForkJoinPool最核心的参数叫parallelism中文文档里叫并行度。很多人以为它等同普通线程池的corePoolSize这个理解不准确。2.1 parallelism到底控制什么在ForkJoinPool内部工作线程数组的大小围绕parallelism构建。它决定的是**「常用工作线程」的上限**并不是一个“核心数加临时数”的动态范围。更准确地说它定义了池中活动的工作线程数量目标。// 创建一个并行度为4的池 ForkJoinPool pool new ForkJoinPool(4); ForkJoinTaskLong task new SumTask(arr, 0, arr.length); long result pool.invoke(task);这里pool.invoke(task)会阻塞当前调用线程直到整个分治树算完。注意当前调用线程并不是干等它会以“提交者/参与者”的身份混进执行也参与窃取任务。这也是为什么默认并行度要“核数减一”——因为调用线程自己也要出一份力。2.2 commonPool默认值的数学来源日常用parallelStream()或者CompletableFuture默认的异步任务时根本没有创建池默认用的就是ForkJoinPool.commonPool()。这个公共池的默认并行度有严格的计算规则// ForkJoinPool源码中的默认逻辑简化 int parallelism Math.max(1, Runtime.getRuntime().availableProcessors() - 1);所以一台8核机器上默认并行度是7不是8。JVM留出1个处理器的余量给调用线程、主线程或外部系统。但很多人并不知情于是会出现“8核机器只有7个线程在处理我的并行流”的情况。如果你确实需要改有两个途径# 方式一启动参数 -Djava.util.concurrent.ForkJoinPool.common.parallelism16// 方式二代码里读取常用并行度 int currentParallelism ForkJoinPool.getCommonPoolParallelism(); // 实际使用时应避免在运行期动态修改公共池属性2.3 自定义池时怎么定并行度如果是自建ForkJoinPool并行度选择要分场景场景推荐并行度原因纯CPU密集的分治计算等于或略小于CPU核数每个工作线程都在算超配只会增加上下文切换CPU密集但调用线程也参与核数减1调用线程自己也能干活池线程减一个形成互补任务内存在等待/轻量阻塞核数乘以1.5~2补偿部分线程因等待浪费的时间任务内有重度阻塞IO不建议硬扛用ManagedBlocker见4.1单纯加大线程数会让任务队列膨胀阻塞还是阻塞这个表格不是拍脑袋是我在实际压测里反复试出来的结论。特别是最后一条只要有重的阻塞IO就不要靠无脑调高并行度去解决问题ForkJoinPool的协作式调度最怕长时间挂起。2.4 线程工厂与异常处理的控制入口ForkJoinPool的构造函数也接受ForkJoinWorkerThreadFactory和UncaughtExceptionHandler。默认线程工厂直接创建“裸线程”名字看着费力。在生产环境我强烈建议自定义线程工厂至少让dump里的线程名能一眼认出是谁家的池ForkJoinPool pool new ForkJoinPool( 4, poolTask - { ForkJoinWorkerThread worker new ForkJoinWorkerThread(poolTask) {}; worker.setName(biz-calc-worker- worker.getPoolIndex()); return worker; }, (thread, throwable) - { System.err.println(worker crashed: thread.getName()); throwable.printStackTrace(); }, false );另外ForkJoinPool没有像ThreadPoolExecutor那样暴露RejectedExecutionHandler它继承的是AbstractExecutorService池关闭后再提交任务会抛RejectedExecutionException。这一点很多人不知道以为池关了还会默默接收任务结果在停机时遇到奇怪的提交异常。3. fork与join的线程流转真相谁在派活、谁在等结果、谁在偷任务很多熟悉ExecutorService的人刚接触fork/join时会把fork()脑补成“开两个线程执行”把join()脑补成future.get()。这两条理解都是错的而且会导致你日后debug无从下手。3.1 fork()不是创建线程而是“把任务放到队列里”task.fork()的语义是把当前任务提交给当前工作线程的私有任务队列。如果没有线程归属比如外部线程直接调用fork则会抛异常或者进入全局提交队列。我用一个生活化的类比来解释你正在柜台后面做菜做菜过程中需要洗菜和切菜。你不会另外去雇两个人来帮你做这两件事你只是把洗菜这件事先放到手边的待办筐里然后自己继续切菜。等你手头空下来再从筐里拿来做。// 常见错误理解 left.fork(); // 以为这里开了两个线程 right.fork(); return left.join() right.join(); // ForkJoin官方推荐的范式 left.fork(); // left入队 long rightResult right.compute(); // 当前线程直接执行right long leftResult left.join(); // 取回left的结果 return leftResult rightResult;为什么推荐“fork一个、compute一个”而不是“两个都fork”因为两个都fork时当前线程没有活干只能join等待等于放弃了参与计算而只fork一个、自己算另一个当前线程全程在算计算密度更高。invokeAll()内部其实也是类似的思路先fork第一个再invoke第二个然后把前者join回来。3.2 join()的等待机制不是简单的线程sleepjoin()的底层等待并不是普通阻塞。工作线程在执行join时如果发现自己等待的任务还没完成它会主动去自己的队列尾部继续取任务执行或者去其他线程的队列偷任务。只有当实在是没任务可偷、任务树已经收敛到差一步时才会真正挂起。这个机制说白了一句话工作线程不愿闲着它把“等待子任务”的时间用来干别的活。这和“线程阻塞然后被唤醒”的模型完全不同。所以当你看到线程dump里join()栈的时候别急着下结论说是死锁先看看它是不是正在别的任务里“以waiting为由偷着干”。3.3 工作窃取为什么能从队尾偷出高效率每个工作线程有一个双端队列。分治产生的子任务常规情况下是压入当前线程的队尾当前线程取任务时从队尾取后进先出。而别的线程来偷任务时从队首取先进先出。自己从队尾取刚拆出来的子任务数据大概率还在CPU缓存里局部性好且和父任务的合并过程天然衔接。别人从队首取队首任务是更早拆出来的“老任务”离“新鲜数据”最远被偷走对原线程的局部性伤害最小。正是这种“自己拿最新的别人偷最老的”设计让ForkJoinPool在高并发任务调度中比无脑抢一个公共队列的模型拥有更好的吞吐。3.4 方法选择invoke、submit、execute、fork、join的区别方法适用位置返回结果等待方式invoke(task)外部线程直接返回结果阻塞当前线程直到完成submit(task)外部线程返回ForkJoinTask可后续getexecute(task)外部线程无不等待fork()任务内部分治无立即返回异步入队join()任务内部等待fork结果返回子任务结果协作等待会找活干这里额外提醒一个细节不要从ForkJoinTask里再调用另一个ForkJoinPool的invoke。跨池嵌套调用会让两个池的线程互相等待很容易出现“池A的线程在等池B的任务池B又在忙别的池的任务”的纠缠局面。想嵌套分治就在同一个池里做别再造第二张池子。4. 六类线程控制坑位我踩过也帮别人排过的ForkJoinPool表面优雅实际隐藏的坑一点都不少。下面这些是我在真实项目中遇到或者伸手帮忙排查过的每条都能写一段事故报告。4.1 坑一compute()里做阻塞IO工作线程“堵死”了一个池最典型的问题在RecursiveTask.compute()里调了阻塞式的数据库查询、网络请求或Thread.sleep()导致工作线程被长时间挂起而ForkJoinPool的常规工作线程数是围绕parallelism设计的没有那么多线程去“补位”整个池的任务树就会卡住。解决办法不是调大并行度而是使用ForkJoinPool.managedBlock()它允许当前线程在完成特定阻塞式操作时让池动态提供临时补偿线程ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker() { volatile boolean done; Override public boolean block() throws InterruptedException { // 在这里执行阻塞IO doDbQuery(); done true; return true; } Override public boolean isReleasable() { return done; } });4.2 坑二parallelism1你以为并行其实是单线程串行new ForkJoinPool(1)的时候别以为池里还能靠任务内联实现并行。它的含义是工作线程数为1加上外部提交线程也不过两个执行者绝大多数情况下分治树仍然近似串行。并行度设置在核数以内才有意义默认“核数减一”已经是求稳的选择。4.3 坑三递归阈值设得太小任务爆炸比收益更早到来如果数组只有100万个元素你却把阈值设为100会产生约2万个子任务。每个子任务都是对象创建、入队、出队、合并结果都有开销。如果任务本身极其轻量这些固定开销会吞掉并行的收益。一个经验参考每个叶任务的绝对计算时间不要低于100微秒。换算成数组处理阈值从几千到几万通常比较合理。太小则开销大于收益太大则任务粒度太粗、并行度打不满。4.4 坑四ForkJoinPool的工作线程是长生命的ThreadLocal会被“串味”ThreadLocal在普通线程池里也容易脏但ForkJoinPool的工作线程生命周期更长、更稳定一个池可以服务成千上万个并发任务。如果你在任务里写了ThreadLocal又忘了remove()下一个任务会读到上一个任务留下的数据轻则数据错乱重则整个线程状态污染。正确做法是在每个compute()结尾的finally里清理try { localContext.set(traceId); // ... 分治逻辑 } finally { localContext.remove(); }4.5 坑五parallelStream和CompletableFuture共用一个commonPool互相“占坑”很多人喜欢在接口服务里同时用并行流处理批量数据又用CompletableFuture做异步IO。看似互不相干其实它们默认都挤在ForkJoinPool.commonPool()里。一个池被并行流的大任务占满另一个异步调用的等待时间就会飙升。解决思路有两个要么给异步任务自定义专门的ForkJoinPool要么把并行流改成显式的自定义池并行流本身不给你直接指定池的参数需要绕一层forkJoinPool.submit(() - list.parallelStream()...)。总之不要让两拨任务共享一个不可控的公共池。4.6 坑六异常在任务树里被“静默吞掉”只有join/get时才会炸出来ForkJoinTask执行过程中如果抛了异常不会立刻打断其他任务。只有在调用join()或get()时异常才会被重新抛出。而且join()会抛原本的运行时异常get()会包装成ExecutionException。看起来差不多实际上debug体验差别很大。更麻烦的是如果你只fork()了任务但从没join()过它异常可能被完全吞掉。所以规范是每个fork出去的任务都必须被join或取消流程接管。该异常处理的地方不要等到日志里出现诡异的“任务不存在”才回头找。5. 实测复盘一次百万级递归聚合的线程控制压测记录最后用一组我压测时的数据来收尾。任务很简单对1000万个int元素求和。机器是有8个逻辑核心的开发机。拆分逻辑是最经典的递归二分合并阶段就是两个子结果相加。5.1 压测场景与参数设定我分别测了不同并行度、不同阈值组合每个组合跑20次取中位数避免JIT冷启动干扰。实现采用标准的RecursiveTaskLongclass SumTask extends RecursiveTaskLong { private final int[] arr; private final int lo, hi; private final int threshold; SumTask(int[] arr, int lo, int hi, int threshold) { this.arr arr; this.lo lo; this.hi hi; this.threshold threshold; } Override protected Long compute() { int length hi - lo; if (length threshold) { long sum 0; for (int i lo; i hi; i) sum arr[i]; return sum; } int mid (lo hi) 1; SumTask left new SumTask(arr, lo, mid, threshold); left.fork(); SumTask right new SumTask(arr, mid, hi, threshold); long rightResult right.compute(); long leftResult left.join(); return leftResult rightResult; } }5.2 实测结果阈值和并行度的影响并行度阈值中位耗时ms线程数观察110万951个工作线程410万324个工作线程810万218个工作线程81万248个工作线程8100万278个工作线程81000438个工作线程任务数激增明显1610万2216个工作线程但CPU核心只有8个这里的趋势非常清晰并行度从1提到8耗时从95降到21提升约4.5倍阈值过小1000时任务拆分开销反噬收益耗时反而比阈值10万时多了近一倍。并行度超配到16并没有带来额外收益因为CPU核心就8个多出来的线程都在排队等时间片。5.3 三个让我比较意外的结论第一任务拆分也是有天花板的。阈值设到1万时性能反而比10万稍微好一点但低到1000就明显变差。第二8核机器上并行度8和并行度16的差距几乎没有一度让我怀疑设置出问题后来看了CPU占用率才明白物理核已经满负荷多余的线程只能互相穿插。第三结果的合并阶段在求和场景里几乎无感但换成归并排序或更重的合并逻辑后合并占比会迅速抬高这时候压测重点要从“拆了多少任务”转向“合并了多少数据”。5.4 线程控制自检清单每个分治任务上线前建议按这个表过一遍检查项观察重点并行度是否小于等于CPU核数调用线程是否也要参与阈值每个叶子任务耗时是否在100微秒以上阻塞操作compute()里是否IO是否用了ManagedBlocker补偿线程工厂线程名是否可辨识异常处理器是否配置池隔离并行流和异步任务是否挤在同一个commonPoolThreadLocal任务结束是否清理异常处理每个fork是否都有join或取消路径检查列表看着简单但任何一个漏网之鱼都可能在线上重演“线程数翻倍、服务卡死”的故事。那个朋友最后做的事情其实也很简单把IO改成ManagedBlocker给批处理业务单独开了一个自定义ForkJoinPool并行度设为7阈值调回5万ThreadLocal加清理逻辑。再上线后线程数曲线稳成一条直线。回头想fork/join真正让人失控的地方从来不是任务拆分本身而是你误以为它只是一个“自动并行的ExecutorService”。搞清楚线程怎么被派活、怎么等你、怎么偷任务这把武器才算真正上了膛。
前段时间一个朋友拉我过去说他们的批量计算服务每隔几小时就会出现一次“线程数莫名翻倍然后整柜卡住”的怪象。日志里全是java.util.concurrent.ForkJoinPool的痕迹线程转储拉出来一看工作线程几乎全卡在join()调用上。聊到最后发现问题根子不在于join用得太多而在于他们对fork/join家族的线程控制理解得过于理想化——只看到了“分治并行”的甜头没搞清背后线程是怎么被创建、指派、入队、偷取和等待的。这篇文章就把fork/join这套东西的线程控制细节摊开讲一遍包括ForkJoinPool的并行度设计、fork/join的线程流转真相、工作窃取队列、以及我踩过的几种坑和实测数据。适合正在用或者准备用并行流、递归任务、CompletableFuture的开发者尤其是对“为什么线程池会莫名暴涨”“为什么任务偶发性饥饿”这类问题没头绪的朋友。1. 为什么ExecutorService在递归分治场景下会“裁跟头”Fork/Join模型改了什么先回到一个最朴素的问题既然ThreadPoolExecutor已经很成熟了为什么JDK还要在Java 7里引进ForkJoinPool1.1 一个典型的分治任务数组递归聚合假设你要对一个长度上千万的整数数组求和。最简单粗暴的并行方式是把它拆成两半每一半再往下拆直到任务小到值得顺序执行。伪代码大概是if (length THRESHOLD) { return directSum(arr, low, high); } else { int mid (low high) 1; SumTask left new SumTask(arr, low, mid); SumTask right new SumTask(arr, mid, high); // 到这里怎么并行 }如果你用ThreadPoolExecutor最直觉的写法是这样的FutureLong leftFuture pool.submit(left); FutureLong rightFuture pool.submit(right); return leftFuture.get() rightFuture.get();表面上看没错但问题很快就来了。1.2 固定线程池面对递归任务的三层崩溃第一层崩溃在任务数量的不可控。一个上千万的数组拆到阈值1万任务数量直接到千级甚至万级。ThreadPoolExecutor的任务是放阻塞队列的队列一长内存和出队开销都很难看。第二层崩溃在父任务占着线程等子任务。这是最致命的一点。父任务调future.get()时整个工作线程是挂起等待的。假设池里有8个线程每个线程执行到一个需要等待子任务的父任务这8个线程就全“挂”在等结果上子任务反而排不进队列执行形成典型的池饥饿。现实中你可能会看到线程池里的线程数长期不涨、任务堆积或者干脆超时。第三层崩溃是任务之间的“父子局部性”完全丢失。子任务是从队列里被任意线程捡走的父任务刚拆出来的数据可能还在CPU缓存里结果被别的线程拿去算缓存命中率直线下降。1.3 ForkJoinPool的三个关键改变ForkJoinPool针对这三层问题做了三个核心设计任务内联执行工作线程拆出子任务后不会傻等。它会自己动手执行其中一个子任务另一个放进队列等待时还能去“偷”别的任务来干。线程不会被空等浪费。工作窃取队列每个工作线程维护自己的双端队列本地任务按“后进先出”处理空闲线程按“先进先出”从别人队里偷任务既减少争抢又保持了父任务的数据局部性。协作式调度取代抢占式调度普通线程池是“任务排队、线程抢”ForkJoinPool是“任务分治、线程协作”任务不再是孤立单元而是有父子关系、可合并结果的计算树。所以ForkJoinPool不是简单地把固定线程池换了个名字而是把“线程控制”的粒度从外部提交下降到了内部任务协作。理解这一点后文的所有坑都可以顺着这条主线推出来。2. 并行度的数字游戏CommonPool默认值、自定义池、线程工厂与补偿机制“控制线程”第一件事就是控制线程数量。ForkJoinPool最核心的参数叫parallelism中文文档里叫并行度。很多人以为它等同普通线程池的corePoolSize这个理解不准确。2.1 parallelism到底控制什么在ForkJoinPool内部工作线程数组的大小围绕parallelism构建。它决定的是**「常用工作线程」的上限**并不是一个“核心数加临时数”的动态范围。更准确地说它定义了池中活动的工作线程数量目标。// 创建一个并行度为4的池 ForkJoinPool pool new ForkJoinPool(4); ForkJoinTaskLong task new SumTask(arr, 0, arr.length); long result pool.invoke(task);这里pool.invoke(task)会阻塞当前调用线程直到整个分治树算完。注意当前调用线程并不是干等它会以“提交者/参与者”的身份混进执行也参与窃取任务。这也是为什么默认并行度要“核数减一”——因为调用线程自己也要出一份力。2.2 commonPool默认值的数学来源日常用parallelStream()或者CompletableFuture默认的异步任务时根本没有创建池默认用的就是ForkJoinPool.commonPool()。这个公共池的默认并行度有严格的计算规则// ForkJoinPool源码中的默认逻辑简化 int parallelism Math.max(1, Runtime.getRuntime().availableProcessors() - 1);所以一台8核机器上默认并行度是7不是8。JVM留出1个处理器的余量给调用线程、主线程或外部系统。但很多人并不知情于是会出现“8核机器只有7个线程在处理我的并行流”的情况。如果你确实需要改有两个途径# 方式一启动参数 -Djava.util.concurrent.ForkJoinPool.common.parallelism16// 方式二代码里读取常用并行度 int currentParallelism ForkJoinPool.getCommonPoolParallelism(); // 实际使用时应避免在运行期动态修改公共池属性2.3 自定义池时怎么定并行度如果是自建ForkJoinPool并行度选择要分场景场景推荐并行度原因纯CPU密集的分治计算等于或略小于CPU核数每个工作线程都在算超配只会增加上下文切换CPU密集但调用线程也参与核数减1调用线程自己也能干活池线程减一个形成互补任务内存在等待/轻量阻塞核数乘以1.5~2补偿部分线程因等待浪费的时间任务内有重度阻塞IO不建议硬扛用ManagedBlocker见4.1单纯加大线程数会让任务队列膨胀阻塞还是阻塞这个表格不是拍脑袋是我在实际压测里反复试出来的结论。特别是最后一条只要有重的阻塞IO就不要靠无脑调高并行度去解决问题ForkJoinPool的协作式调度最怕长时间挂起。2.4 线程工厂与异常处理的控制入口ForkJoinPool的构造函数也接受ForkJoinWorkerThreadFactory和UncaughtExceptionHandler。默认线程工厂直接创建“裸线程”名字看着费力。在生产环境我强烈建议自定义线程工厂至少让dump里的线程名能一眼认出是谁家的池ForkJoinPool pool new ForkJoinPool( 4, poolTask - { ForkJoinWorkerThread worker new ForkJoinWorkerThread(poolTask) {}; worker.setName(biz-calc-worker- worker.getPoolIndex()); return worker; }, (thread, throwable) - { System.err.println(worker crashed: thread.getName()); throwable.printStackTrace(); }, false );另外ForkJoinPool没有像ThreadPoolExecutor那样暴露RejectedExecutionHandler它继承的是AbstractExecutorService池关闭后再提交任务会抛RejectedExecutionException。这一点很多人不知道以为池关了还会默默接收任务结果在停机时遇到奇怪的提交异常。3. fork与join的线程流转真相谁在派活、谁在等结果、谁在偷任务很多熟悉ExecutorService的人刚接触fork/join时会把fork()脑补成“开两个线程执行”把join()脑补成future.get()。这两条理解都是错的而且会导致你日后debug无从下手。3.1 fork()不是创建线程而是“把任务放到队列里”task.fork()的语义是把当前任务提交给当前工作线程的私有任务队列。如果没有线程归属比如外部线程直接调用fork则会抛异常或者进入全局提交队列。我用一个生活化的类比来解释你正在柜台后面做菜做菜过程中需要洗菜和切菜。你不会另外去雇两个人来帮你做这两件事你只是把洗菜这件事先放到手边的待办筐里然后自己继续切菜。等你手头空下来再从筐里拿来做。// 常见错误理解 left.fork(); // 以为这里开了两个线程 right.fork(); return left.join() right.join(); // ForkJoin官方推荐的范式 left.fork(); // left入队 long rightResult right.compute(); // 当前线程直接执行right long leftResult left.join(); // 取回left的结果 return leftResult rightResult;为什么推荐“fork一个、compute一个”而不是“两个都fork”因为两个都fork时当前线程没有活干只能join等待等于放弃了参与计算而只fork一个、自己算另一个当前线程全程在算计算密度更高。invokeAll()内部其实也是类似的思路先fork第一个再invoke第二个然后把前者join回来。3.2 join()的等待机制不是简单的线程sleepjoin()的底层等待并不是普通阻塞。工作线程在执行join时如果发现自己等待的任务还没完成它会主动去自己的队列尾部继续取任务执行或者去其他线程的队列偷任务。只有当实在是没任务可偷、任务树已经收敛到差一步时才会真正挂起。这个机制说白了一句话工作线程不愿闲着它把“等待子任务”的时间用来干别的活。这和“线程阻塞然后被唤醒”的模型完全不同。所以当你看到线程dump里join()栈的时候别急着下结论说是死锁先看看它是不是正在别的任务里“以waiting为由偷着干”。3.3 工作窃取为什么能从队尾偷出高效率每个工作线程有一个双端队列。分治产生的子任务常规情况下是压入当前线程的队尾当前线程取任务时从队尾取后进先出。而别的线程来偷任务时从队首取先进先出。自己从队尾取刚拆出来的子任务数据大概率还在CPU缓存里局部性好且和父任务的合并过程天然衔接。别人从队首取队首任务是更早拆出来的“老任务”离“新鲜数据”最远被偷走对原线程的局部性伤害最小。正是这种“自己拿最新的别人偷最老的”设计让ForkJoinPool在高并发任务调度中比无脑抢一个公共队列的模型拥有更好的吞吐。3.4 方法选择invoke、submit、execute、fork、join的区别方法适用位置返回结果等待方式invoke(task)外部线程直接返回结果阻塞当前线程直到完成submit(task)外部线程返回ForkJoinTask可后续getexecute(task)外部线程无不等待fork()任务内部分治无立即返回异步入队join()任务内部等待fork结果返回子任务结果协作等待会找活干这里额外提醒一个细节不要从ForkJoinTask里再调用另一个ForkJoinPool的invoke。跨池嵌套调用会让两个池的线程互相等待很容易出现“池A的线程在等池B的任务池B又在忙别的池的任务”的纠缠局面。想嵌套分治就在同一个池里做别再造第二张池子。4. 六类线程控制坑位我踩过也帮别人排过的ForkJoinPool表面优雅实际隐藏的坑一点都不少。下面这些是我在真实项目中遇到或者伸手帮忙排查过的每条都能写一段事故报告。4.1 坑一compute()里做阻塞IO工作线程“堵死”了一个池最典型的问题在RecursiveTask.compute()里调了阻塞式的数据库查询、网络请求或Thread.sleep()导致工作线程被长时间挂起而ForkJoinPool的常规工作线程数是围绕parallelism设计的没有那么多线程去“补位”整个池的任务树就会卡住。解决办法不是调大并行度而是使用ForkJoinPool.managedBlock()它允许当前线程在完成特定阻塞式操作时让池动态提供临时补偿线程ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker() { volatile boolean done; Override public boolean block() throws InterruptedException { // 在这里执行阻塞IO doDbQuery(); done true; return true; } Override public boolean isReleasable() { return done; } });4.2 坑二parallelism1你以为并行其实是单线程串行new ForkJoinPool(1)的时候别以为池里还能靠任务内联实现并行。它的含义是工作线程数为1加上外部提交线程也不过两个执行者绝大多数情况下分治树仍然近似串行。并行度设置在核数以内才有意义默认“核数减一”已经是求稳的选择。4.3 坑三递归阈值设得太小任务爆炸比收益更早到来如果数组只有100万个元素你却把阈值设为100会产生约2万个子任务。每个子任务都是对象创建、入队、出队、合并结果都有开销。如果任务本身极其轻量这些固定开销会吞掉并行的收益。一个经验参考每个叶任务的绝对计算时间不要低于100微秒。换算成数组处理阈值从几千到几万通常比较合理。太小则开销大于收益太大则任务粒度太粗、并行度打不满。4.4 坑四ForkJoinPool的工作线程是长生命的ThreadLocal会被“串味”ThreadLocal在普通线程池里也容易脏但ForkJoinPool的工作线程生命周期更长、更稳定一个池可以服务成千上万个并发任务。如果你在任务里写了ThreadLocal又忘了remove()下一个任务会读到上一个任务留下的数据轻则数据错乱重则整个线程状态污染。正确做法是在每个compute()结尾的finally里清理try { localContext.set(traceId); // ... 分治逻辑 } finally { localContext.remove(); }4.5 坑五parallelStream和CompletableFuture共用一个commonPool互相“占坑”很多人喜欢在接口服务里同时用并行流处理批量数据又用CompletableFuture做异步IO。看似互不相干其实它们默认都挤在ForkJoinPool.commonPool()里。一个池被并行流的大任务占满另一个异步调用的等待时间就会飙升。解决思路有两个要么给异步任务自定义专门的ForkJoinPool要么把并行流改成显式的自定义池并行流本身不给你直接指定池的参数需要绕一层forkJoinPool.submit(() - list.parallelStream()...)。总之不要让两拨任务共享一个不可控的公共池。4.6 坑六异常在任务树里被“静默吞掉”只有join/get时才会炸出来ForkJoinTask执行过程中如果抛了异常不会立刻打断其他任务。只有在调用join()或get()时异常才会被重新抛出。而且join()会抛原本的运行时异常get()会包装成ExecutionException。看起来差不多实际上debug体验差别很大。更麻烦的是如果你只fork()了任务但从没join()过它异常可能被完全吞掉。所以规范是每个fork出去的任务都必须被join或取消流程接管。该异常处理的地方不要等到日志里出现诡异的“任务不存在”才回头找。5. 实测复盘一次百万级递归聚合的线程控制压测记录最后用一组我压测时的数据来收尾。任务很简单对1000万个int元素求和。机器是有8个逻辑核心的开发机。拆分逻辑是最经典的递归二分合并阶段就是两个子结果相加。5.1 压测场景与参数设定我分别测了不同并行度、不同阈值组合每个组合跑20次取中位数避免JIT冷启动干扰。实现采用标准的RecursiveTaskLongclass SumTask extends RecursiveTaskLong { private final int[] arr; private final int lo, hi; private final int threshold; SumTask(int[] arr, int lo, int hi, int threshold) { this.arr arr; this.lo lo; this.hi hi; this.threshold threshold; } Override protected Long compute() { int length hi - lo; if (length threshold) { long sum 0; for (int i lo; i hi; i) sum arr[i]; return sum; } int mid (lo hi) 1; SumTask left new SumTask(arr, lo, mid, threshold); left.fork(); SumTask right new SumTask(arr, mid, hi, threshold); long rightResult right.compute(); long leftResult left.join(); return leftResult rightResult; } }5.2 实测结果阈值和并行度的影响并行度阈值中位耗时ms线程数观察110万951个工作线程410万324个工作线程810万218个工作线程81万248个工作线程8100万278个工作线程81000438个工作线程任务数激增明显1610万2216个工作线程但CPU核心只有8个这里的趋势非常清晰并行度从1提到8耗时从95降到21提升约4.5倍阈值过小1000时任务拆分开销反噬收益耗时反而比阈值10万时多了近一倍。并行度超配到16并没有带来额外收益因为CPU核心就8个多出来的线程都在排队等时间片。5.3 三个让我比较意外的结论第一任务拆分也是有天花板的。阈值设到1万时性能反而比10万稍微好一点但低到1000就明显变差。第二8核机器上并行度8和并行度16的差距几乎没有一度让我怀疑设置出问题后来看了CPU占用率才明白物理核已经满负荷多余的线程只能互相穿插。第三结果的合并阶段在求和场景里几乎无感但换成归并排序或更重的合并逻辑后合并占比会迅速抬高这时候压测重点要从“拆了多少任务”转向“合并了多少数据”。5.4 线程控制自检清单每个分治任务上线前建议按这个表过一遍检查项观察重点并行度是否小于等于CPU核数调用线程是否也要参与阈值每个叶子任务耗时是否在100微秒以上阻塞操作compute()里是否IO是否用了ManagedBlocker补偿线程工厂线程名是否可辨识异常处理器是否配置池隔离并行流和异步任务是否挤在同一个commonPoolThreadLocal任务结束是否清理异常处理每个fork是否都有join或取消路径检查列表看着简单但任何一个漏网之鱼都可能在线上重演“线程数翻倍、服务卡死”的故事。那个朋友最后做的事情其实也很简单把IO改成ManagedBlocker给批处理业务单独开了一个自定义ForkJoinPool并行度设为7阈值调回5万ThreadLocal加清理逻辑。再上线后线程数曲线稳成一条直线。回头想fork/join真正让人失控的地方从来不是任务拆分本身而是你误以为它只是一个“自动并行的ExecutorService”。搞清楚线程怎么被派活、怎么等你、怎么偷任务这把武器才算真正上了膛。
