使用 Fork Join 框架

前言

当需要执行大量的小任务的时候，我们需要将多个小任务进行拆分，类似 快速排序 的 分而治之。

Fork 将一个大任务进行递归，不断的把它切割成符合条件的小任务，然后这些子任务分配到不同 CPU 核心上去分别计算，提高效率，Join 是 获取到小任务的计算结果，最后合并返回。

它充分利用了现在多核 CPU 的性能。

正文

Fork/Join框架的核心类是ForkJoinPool，它能够接收一个ForkJoinTask，并得到计算结果。ForkJoinTask有两个子类，RecursiveTask（有返回值）和RecursiveAction（无返回结果），我们自己定义任务时，只需选择这两个类继承即可。

RecursiveAction

// 没有返回值的 fork / join 任务框架
public class PrintTask extends RecursiveAction {
    private static final int THRESHOLD = 5;
    private int start;
    private int end;

    PrintTask(int start, int end) {
        this.start = start;
        this.end = end;
    }

    public static void main(String[] args) {
        PrintTask task = new PrintTask(0, 25);
        // 分配四个线程给它
        ForkJoinPool pool = new ForkJoinPool(4);
        pool.execute(task);
        pool.shutdown();
    }

    @Override
    protected void compute() {
        if (THRESHOLD >= (end - start)) {
            // 满足小任务条件，分配打印任务
            for (int i = start; i < end; i++) {
                System.out.println(Thread.currentThread().getName() + ": " + i);
            }
        } else {
            // 任务还能继续拆分
            int division = (start + end) >> 1;
            PrintTask task1 = new PrintTask(start, division);
            PrintTask task2 = new PrintTask(division, end);
            invokeAll(task1, task2);
        }
    }
}

RecursiveTask

// 有返回值的 fork / join 任务框架 RecursiveTask<T>
public class SumTask extends RecursiveTask<Integer> {
    private static final int THRESHOLD = 50;
    private int start;
    private int end;

    public SumTask(int start, int end) {
        this.start = start;
        this.end = end;
    }

    public static void main(String[] args) throws InterruptedException {
        final int works = 200;
        SumTask task = new SumTask(0, works);
        ForkJoinPool forkJoinPool = new ForkJoinPool(4);
        long beginTime = System.currentTimeMillis();
        int result = forkJoinPool.invoke(task);
        long consumeTime = System.currentTimeMillis() - beginTime;
        System.out.println("Fork Join calculated the result is " + result + ",consume " + consumeTime + "ms");
        forkJoinPool.shutdown();

        result = 0;
        beginTime = System.currentTimeMillis();
        for (int i = 0; i < works; i++) {
            TimeUnit.SECONDS.sleep(1);
            result += i;
        }
        consumeTime = System.currentTimeMillis() - beginTime;
        System.out.println("The correct result is " + result + ",consume " + consumeTime + "ms");
    }

    @Override
    protected Integer compute() {
        int sum = 0;
        if (THRESHOLD >= (end - start)) {
            for (int i = start; i < end; i++) {
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                sum += i;
            }
            System.out.println(Thread.currentThread().getName() + " result is ：" + sum);
            return sum;
        } else {
            int division = (start + end) >> 1;
            SumTask task1 = new SumTask(start, division);
            SumTask task2 = new SumTask(division, end);
            invokeAll(task1, task2);
            int result1 = task1.join();
            int result2 = task2.join();
            return result1 + result2;
        }
    }
}

• 结果：

  ForkJoinPool-1-worker-1 result is ：1225
  ForkJoinPool-1-worker-0 result is ：8725
  ForkJoinPool-1-worker-3 result is ：3725
  ForkJoinPool-1-worker-2 result is ：6225
  Fork Join calculated the result is 19900,consume 50018ms
  The correct result is 19900,consume 200053ms

总结

1. 我们虽然可以通过调整阈值 THRESHOLD 控制子任务的大小，从而控制了任务的数量，但是我们分配的 Fork/Join Pool 数量却是根据 CPU 性能而定的，所以，切割任务的大小和数量需要进行预先计算好，不是子任务越多就越好，而且合并结果集，需要消耗其他的计算性能。如果任务大小不能控制，可以设计可伸缩算法，动态来计算出合理的阈值，以符合要求。
2. 注意正确的 Fork/Join 框架的写法，通过廖老师的文章 Java的Fork/Join任务，你写对了吗？ ，指出了错误的写法的问题所在。

源码地址