class MyTask extends RecursiveTask<Long> {
    @Override
    protected Long compute() {
        if (任务够小) {
            return 直接计算();
        }
        MyTask leftTask = new MyTask(左半部分);
        MyTask rightTask = new MyTask(右半部分);
        
        leftTask.fork();    // 异步提交左任务
        Long rightResult = rightTask.compute();  // 当前线程直接执行右任务
        Long leftResult = leftTask.join();       // 等待左任务结果
        
        return merge(leftResult, rightResult);
    }
}

注意：fork()一个任务后，当前线程应该直接compute()另一个子任务，而不是两个都fork()然后都join()。直接compute利用了当前线程，避免不必要的任务调度开销。

2.3 ForkJoinPool的并行度

ForkJoinPool的并行度（parallelism）不等于线程数。parallelism是"可以同时运行的最大线程数"，但ForkJoinPool会在需要时（如任务阻塞等待）创建额外的"补偿线程"（compensation threads），确保总有parallelism个线程在运行。

默认ForkJoinPool.commonPool()的parallelism = Runtime.getRuntime().availableProcessors() - 1。

可以通过JVM参数调整：

-Djava.util.concurrent.ForkJoinPool.common.parallelism=8

三、完整代码实现

3.1 大数组并行求和

package com.laozhang.concurrent.forkjoin;

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;

/**
 * ForkJoinPool实战：大数组并行求和
 *
 * 分治策略：数组长度 > 10000时拆分，否则直接求和
 *
 * 性能对比（JDK 11，8核机器，1亿元素数组）：
 * 串行求和：    约 180ms
 * ForkJoin（8线程）：约 35ms（约5倍提速）
 * 普通线程池（固定8分割）：约 42ms（无法处理数据倾斜）
 */
public class ParallelSumDemo {

    /**
     * 并行求和任务
     */
    static class SumTask extends RecursiveTask<Long> {
        private static final int THRESHOLD = 10_000;  // 拆分阈值
        private final long[] array;
        private final int start;
        private final int end;

        SumTask(long[] array, int start, int end) {
            this.array = array;
            this.start = start;
            this.end = end;
        }

        @Override
        protected Long compute() {
            int length = end - start;

            // 足够小，直接计算
            if (length <= THRESHOLD) {
                long sum = 0;
                for (int i = start; i < end; i++) {
                    sum += array[i];
                }
                return sum;
            }

            // 拆分为两个子任务
            int mid = start + length / 2;
            SumTask leftTask = new SumTask(array, start, mid);
            SumTask rightTask = new SumTask(array, mid, end);

            // 关键：fork左任务（提交给线程池），当前线程直接执行右任务
            leftTask.fork();
            long rightResult = rightTask.compute();  // 当前线程执行
            long leftResult = leftTask.join();       // 等待左任务

            return leftResult + rightResult;
        }
    }

    public static void main(String[] args) throws InterruptedException {
        int size = 100_000_000;
        long[] array = new long[size];
        for (int i = 0; i < size; i++) {
            array[i] = i % 100;
        }

        // 串行基准
        long t1 = System.currentTimeMillis();
        long serialSum = 0;
        for (long v : array) serialSum += v;
        System.out.printf("串行求和：%dms，结果：%d%n",
            System.currentTimeMillis() - t1, serialSum);

        // ForkJoinPool并行
        ForkJoinPool pool = new ForkJoinPool(
            Runtime.getRuntime().availableProcessors());

        long t2 = System.currentTimeMillis();
        long parallelSum = pool.invoke(new SumTask(array, 0, size));
        System.out.printf("ForkJoin并行求和：%dms，结果：%d%n",
            System.currentTimeMillis() - t2, parallelSum);

        System.out.printf("结果验证：%s%n", serialSum == parallelSum ? "通过" : "失败！");
        System.out.printf("加速比：%.2fx%n",
            (double)(System.currentTimeMillis() - t1) / (System.currentTimeMillis() - t2));

        pool.shutdown();
    }
}

3.2 文件目录统计（模拟工作窃取效果）

package com.laozhang.concurrent.forkjoin;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.atomic.LongAdder;

/**
 * ForkJoinPool实战：模拟文件目录统计
 *
 * 模拟场景：
 * - 目录树结构，根节点有若干子目录
 * - 每个目录有若干文件，文件大小随机（模拟数据倾斜）
 * - 统计整个目录树的总文件大小
 *
 * 演示工作窃取的效果：
 * - 某些大目录处理时间长（文件多），任务自然分解
 * - 空闲线程窃取任务帮忙，整体效率更高
 *
 * 测试环境：JDK 11
 */
public class DirectoryStatDemo {

    // 模拟目录节点
    static class DirNode {
        String name;
        List<DirNode> children;
        List<Long> fileSizes;  // 该目录下的文件大小

        DirNode(String name) {
            this.name = name;
            this.children = new ArrayList<>();
            this.fileSizes = new ArrayList<>();
        }
    }

    // 统计任务
    static class DirScanTask extends RecursiveTask<Long> {
        private final DirNode node;
        private static final LongAdder taskCount = new LongAdder();

        DirScanTask(DirNode node) {
            this.node = node;
            taskCount.increment();
        }

        @Override
        protected Long compute() {
            // 当前目录的文件大小
            long dirSize = node.fileSizes.stream().mapToLong(Long::longValue).sum();

            if (node.children.isEmpty()) {
                return dirSize;
            }

            // Fork所有子目录的任务
            List<ForkJoinTask<Long>> subtasks = new ArrayList<>();
            for (DirNode child : node.children) {
                DirScanTask task = new DirScanTask(child);
                subtasks.add(task.fork());
            }

            // 汇总子目录结果
            for (ForkJoinTask<Long> task : subtasks) {
                dirSize += task.join();
            }

            return dirSize;
        }

        public static long getTaskCount() {
            return taskCount.sum();
        }
    }

    /**
     * 生成模拟目录树（故意让某些目录很大，模拟数据倾斜）
     */
    static DirNode generateDirTree(int depth, int maxChildren, Random random) {
        DirNode node = new DirNode("dir_depth" + depth + "_" + random.nextInt(1000));

        // 添加文件（故意让深度为0的节点有更多文件，模拟大文件目录）
        int fileCount = (depth == 0) ? random.nextInt(500) + 100 : random.nextInt(50);
        for (int i = 0; i < fileCount; i++) {
            node.fileSizes.add((long) random.nextInt(1024 * 1024));  // 0-1MB
        }

        // 递归创建子目录
        if (depth > 0) {
            int childCount = random.nextInt(maxChildren) + 1;
            for (int i = 0; i < childCount; i++) {
                node.children.add(generateDirTree(depth - 1, maxChildren, random));
            }
        }

        return node;
    }

    public static void main(String[] args) {
        Random random = new Random(42);
        DirNode root = generateDirTree(4, 5, random);

        System.out.println("目录树生成完成，开始统计...");

        ForkJoinPool pool = new ForkJoinPool(
            Runtime.getRuntime().availableProcessors());

        long start = System.currentTimeMillis();
        long totalSize = pool.invoke(new DirScanTask(root));
        long elapsed = System.currentTimeMillis() - start;

        System.out.printf("总文件大小：%,d bytes (%.2f MB)%n",
            totalSize, totalSize / 1024.0 / 1024.0);
        System.out.printf("处理耗时：%dms%n", elapsed);
        System.out.printf("总任务数：%d，使用线程：%d%n",
            DirScanTask.getTaskCount(),
            pool.getPoolSize());
        System.out.printf("偷取任务数：%d%n", pool.getStealCount());

        pool.shutdown();
    }
}

四、踩坑实录

坑1：在ForkJoinPool中调用阻塞IO导致并行度下降

报错现象： 把ForkJoinPool用于处理包含数据库查询的任务，整体吞吐量比普通线程池还差，CPU利用率很低。

原因分析： ForkJoinPool是为CPU密集型任务设计的。当任务在compute()中调用阻塞IO（如JDBC查询），线程被阻塞，ForkJoinPool会创建补偿线程（ManagedBlocker机制），但补偿线程创建有开销，且系统中线程过多会引发上下文切换。

ForkJoinPool的工作窃取只对可以继续fork的CPU任务有优势。IO任务阻塞期间，线程空转或补偿线程浪费资源。

解法： IO密集型任务用ThreadPoolExecutor；如果必须在ForkJoinPool中做IO，实现ForkJoinPool.ManagedBlocker接口，让ForkJoinPool知道任务会阻塞并创建补偿线程：

ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker() {
    @Override public boolean block() throws InterruptedException {
        result = doBlockingIO();
        return true;
    }
    @Override public boolean isReleasable() {
        return result != null;
    }
});

坑2：任务粒度太细导致开销大于收益

报错现象： 设置了很小的THRESHOLD（如10），ForkJoin的总耗时比串行还慢。

原因分析： 每次fork()都有开销：创建ForkJoinTask对象、队列入队、可能的线程唤醒。如果任务太小（比如10个元素的求和只需要几十纳秒），fork的开销（几微秒）远超任务本身，总开销反而增加。

JVM JIT基准测试数据（JDK 11，JMH）：

fork一个空任务的开销：约2μs
对10个long元素求和：约20ns
当THRESHOLD=10时，fork开销是计算开销的100倍！

解法： THRESHOLD的合理值取决于任务的计算量。对于简单的数值计算，THRESHOLD至少要在1000-10000以上。测量为准，不要凭感觉设置。

坑3：join()死锁：在递归任务中同步等待

报错现象： ForkJoinPool挂死，所有线程都在等待，任务永远无法完成。

原因分析： ForkJoinPool的并行度如果设置为1，一个任务fork出子任务后调用join()等待，但池中只有1个线程（正在等待），子任务永远无法执行，死锁。

即使并行度>1，如果代码里有循环fork（任务A等任务B，任务B等任务C，最终形成等待环），也可能死锁。

解法：

ForkJoinPool并行度不要设为1
遵循"fork左，compute右，join左"的模式，不要出现循环依赖
使用helpComplete()或quietlyJoin()而不是直接join()（更适合某些场景）

坑4：commonPool被阻塞，Stream并行操作受影响

报错现象： 系统中有业务代码使用parallelStream()，同时有ForkJoinTask在运行，两者互相影响，有时stream处理莫名变慢。

原因分析： parallelStream()和CompletableFuture.supplyAsync()（不指定executor时）都使用ForkJoinPool.commonPool()。如果有ForkJoinTask占用了commonPool的所有线程，parallelStream就没有线程可用。

解法： 业务代码不要依赖commonPool，显式指定自定义ForkJoinPool：

ForkJoinPool customPool = new ForkJoinPool(4);
customPool.submit(() -> 
    myList.parallelStream().map(...).collect(...)
).get();

五、总结与延伸

ForkJoinPool vs 普通ThreadPoolExecutor的选型：

维度	ThreadPoolExecutor	ForkJoinPool
适合任务类型	IO密集型、独立任务	CPU密集型、分治任务
任务分解	不支持	原生支持（fork/join）
队列设计	全局共享队列	每线程独立双端队列
工作窃取	无	有（空闲线程主动帮忙）
阻塞支持	良好	较差（需ManagedBlocker）
JDK应用	大多数并发场景	parallelStream、Stream、CompletableFuture默认池

何时用ForkJoinPool：

任务可以被递归分解（分治算法）
CPU密集型计算（图像处理、数值计算、排序）
数据量大但各部分计算量不均匀（利用工作窃取）

JDK 8+的影响： parallelStream()底层就是ForkJoinPool，理解工作窃取有助于正确使用Stream并行。