Fork/Join框架原理

Fork/Join是一个用户并行执行任务的框架,把一个大任务分割成若干个小人物,最终汇总每一个小任务的结果后获得大任务的结果。

讲到Fork/Join必需要讲工做窃取算法。

下面讲下什么叫工做窃取算法：某个线程从其余队列里窃取任务来执行，听名字感受很奇怪，不经想什么鬼？黑人问号脸。。。。

咱们来举个例子：假如咱们须要作一个比较大的任务，能够把这个任务分割为若干互不相依的子任务，为了减小线程间的竞争，把这些子任务放到不一样的队列里，并为每一个队列建立一个单独的线程来执行队列里的任务，线程和队列对应。这时A线程负责的A队列里的任务都处理完了，而其余队列里还有任务等待处理，干完活的线程若是干等着就显得很傻，老板也不肯意是不（ps：老板就喜欢压榨咱们搬砖的），因而A就去其余线程的队列里窃取一个任务来执行。这里A和B（被窃取的线程）同时访问一个队列会出现竞争问题，那么为了不这个问题，A（窃取线程）会从双端队列的尾部拿任务，B从头部拿任务。

上面讲了理论，那么在执行的时候其实会有几个问题，好比

1.双端队列只有一个任务时会出现竞争

2.该算法会消耗更多的系统资源，好比建立多个线程和多个双端队列

好处也显而易见：充分利用线程并行计算，减小线程间的竞争

咱们来看代码若是实现

ForkJoinTask:ForkJoin任务类,它提供在任务中执行fork和join的操做机制，一般状况下咱们不须要直接继承ForkJoinTask类,只须要继承它的子类就能够了,目前提供了两个子类

RecursiveAction:用于没有返回结果的任务

RecursiveTask:用于有返回结果的任务

ForkJoinPool:ForkJoinTask须要经过ForkJoinTask来执行.

下面看一个例子实现:

首先建立一个任务类

public class CountTask  extends RecursiveTask<Integer> {
    private static final int THRESHOLD=2;//阈值
    private  int start;
    private  int end;
    public  CountTask(int start,int end){
        this.start=start;
        this.end=end;
    }

    @Override
    protected Integer compute() {
        int sum=0;
        //若是任务足够小就计算任务
        boolean canCompute=(end-start)<=THRESHOLD;
        if(canCompute){
            for (int i=start;i<=end;i++){
                sum+=i;
                try {
                    TimeUnit.SECONDS.sleep(1);
                }catch (Exception e){

                }
            }
        }else {
            int middle=(start+end)/2;
            CountTask leftTest=new CountTask(start,middle);
            CountTask rightTest=new CountTask(middle+1,end);
            //执行子任务
            leftTest.fork();
            rightTest.fork();
            //等待子任务执行完,并获得其结果
            int leftResult=leftTest.join();
            int rightResult=rightTest.join();
            //合并子任务
            sum=leftResult+rightResult;
        }
        return  sum;
    }
}

　　调用类:

@RestController
public class ForkJoinTest {

    private int start = 1;
    private int end = 4;

    @GetMapping("ForkJoinTest")
    public void forkjointest() throws InterruptedException, ExecutionException {
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        //生成一个计算任务,负责计算1+2+3+4
        CountTask test = new CountTask(start, end);
        //执行任务
        Future<Integer> result = forkJoinPool.submit(test);
        System.out.println(result.get());
    }

    @GetMapping("test")
    public void test() throws InterruptedException {
        int sum = 0;
        for (int i = start; i <=end; i++) {
            sum += i;
            TimeUnit.SECONDS.sleep(1);
        }
        System.out.println(sum);
    }
}

 

这里写了两个方法,一个是经过ForkJoinTask来执行的,一个没有,咱们执行看下耗时

方法:forkjointest,参数:[Ljava.lang.Object;@6f277213
10
方法:forkjointest--cost:2009mills
方法:test,参数:[Ljava.lang.Object;@a52eb2f
10
方法:test--cost:4001mills