ForkJoin使用

一、Fork Join　　　　
分而治之的办法

JDk为Fork/Join框架提供了很好的支持，我们想要用这个算法首先得创建一个Fork/Join任务，在JDK中这个任务就叫做：ForJoinTask，只要继承这个类就可以创建一个任务类，但是实际使用中并不是直接继承ForkJoinTask类，而是继承它的子类，它有两个子类，分别是RecursiveAction和RecursiveTask，它们之间的区别是是否返回任务结果，前者用于没有返回结果的任务，后者用于有返回结果的任务。

ForkJoinPool
RecursiveTask
RecursiveAction

二、RecursiveTask使用

ForkJoinPoolTest：

package com.cy.java8;

import java.util.concurrent.ForkJoinPool;
import com.cy.java8.AccumulatorRecursiveAction.AccumulatorHelper;

public class ForkJoinPoolTest {

    private static int[] data = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10};

    public static void main(String[] args) {
        System.out.println("result = " + calc());

        AccumulatorRecursiveTask task = new AccumulatorRecursiveTask(0, data.length - 1, data);
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        Integer result = forkJoinPool.invoke(task);     //同步调用
        System.out.println("AccumulatorRecursiveTask result = " + result);

        AccumulatorRecursiveAction action = new AccumulatorRecursiveAction(0, data.length - 1, data);
        forkJoinPool.invoke(action);
        System.out.println("AccumulatorRecursiveAction result = " + AccumulatorHelper.getResult());
        AccumulatorHelper.reset();
    }

    /**
     * 普通方法计算data的和
     */
    private static int calc() {
        int result = 0;
        for (int i = 0; i < data.length; i++) {
            result += data[i];
        }
        return result;
    }
}

AccumulatorRecursiveTask

package com.cy.java8;

import java.util.concurrent.RecursiveTask;

/**
 * 继承RecursiveTask，用于与结果返回的任务
 */
public class AccumulatorRecursiveTask extends RecursiveTask<Integer> {

    //累加的起始下标值
    private final int start;
    private final int end;
    private final int[] data;

    //阈值为3，当小于等于这个值的时候进行计算
    private final int LIMIT = 3;

    public AccumulatorRecursiveTask(int start, int end, int[] data) {
        this.start = start;
        this.end = end;
        this.data = data;
    }

    @Override
    protected Integer compute() {
        if (end - start < LIMIT) {
            int result = 0;
            for (int i = start; i <= end; i++) {
                result += data[i];
            }
            return result;
        }

        int mid = (start + end) / 2;

        //分为两个任务
        AccumulatorRecursiveTask left = new AccumulatorRecursiveTask(start, mid, data);
        AccumulatorRecursiveTask right = new AccumulatorRecursiveTask(mid + 1, end, data);
        /*
        //执行子任务
        left.fork();
        right.fork();
        //等待子任务完成
        Integer leftResult = left.join();
        Integer rightResult = right.join();
        //合并子任务
        return rightResult + leftResult;
        */

        //提交任务
        invokeAll(left, right);
        return left.join() + right.join();
    }
}

三、RecursiveAction使用　　

package com.cy.java8;

import java.util.concurrent.RecursiveAction;
import java.util.concurrent.atomic.AtomicInteger;

public class AccumulatorRecursiveAction extends RecursiveAction {
    private int start;
    private int end;
    private int[] data;
    private final int LIMIT = 3;

    public AccumulatorRecursiveAction(int start, int end, int[] data) {
        this.start = start;
        this.end = end;
        this.data = data;
    }

    @Override
    protected void compute() {
        if (end - start < LIMIT) {
            for (int i = start; i <= end; i++) {
                AccumulatorHelper.accumulate(data[i]);
            }
        } else {
            int mid = (start + end) / 2;
            AccumulatorRecursiveAction left = new AccumulatorRecursiveAction(start, mid, data);
            AccumulatorRecursiveAction right = new AccumulatorRecursiveAction(mid + 1, end, data);
            /*left.fork();
            right.fork();
            left.join();
            right.join();*/
            invokeAll(left, right);
        }
    }

    static class AccumulatorHelper{
        private static final AtomicInteger result = new AtomicInteger(0);

        private static void accumulate(int value){
            result.getAndAdd(value);
        }

        public static int getResult(){
            return result.get();
        }

        public static void reset(){
            result.set(0);
        }
    }
}

ForkJoinPoolTest的执行结果：

result = 55
AccumulatorRecursiveTask result = 55
AccumulatorRecursiveAction result = 55

 注：

1、RecursiveAction 提供的方法 invokeAll()。它表示：启动所有的任务，并在所有任务都正常结束后返回。如果其中一个任务出现异常，则其它所有的任务都取消。invokeAll() 的参数还可以是任务的数组。

-----------