bigywater.tech
Java · 233

ManagedBlocker:让 ForkJoinPool 正确处理阻塞

ForkJoinPool 的并行度是固定的,任务一旦阻塞就会占住线程。ManagedBlocker 让池知道线程要阻塞、可以补充线程。这篇真实运行这套机制,并逐个测 JDK 常见的阻塞 API 到底会不会补偿。

432 → 110 ms
16 个阻塞 100 ms 的任务:普通阻塞 → managedBlock
RejectedExecutionException
maximumPoolSize 小于需要时 managedBlock 抛出的异常
2 与 6
阻塞的任务不补偿 / 补偿时池里的线程数(并行度 2、5 个任务)

普通阻塞与 managedBlock

实测 · 本机执行输出
  并行度 4 的 ForkJoinPool,16 个各 sleep 100 ms 的任务(普通阻塞,不用 ManagedBlocker):耗时 432 ms(理论 16/4×100 = 400 ms),池里线程数峰值 = 4
  同样的任务,用 ForkJoinPool.managedBlock 包住 sleep:耗时 110 ms,池里线程数峰值 = 17(池补充了线程,让阻塞的任务不占用并行度)
  在普通线程(非 FJP 线程)里调 managedBlock:isReleasable 被调用 1 次,block 被调用 1 次(先问 isReleasable,为 false 才调用 block,block 返回 true 就结束)
  isReleasable 一开始就返回 true:block 被调用 0 次(不阻塞,没有补充线程)

补充线程的上限

实测 · 本机执行输出
  并行度 4、maximumPoolSize = 4 :32 个各阻塞 100 ms 的任务(managedBlock):有任务失败:RejectedExecutionException: Thread limit exceeded replacing blocked worker;线程数峰值 4
  并行度 4、maximumPoolSize = 6 :32 个各阻塞 100 ms 的任务(managedBlock):有任务失败:RejectedExecutionException: Thread limit exceeded replacing blocked worker;线程数峰值 6
  并行度 4、maximumPoolSize = 12:32 个各阻塞 100 ms 的任务(managedBlock):有任务失败:RejectedExecutionException: Thread limit exceeded replacing blocked worker;线程数峰值 12
  并行度 4、maximumPoolSize = 40:32 个各阻塞 100 ms 的任务(managedBlock):全部完成,耗时 112 ms;线程数峰值 33
  默认构造(并行度 4)300 个各阻塞 50 ms 的任务:耗时 152 ms,线程数峰值 = 301

嵌套等待

实测 · 本机执行输出
  并行度 2:两个外层任务各自再向同一个池提交内层任务并 Future.get(1 s):外层结果 = [inner, inner](没有出现超时)
  并行度 2:外层任务在池内用 CompletableFuture.supplyAsync(…, 同一个池) 再 get(1 s):[inner-cf, inner-cf]
  对照:普通的 2 线程 FixedThreadPool 同样嵌套提交并 Future.get(500 ms):[TimeoutException, TimeoutException](线程都被外层任务占住,内层排不上队)
  公共池 parallelStream 里 16 个 sleep(100) 的元素(普通阻塞):耗时 218 ms,公共池并行度 = 9,用到的线程数 = 10
  同样的 parallelStream 用 managedBlock 包住 sleep:耗时 111 ms,用到的线程数 = 16

JDK 的哪些阻塞 API 会补偿

实测 · 本机执行输出
  并行度 2 的池里,6 个任务都在 gate.join()(CompletableFuture.join 内部用了 managedBlock):阻塞期间池里线程数 = 7,getActiveThreadCount = 0,峰值 = 7;全部完成耗时 315 ms
  (对照)并行度 2 的池里 6 个任务阻塞在 CountDownLatch.await():池里线程数 = 2,排队任务数 = 4
  5 个任务在并行度 2 的池里 Phaser(6).arriveAndAwaitAdvance() 阻塞:池里线程数 = 6,排队任务 = 0
  5 个任务在并行度 2 的池里 CyclicBarrier(6).await() 阻塞:池里线程数 = 6,排队任务 = 0
  5 个任务在并行度 2 的池里 Semaphore.acquire() 阻塞:池里线程数 = 2,排队任务 = 3
  5 个任务在并行度 2 的池里 LinkedBlockingQueue.take() 阻塞:池里线程数 = 6,排队任务 = 0
  5 个任务在并行度 2 的池里 ReentrantLock.lock() 阻塞(锁被主线程占着):池里线程数 = 2,排队任务 = 3

分步演示

ManagedBlocker
本步的对象

    本文用到的工具与边界

    • 本机 macOS 27 arm64 / OpenJDK(Temurin 17.0.20、Homebrew 21.0.12 与 27,每篇注明实际用的版本);用 javac 编译、java 直接运行,需要第三方库时用 Maven 解析依赖;计时用 System.nanoTime,数字每次运行会有小幅抖动。JDK 21.0.12,10 核 macOS
    • 没有验证:每个 API 补偿的具体源码路径(只测了现象)、公共池的补偿上限(没有测)、saturate 谓词参数、虚拟线程调度器的补偿(它也基于 ForkJoinPool,见虚拟线程相关的篇目)、JDK 17 与 27 的差别
    京ICP备2026067502号-1京公网安备11010502063487号