嵌套ParallelStream死锁问题：JVM资源管理层面如何处理？

在Java里嵌套使用并行流通常不被推荐，理论上可能因为耗尽ForkJoinPool资源导致死锁：外层流等待内层流完成，但内层流拿不到线程资源无法执行，最终陷入循环等待。但实际场景里，默认池下哪怕迭代百万次也很难复现死锁，只有指定了小并行度的自定义ForkJoinPool时，才容易触发全线程阻塞的死锁。这背后的核心差异在于ForkJoinPool的资源管理机制：

一、默认公共ForkJoinPool的处理逻辑 #

默认并行流使用ForkJoinPool.commonPool()，它的并行度默认等于CPU核心数减1（可通过java.util.concurrent.ForkJoinPool.common.parallelism系统属性修改）。关键的设计是工作线程的"任务窃取与自执行"机制：
当线程执行外层并行流的任务时，若遇到嵌套的并行流（本质是提交新的ForkJoin任务），当前线程不会阻塞等待线程池分配新线程，而是直接接手执行内层的任务。这种"自执行"避免了线程资源的耗尽——外层任务线程不会闲置等待，而是直接处理内层任务，因此不会出现所有线程都阻塞等待内层流的情况，自然很难触发死锁。

二、自定义有限线程数ForkJoinPool的死锁触发原因 #

当我们显式创建并行度极低的ForkJoinPool（比如示例中的并行度2），死锁容易触发的原因在于：

1. 外层并行流会占满池里的所有工作线程；
2. 每个外层线程在执行任务时，遇到内层并行流会尝试从池里获取空闲线程，但此时池里已无可用线程；
3. 如果外层线程没有采用"自执行"内层任务的逻辑（或者内层任务数量过多，当前线程无法快速处理完），外层线程会阻塞等待内层流完成；
4. 所有线程都陷入"等待内层任务完成，但内层任务无线程可执行"的循环，最终导致死锁。

示例代码里的TimeUnit.SECONDS.sleep(5)模拟了长时间任务，让外层线程持续占用资源，内层任务无法被窃取执行，直接触发了死锁场景。

默认池场景测试代码 #

List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);

numbers.parallelStream().forEach(i -> {
    numbers.parallelStream().forEach(j -> {
        System.out.println(i * j);
    });
});

指定线程池场景测试代码 #

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

public class DeadlockDemo {
    public static void main(String[] args) {
        ForkJoinPool forkJoinPool = new ForkJoinPool(2); // 限制线程数为2
 
        forkJoinPool.submit(() ->
            IntStream.range(1, 100).parallel().forEach(i -> {
                System.out.println("Task " + i + " started");
                // 每个任务启动另一个并行流
                IntStream.range(1, 100).parallel().forEach(j -> {
                    System.out.println("Subtask " + j + " started");
                    try {
                        TimeUnit.SECONDS.sleep(5); // 模拟耗时工作
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                    System.out.println("Subtask " + j + " finished");
                });
                System.out.println("Task " + i + " finished");
            })
        );
 
        forkJoinPool.shutdown();
        try {
            forkJoinPool.awaitTermination(1, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

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