《Java并发编程实践》笔记8——Fork/Join框架

《java并发编程实践》书中并没有介绍Fork/Join框架,该框架和并发编程关系密切,也是由本书作者之一Doug Lea在JDK1.7引入,因此作为最后一篇笔记简单介绍。
Fork/Join框架是JDK1.7提供的一个用于并行执行任务的框架,它把一个大任务分割为若干个小任务执行,最后将小任务的执行结果汇总得到大任务的结果。当看到这个定义时,我想很多人会和我一样立刻想到现在非常热门的Map-Reduce思想。二者的对比如下:
共同点:
都是用于执行并行任务的。基本思想都是把复杂问题分解为一个个简单的子问题分别计算,再合并结果。应该说并行计算都是这种思想,彼此独立的或可分解的。从名字上看Fork和Map都有切分的意思,Join和Reduce都有合并的意思,比较类似。

区别:


(1).环境差异:
分布式 vs 单机多核:Fork/Join设计初衷针对单机多核(处理器数量很多的情况)。Map-Reduce一开始就明确是针对很多机器组成的集群环境的。也就是说一个是想充分利用多处理器,而另一个是想充分利用很多机器做分布式计算。这是两种不同的的应用场景,有很多差异,因此在细的编程模式方面有很多不同。

(2).编程差异:
Map-Reduce一般是:做较大粒度的切分,一开始就先切分好任务然后再执行,并且彼此间在最后合并之前不需要通信。这样可伸缩性更好,适合解决巨大的问题,但限制也更多。Fork/Join可以是较小粒度的切分,任务自己知道该如何切分自己,递归地切分到一组合适大小的子任务来执行,因为是同一个JVM内,所以彼此间通信是很容易的,更像是传统编程方式。
综上所述,我们可以认为Fork/Join是同一个JVM内的Map-Reduce的一种实现方式。
Fork/Join的基本工作流程如下图:
《Java并发编程实践》笔记8——Fork/Join框架
Fork-Join框架的实现原理为:
首先,分割任务,只要任务的粒度超过阀值,就不停地将任务分拆为小任务;
然后,将分割的任务任务添加到双端队列中,启动线程从双端队列获取任务执行,将执行结果统一放到一个队列中;
最后,再启动一个线程合并结果队列的值。
Fork-Join框架涉及的主要类如下:
《Java并发编程实践》笔记8——Fork/Join框架
RecursiveAction:用于没有返回值的任务。
RecursiveTask:用于需要返回值的任务,通过泛型参数设置计算的返回值类型。
ForkJoinPool:提供了一系列的submit方法,计算ForkJoinTask(需要实现computer方法)。
我们以计算1+2+3+4结果为例,来演示Fork/Join的使用,我们设置阀值为2,则由于是4个数字相加,因此Fork/Join框架会分拆为两个子任务,一个计算1+2,另一个计算3+4,最后在把两个子任务的结果合并,例子代码如下:

public class CounterTask extends RecursiveTask<Integer>{
    private static final int THRESHOLD = 2;
    private int start;
    private int end;

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

    protected Integer computer(){
        int sum = 0;
        boolean canComputer = (end - start) <= THRESHOLD;
        if(canComputer){
            for(int i = start; i <= end; i++){
                sum += i;
            }
        }else{
            int middle = (start + end) / 2;
            CounterTask leftTask = new CounterTask(start, middle);
            CounterTask rightTask = new CounterTask(middle + 1, end);
            leftTask.fork();
            rightTask.fork();
            int leftResult = leftTask.join();
            int rightResult = rightTask.join();
            sum = leftResult + rightResult;
        }
        return sum;
    }

    public static void main(String[] args){
        ForkJoinPool pool = new ForkJoinPool();
        CounterTask task = new CounterTask(1, 4);
        Future<Integer> result = pool.submit(task);
        try{
            System.out.println("result is:" + result.get());
        }catch(InterruptedException e){
        }catch(ExecutionException e){
        }finally{
            pool.shutdown();
        }
    }
}

相关推荐