并发编程之：ForkJoin

当前位置:

首页 > 编程开发 > Java教程 >

并发编程之：ForkJoin

大家好，我是小黑，一个在互联网苟且偷生的农民工。

在JDK1.7中引入了一种新的Fork/Join线程池，它可以将一个大的任务拆分成多个小的任务并行执行并汇总执行结果。

Fork/Join采用的是分而治之的基本思想，分而治之就是将一个复杂的任务，按照规定的阈值划分成多个简单的小任务，然后将这些小任务的结果再进行汇总返回，得到最终的任务。

分治法

分治法是计算机领域常用的算法中的其中一个，主要思想就是将将一个规模为N的问题，分解成K个规模较小的子问题，这些子问题相互独立且与原问题性质相同；求解出子问题的解，合并得到原问题的解。

解决问题的思路

分割原问题；
求解子问题；
合并子问题的解为原问题的解。

使用场景

二分查找，阶乘计算，归并排序，堆排序、快速排序、傅里叶变换都用了分治法的思想。

ForkJoin并行处理框架

在JDK1.7中推出的ForkJoinPool线程池，主要用于ForkJoinTask任务的执行，ForkJoinTask是一个类似线程的实体，但是比普通线程更轻量。

我们来使用ForkJoin框架完成以下1-10亿求和的代码。

public class ForkJoinMain {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        ForkJoinTask<Long> rootTask = forkJoinPool.submit(new SumForkJoinTask(1L, 10_0000_0000L));
        System.out.println("计算结果：" + rootTask.get());
    }
}

class SumForkJoinTask extends RecursiveTask<Long> {
    private final Long min;
    private final Long max;
    private Long threshold = 1000L;

    public SumForkJoinTask(Long min, Long max) {
        this.min = min;
        this.max = max;
    }
    @Override
    protected Long compute() {
        // 小于阈值时直接计算
        if ((max - min) <= threshold) {
            long sum = 0;
            for (long i = min; i < max; i++) {
                sum = sum + i;
            }
            return sum;
        }
        // 拆分成小任务
        long middle = (max + min) >>> 1;
        SumForkJoinTask leftTask = new SumForkJoinTask(min, middle);
        leftTask.fork();
        SumForkJoinTask rightTask = new SumForkJoinTask(middle, max);
        rightTask.fork();
        // 汇总结果
        return leftTask.join() + rightTask.join();
    }
}

上述代码逻辑可通过下图更加直观的理解。

ForkJoin框架实现

在ForkJoin框架中重要的一些接口和类如下图所示。

ForkJoinPool

ForkJoinPool是用于运行ForkJoinTasks的线程池，实现了Executor接口。

可以通过new ForkJoinPool()直接创建ForkJoinPool对象。

public ForkJoinPool() {
    this(Math.min(MAX_CAP, Runtime.getRuntime().availableProcessors()),
         defaultForkJoinWorkerThreadFactory, null, false);
}

public ForkJoinPool(int parallelism,
                        ForkJoinWorkerThreadFactory factory,
                        UncaughtExceptionHandler handler,
                        boolean asyncMode){
    this(checkParallelism(parallelism),
             checkFactory(factory),
             handler,
             asyncMode ? FIFO_QUEUE : LIFO_QUEUE,
             "ForkJoinPool-" + nextPoolId() + "-worker-");
        checkPermission();
}