ForkJoinPool实战：工作窃取原理、并行流陷阱与调优指南

这个系列写到第14篇后台话题最多的还是并发这一块。很多人问过“AtomicInteger线程安全吗”“HashMap线程安全吗”“并行流为什么越跑越慢”这类问题其实都指向同一个底层工具——ForkJoinPool。它是Java并发包里最“反常”的一个线程池别人都在控制线程数量它在控制任务拆分方式别人追求队列排队它鼓励线程互相偷活干。 标题里提到的“智能厨房”是对它工作机制非常贴切的类比一个大菜大任务先分成几道工序子任务哪口灶台工作线程闲了就去别人的备菜台队列里顺点活儿干而不是傻等着自己那份做完。这篇文章就把这个池子的原理、写法、调优和踩坑一次性说透。 这个系列前面几篇都是从一个高频面试题切入这篇我准备从更实战的角度出发先分析为什么传统线程池处理“分治”类任务很别扭再拆解ForkJoinPool的工作窃取机制放几段可以直接跑的代码最后把我压测和项目落地时遇到过的坑列出来。1. 先看本质为什么“分治”这种任务模型会让普通线程池很吃力1.1 分治任务的天然形态父子关系分治算法写起来很直观一个大问题拆成两个或多个小问题小问题再拆直到小到可以直接求解然后逐层把结果汇总。归并排序、二分查找、树形结构遍历、大规模矩阵运算都是这个套路。这种任务在并发执行时有一个非常显著的特点任务是带父子关系的树状结构。比如要算一个长度为1000万的数组总和先拆成两个500万两个500万再各拆成两个250万……直到每个任务小到可以顺序求和为止。传统线程池的模型是什么任务之间是平级的你提交一个Runnable或Callable线程池从队列里取出一个任务执行。它不知道任务之间谁等谁也不知道某个任务拆出来多少个子任务。这种模型处理“一批批量任务”很好用但处理“一棵任务树”就很别扭。1.2 普通线程池处理分治任务的三个困境我最早用ExecutorService硬写过归并排序的并行版本写完之后简直是折磨核心问题有三个。第一个困境是线程数量不知道该设多少。分治任务的子任务数量是动态增长的一个拆两个两个拆四个排山倒海一样往里涌。如果固定线程池的核心线程数是8多出来的任务只能在无界队列里堆着。更麻烦的是父任务必须等到子任务全部完成才能汇总而这些等待中的父任务还在占着线程不放。第二个困境是忙闲极度不均。分治任务拆出的子任务真实工作量并不总是均匀的。比如处理树形目录有的分支只有几个文件有的分支有几万个文件比如图像处理的分块某一块算法收敛慢另外几块早就算完了。固定切块提交给普通线程池之后快的线程把队列里的任务抢完就去休息慢的线程还在吭哧吭哧算整体并行度非常低。第三个困境是join等待浪费。父任务提交子任务之后需要调用Future.get()等结果这个等待过程会一直占着一个工作线程。典型场景是8个线程的任务其中7个在等待某个慢分支的结果只有1个在真正干活。这不仅浪费CPU资源还有可能因为两个任务互相等待子任务而出现“饥饿式卡顿”表现就是CPU占用不高但任务迟迟跑不完。1.3 ForkJoinPool的设计师看到了什么ForkJoinPool的设计目标就是解决上面三个问题。它不把任务看作一个独立的工作单元而是看成一棵“可以继续拆分的活树”。每个工作线程不依赖全局队列分配任务而是维护自己的本地队列当一个线程发现自己负责的大任务拆完了本地队列空了它不是闲下来而是去别的线程队列尾部“偷”一个任务过来跑。这套机制的名字叫“工作窃取”Work Stealing它是整个ForkJoinPool的灵魂。JDK 7引入这个东西时很多人在网上写博客都说“ForkJoinPool就是用来做递归拆分的”这个说法没错但没讲到点子上。拆分子任务只是表面真正的价值在于它让所有线程尽量保持忙碌同时不让任何线程因为等待子任务而空转。2. 工作窃取与双端队列ForkJoinPool的“智能厨房”到底在忙什么2.1 用厨房来理解这个池子标题里的“智能厨房”比喻很准确我实际给人讲ForkJoinPool时也这么打比方。想象一个餐厅后厨总厨调用方接到一单大席要做500桌菜。他不可能一口锅全炒出来于是定了个规矩每桌菜拆成冷菜、热菜、汤品三个部分安排三个厨师各管一类。每个厨师发现自己那部分还是太多又拆成更细的小份谁做完谁举手。厨房里每个厨师身边有一个自己的备菜台已经切好的菜放在自己手边。厨师做菜有一个习惯先从自己台面上拿最容易做的做实在没活干了也不坐着等而是直接走到隔壁台子从最底下抽一个半成品回来接着做。这就是工作窃取的日常版本。对应到ForkJoinPool每个工作线程有一个存放任务的WorkQueue这个队列是双端的线程自己从队列尾部取任务执行相当于厨师拿自己手边容易做的菜空闲线程从队列头部“偷”别的线程的任务来跑相当于去别人台子底下抽半成品。不同的是被偷的线程不用担心自己的活被抢光因为队列头部往往放着体积最大、拆分最浅的任务拿走之后自己手边更细的小任务不受影响。2.2 双端队列的四个关键操作具体到实现上ForkJoinPool内部每个WorkQueue有四个核心操作push(task)把新任务放到队列尾部。pop()当前线程从队列尾部取出任务执行。poll()其他线程从队列头部偷走任务执行。top和base两个指针维护队首和队尾位置。为什么自己从尾部取别人从头部偷因为分治任务有一个规律越靠近任务树根部任务量越大越靠近叶子任务量越小。当前线程刚刚fork出来的子任务如果自己也从尾部取拿到的就是最近拆出来的小任务而偷任务的线程从头部拿走的则是被压在底下的大任务块可以继续拆分、继续扩散。这样整个池子的任务粒度会自动趋于平衡不会出现一个大任务卡在一个线程上半天拆不完。2.3 fork和join到底干了什么网上不少文章把fork()说成“给子任务开一个新线程执行”这个说法是不准确的也是很多人理解错ForkJoinPool的起点。fork()的JDK源码语义是把当前任务放入当前工作线程的本地队列尾部然后立即返回。它并不承诺立刻异步执行更不保证由一个新线程执行。真正执行子任务的时机可能是当前线程后续自己取出来做也可能是某个空闲线程从本地队列偷走做。join()也不是傻等。它要等待子任务完成并拿到结果但等待期间当前线程并不会一直挂起而是会运行helpJoin逻辑反复从自己的队列或其他队列中取出任务来执行直到要等的结果就绪。这就解决了普通线程池最头疼的“join等待浪费线程”问题。我见过一个很经典的错误写法在compute()里调了left.fork()又写了right.fork()然后left.join()、right.join()看起来没问题但这个写法等于把两个子任务全部推给队列当前线程在join前反而没事干。更合理的写法是left.fork()之后自己直接right.compute()再left.join()。这样当前线程始终在参与真实计算任务树更扁平。2.4 为什么LIFO比FIFO更适合ForkJoinPool普通线程池的全局队列是FIFO的先提交的任务先执行这保证了公平性。但ForkJoinPool的本地队列走LIFO后进先出也就是自己线程优先拿最新放入的任务而不是最早放入的。原因很简单最新放入的任务体积最小、最容易在短时间内解决。当前线程做完这个小任务后继续从尾部拿下一个本地队列里的任务会像剥洋葱一样一点点变小。而大任务被压在队列头部等着其他空闲线程来偷。这样偷与不偷、看与不看任务粒度都被动态调整到最合适的水平。我写过一个极端的对比测试同一个分治累加任务把WorkQueue模拟成FIFO之后8线程并行耗时反而比LIFO慢了一倍。原因正是FIFO下偷任务的线程很容易和地方线程抢同一个大任务冲突加剧缓存命中率也下降。3. RecursiveTask与RecursiveAction一段可复跑的分治求和代码3.1 先选类型有返回值和纯任务的差别写ForkJoinPool任务必须继承两个类之一这是绕不开的第一步。抽象类适用场景核心方法返回值RecursiveTaskV子任务需要返回计算结果比如求和、求最值、统计数量compute()返回V有RecursiveAction子任务只需要执行比如遍历目录、数组排序、数据拷贝compute()返回void无选错类型会影响代码结构但两者的执行机制完全一样底层都是同一个任务队列和工作窃取逻辑。3.2 一个完整的求和任务先看最经典的分治求和。目标对一个长度为1000万的long数组求和。public class SumTask extends RecursiveTaskLong { private static final int THRESHOLD 10_000; private final long[] array; private final int start; private final int end; public SumTask(long[] array, int start, int end) { this.array array; this.start start; this.end end; } Override protected Long compute() { if (end - start THRESHOLD) { long sum 0; for (int i start; i end; i) { sum array[i]; } return sum; } int mid (start end) 1; SumTask left new SumTask(array, start, mid); SumTask right new SumTask(array, mid, end); left.fork(); Long rightSum right.compute(); return rightSum left.join(); } }调用方式long[] data new long[10_000_000]; Arrays.fill(data, 1L); ForkJoinPool pool new ForkJoinPool(); Long result pool.invoke(new SumTask(data, 0, data.length)); pool.shutdown(); System.out.println(sum result);这段代码的运行逻辑是这样的外层传入整个数组区间判断区间长度超过阈值就拆成两半左边任务fork()丢进本地队列右边任务自己动手计算左右都完成后合并结果。实际执行时不一定所有子任务都由同一个线程完成空闲线程会从队列里偷走左边的子任务从而并行起来。3.3 阈值怎么定这个参数比看上去敏感得多阈值是整个分治代码里最容易“想当然”的参数实际影响极大。以我自己的机器和典型数据为例对1000万元素求和阈值设为1拆到单个元素时任务对象数量达到千万级每个任务都要经历一次fork/join队列竞争和对象分配成本压过计算收益耗时会比单线程还慢好几倍。阈值设为10000任务数控制在千级8线程并行耗时明显下降。我的经验是最小任务的实际执行时间最好在微秒级到百微秒级不要低于纳秒级。每个任务对象的创建、入队、出队、join都有固定开销如果任务本身只计算几个加法拆分就是负收益。折中做法是让拆分深度不超过树高的十几层或者直接参考THRESHOLD 数组长度 / (parallelism * 4)来定初始阈值。关于“拆多少层”也有一个经验法则任务拆到大约parallelism * 几十倍的数量级就够了。8线程池拆出200到500个子任务工作窃取就有足够的调度空间。拆出十万、百万个子任务只会让池子花大量时间在队列调度上。3.4 无返回值的典型案例并行扫描大目录递归汇总适合展示RecursiveTask那RecursiveAction我用一个真实项目里的场景来讲扫描目录树找出所有大于50MB的文件。这类任务天然递归而且每层任务数量不确定非常适合分治合并。public class LargeFileScanTask extends RecursiveAction { private final File file; private final ConcurrentLinkedQueueString result; public LargeFileScanTask(File file, ConcurrentLinkedQueueString result) { this.file file; this.result result; } Override protected void compute() { if (file.isFile()) { if (file.length() 50L * 1024 * 1024) { result.add(file.getAbsolutePath()); } return; } File[] children file.listFiles(); if (children null || children.length 0) { return; } ListLargeFileScanTask subTasks new ArrayList(); for (File child : children) { subTasks.add(new LargeFileScanTask(child, result)); } invokeAll(subTasks); } }这里有两个值得注意的细节。第一个result用的是ConcurrentLinkedQueue而不是普通的ArrayList因为多个工作线程会并发写入结果集合HashMap、ArrayList这类非线程安全容器在这里会被直接打穿。第二个invokeAll(subTasks)接收的是整个子任务列表它内部会根据列表长度决定哪些任务fork、哪些任务自己直接compute比手动写“全部fork再全部join”高效得多。4. 实测对比单线程、普通线程池、ForkJoinPool处理一千万累计相加4.1 压测的设计思路有人可能会问求和任务不是用普通线程池固定切块也能并行吗确实能。这里我放了一组实际跑过的压测数据重点不是得出“谁碾压谁”的结论而是看清各自的适用边界。测试环境8核16线程CPUJDK 17数组长度1000万每个元素是随机生成的long值任务就是全部累加。每种方案先预热3次再取5次平均值。为保证公平所有方案都只考虑CPU计算不掺入任何IO和锁。4.2 三种实现的核心代码单线程方案就是普通for循环没什么好说的。普通线程池方案用固定8线程手动把数组切成8段提交int core Runtime.getRuntime().availableProcessors(); ExecutorService service Executors.newFixedThreadPool(core); int batch data.length / core; FutureLong[] futures new Future[core]; for (int i 0; i core; i) { int start i * batch; int end (i core - 1) ? data.length : (i 1) * batch; futures[i] service.submit(new CallableLong() { Override public Long call() { long sum 0; for (int j start; j end; j) { sum data[j]; } return sum; } }); } long total 0; for (FutureLong f : futures) { total f.get(); } service.shutdown();ForkJoinPool方案就是前面SumTask的代码用默认构造器创建池invoke提交。4.3 结果解读谁在什么情况下赢实测数据大致如下方案耗时毫秒说明单线程for循环约42数据量够大但只有单核在算普通线程池固定切块约1516线程跑满任务均匀ForkJoinPool阈值1万约9动态拆分工作窃取负载更均衡ForkJoinPool阈值512约23拆分过细调度成本上升ForkJoinPool阈值100万约18拆分太粗并行度不足纯数组求和这种极度规则的任务普通线程池和ForkJoinPool的差距没有想象中大因为固定切块就已经让所有线程均匀跑满了。ForkJoinPool的真正优势体现在任务工作量不规则的场景比如遍历一颗节点深度参差不齐的树或者在每段数据上做耗时不同的处理这个时候固定切块会出现明显的长尾工作窃取则会把其他线程空闲的算力自动补到慢分支上。4.4 优势不明显时反而要警惕过度设计我在项目里见过不少把ForkJoinPool用错地方的案例。比如某个报表接口总共就1000条数据要做字段拼接有人也套一个RecursiveTask上去结果创建任务对象的时间比计算时间还长接口响应反而更慢了。分治并行的收益公式大概是收益 并行度提升带来的时间节省 − 任务拆分和合并带来的开销。数据量小、任务粒度细的时候第二项会完全吞掉第一项。所以我的建议是任务总耗时低于几毫秒的场景别用ForkJoinPool任务可以自然按树形拆分且单任务耗时较长的场景它才有用武之地。5. CommonPool这只“隐形线程池”如何操纵并行流与CompletableFuture5.1 你写的parallelStream底层就是ForkJoinPool很多人在项目里写过类似这样的代码ListOrder orders getOrders(); orders.parallelStream().map(this::processOrder).collect(Collectors.toList());看起来只是把stream()换成了parallelStream()似乎什么都没多写。但底层已经悄悄创建了一个全局共享的ForkJoinPool实例这个静态实例叫commonPool。CompletableFuture的默认异步方法比如supplyAsync()不传Executor时同样是用它来跑任务。这个设计看似方便却埋着一个隐患全JVM共享同一个池线程数默认是CPU核数减1。8核机器上的commonPool只有7个工作线程。如果某个业务接口里的parallelStream正占着5个线程跑计算另一个接口的CompletableFuture再往池里提交几十个任务它们会挤在一起排队。5.2 commonPool的线程数规则和调整方法commonPool的默认线程数JDK源码里有一个静态初始化过程会读取系统属性java.util.concurrent.ForkJoinPool.common.parallelism。如果显式设置了就用设置的值如果没设置就用Runtime.getRuntime().availableProcessors() - 1但对这些工作线程还有一个限制范围池中线程数最少不会低于1。这个“减1”是有讲究的。ForkJoinPool在执行invoke或fork/join链时调用方线程本身也会参与任务的领取和执行所以不需要让工作线程把CPU核全部占满留一个核心给调用方线程跑任务。如果你确实需要调整可以在JVM启动参数里加-Djava.util.concurrent.ForkJoinPool.common.parallelism16不过我得提醒一句改全局并行度影响范围极大所有parallelStream和默认CompletableFuture都会跟着变属于“开大炮打蚊子”的操作。更重要的是它会把线程数顶到CPU核数以上如果任务里混合了IO操作线程数一多CPU上下文切换非常快就会冲上来。大多数情况下更好的做法是给关键业务单独建一个自定义池而不是去动全局配置。5.3 并行流里嵌套并行流最容易翻车的地方这个坑我在刚用并行流那会儿踩得非常惨。当时有个报表任务外层用parallelStream遍历商品列表每个商品内部又要统计多个维度的指标我图省事内层也写了parallelStream。结果就是外层任务把commonPool的7个线程全部占住内层任务排队等待池中线程释放而外层任务的汇总又必须等内层任务返回最终形成一种任务互相等待的局面。机器CPU占用并不高但任务就是卡着不动看起来像死锁实际上是被commonPool的容量饿死了。排查的过程也值得记一下。我先用jstack抓线程栈发现大量工作线程停在ForkJoinTask的等待逻辑上再一数池子里的线程数量发现始终是7个但等待队列里堆了几十个任务。顺着调用链一看内层又出现了一个parallelStream调用问题根源就清楚了。解决办法也很简单嵌套的任务内层改成串行stream()或者内层不要写在同一个池里。我在实际项目中选择了给内层查询单独建一个自定义的普通线程池而不是嵌套parallelStream。5.4 自定义ForkJoinPool的命名与关闭遇到需要独立池子的场景推荐写成应用级别的单例避免每次调用都new一个池子。一个容易被忽略的细节是线程名字——默认的ForkJoinPool线程名是ForkJoinPool-1-worker-1排错的时候很难分清是哪条业务线。可以用自定义线程工厂来改名ForkJoinPool pool new ForkJoinPool( Runtime.getRuntime().availableProcessors(), pool - { Thread t new Thread(pool, biz-report-worker); t.setDaemon(true); return t; }, null, false );自定义池记得在应用停止或不再使用时调用shutdown()。注意不同点commonPool是static全局的JVM退出时自动结束不需要也尽量不要手动关闭但自己new出来的池子不会自动关闭不调用shutdown()的话整线程会一直存在。6. 三个让我踩过坑的ForkJoinPool陷阱以及对应的排查思路6.1 坑一compute()里执行了阻塞型IO或睡眠这是ForkJoinPool使用过程中代价最高、也最隐蔽的坑。我曾在某个数据同步任务里需要在每个子任务的compute()中调用一个远程接口获取补充数据再继续本地计算。表面上看任务被拆得很细ForkJoinPool也是16核机器的默认池按理说没有问题。但实际跑起来整个任务池几乎占用全部线程剩余提交的任务全部排队接口响应变慢CPU使用率还不高。用jstack一抓栈发现工作线程大部分停在网络IO的等待方法上而不是在算数据。原因很简单工作窃取只能优化CPU密集任务之间的调度它解决不了“一个线程阻塞在IO上”的问题。当一个线程阻塞在compute()内部的远程调用上时它占用的线程位就废了其他线程无法替它执行那个子任务如果大量子任务都阻塞池子里的线程就会被全部耗尽。排查链路非常清楚先用jstack看线程状态数一数有多少线程处于WAITING或BLOCKED再看堆栈是不是挂着SocketInputStream.read、Thread.sleep、lock.lock()这类调用。如果确实要在分治任务里做阻塞操作有几个方向可以考虑用ForkJoinPool.managedBlock()包装阻塞逻辑JDK会在内部补偿线程数来应对阻塞但补偿能力有限不能救所有场景。更直接的做法是重新评估这个任务真的适合ForkJoinPool吗如果IO比例高不如用普通线程池并把线程数按IO密集型任务来设计通常是CPU核数的2到4倍。如果只是短暂的打印、同步块确保锁粒度足够小尽量不要在持有锁时调用外部方法。6.2 坑二子任务里再次new ForkJoinPool提交任务另一个我见过很多次的错误写法在compute()方法里又去new ForkJoinPool()然后提交新的分治任务。表面看代码写得很规整外层一个池内层又一个池。但实际运行起来每拆一层就多一个池池中的工作线程树状扩散。我排查一个线上内存飙升问题时用jstack看到的线程列表里有好几组ForkJoinPool-2-worker-x、ForkJoinPool-3-worker-x命名编号一直往上跳这就说明每次递归都在创建新的线程组线程数量按任务树层级指数增长最终把内存和线程句柄打爆。正确做法是让整个应用共享少数几个池。如果能接受全局影响直接复用commonPool如果希望隔离把自定义池写成静态单例。所有分治任务都只提交到这个池里由工作窃取机制去动态调度绝对不要为了“局部可控”而反复建池。6.3 坑三invokeAll与fork/join的混用顺序最后一个坑不算致命但经常导致性能回退就是子任务提交顺序写得不对。常见的错误是left.fork(); right.fork(); return left.join() right.join();这个写法的最大问题是两个子任务都进了队列当前线程在join之前没有任何计算任务等于白等。虽然join的help机制会去拿别的任务来跑但自己手上明明有活却不干增加了队列交互的损耗。JDK推荐的习惯写法是left.fork(); T rightResult right.compute(); return left.join() rightResult;先fork一个剩下的自己亲手计算。这样当前线程不会空转队列里也只多了一个待偷任务。对于三个及以上子任务的场景更推荐直接用invokeAll(left, right)把它交给JDK内部去平摊fork和compute。我后来在代码里统一成这个风格后同样规模的任务耗时又降了约一成成本只是换个写法。最后给想继续深入的朋友几句掏心窝的话我自己入门ForkJoinPool时最大的误区是一上来就追求“写个最快的并行代码”结果被工作窃取、commonPool、阈值这些问题绕得头晕。后来换了个思路先画任务树再想清楚哪些子树会让线程闲着最后才动手写代码反而很快理解了这套机制。想练习的朋友建议按这个顺序来先跑通数组求和再改成二叉树遍历最后试着改造一个普通线程池实现的批量任务对比看看。面试时被问ForkJoinPool与其背八股不如把你亲手压出来的阈值对比数据和那几次线上踩坑经历讲出来效果会好很多。下一篇我会继续拆解CompletableFuture和ForkJoinPool之间的协作细节这两个东西混用出问题的频率比大多数人想象的高得多。