如何用ForkJoinPool/Executors.newWorkStealingPool解决线程池挂起并提升性能？

问题复现与原因 #

以下是复现线程池挂起的Kotlin代码：

fun main(args: Array<String>) {
    val start = System.currentTimeMillis()
    Internal().doWork()
    println("Duration is ${(System.currentTimeMillis() - start)/1000} sec")
}

class Internal {

    fun doWork() {

        val pool = ThreadPoolExecutor(
            3, Integer.MAX_VALUE,
            60L, TimeUnit.SECONDS,
            ArrayBlockingQueue(1000),
        )

        val future = CompletableFuture.supplyAsync(
            {
                // 1 subtask
                val future1 = CompletableFuture.supplyAsync(
                    {
                        (1..10).map {
                            CompletableFuture.supplyAsync(SingleExternalCall(), pool)
                        }.sumOf { it.join() }
                    },
                    pool,
                )
                // 2 subtask
                val future2 = CompletableFuture.supplyAsync(
                    {
                        (1..5).map {
                            CompletableFuture.supplyAsync(SingleExternalCall(), pool)
                        }.sumOf { it.join() }
                    },
                    pool,
                )
                // aggregate
                future1.join() + future2.join()
            },
            pool,
        )
        println(future.join())
    }

    class SingleExternalCall : Supplier<Int> {

        override fun get(): Int {
            Thread.sleep(5000)
            return counter.incrementAndGet().toInt()
        }
    }

    companion object {

        private val counter = AtomicLong()
    }
}

代码挂起的核心原因：

• 核心线程数设为3，三个线程全部进入等待状态：
  • 线程1等待第一个子任务的10个外部调用完成（sumOf { it.join() }）
  • 线程2等待第二个子任务的5个外部调用完成（sumOf { it.join() }）
  • 线程3等待两个子任务的结果聚合（future1.join() + future2.join()）
• 实际执行外部调用的任务被放入队列，但队列容量（1000）远大于待执行任务数，ThreadPoolExecutor不会创建新线程，导致无可用线程执行队列任务，最终线程池挂起。

疑问解答 #

1. 该场景下使用ForkJoinPool/Executors.newWorkStealingPool()是否合适？ #

非常合适。

ForkJoinPool的核心设计目标就是处理嵌套、有依赖的任务，它的工作窃取（Work Stealing）机制能完美解决当前线程池挂起问题：当某个线程因等待子任务结果空闲时，会主动从其他线程的任务队列中窃取未执行任务来执行，避免线程闲置、任务无法推进的情况。

且不需要重写整个应用——CompletableFuture的supplyAsync方法支持指定自定义线程池，只需将原来的ThreadPoolExecutor替换为ForkJoinPool实例，现有任务中的CompletableFuture和join()调用完全无需修改。

2. 最佳API是什么？ #

推荐使用Executors.newWorkStealingPool()，它是ForkJoinPool的封装实现，默认以CPU核心数作为并行度，也可通过参数指定并行度（比如newWorkStealingPool(4)）。

如果需要更精细控制（比如设置线程名称、异常处理器等），也可以直接创建ForkJoinPool实例，示例如下：

// 方式1：使用封装好的newWorkStealingPool
val pool = Executors.newWorkStealingPool()

// 方式2：直接创建ForkJoinPool，指定并行度和线程工厂
val pool = ForkJoinPool(
    Runtime.getRuntime().availableProcessors(),
    ForkJoinPool.defaultForkJoinWorkerThreadFactory,
    null,
    true // 异步模式，适合IO密集型任务
)

替换后，原有任务逻辑无需修改，ForkJoinPool的工作窃取机制会自动处理嵌套任务的依赖等待，避免线程挂起。

其他方案对比 #

• 增加核心线程数：无法确定合适上限，任务嵌套层级变化时可能再次出现挂起问题，扩展性差。
• Executors.newCachedThreadPool()：会无限制创建新线程，任务量较大时可能导致系统资源耗尽（CPU、内存过载），风险较高。

内容的提问来源于stack exchange，提问作者gstackoverflow