JUC系列（十） | Fork&Join框架 并行处理任务

fork()   // 在当前线程运行的线程池中安排一个异步执行。简单的理解就是再创建一个子任务。
join()    //当任务完成的时候返回计算结果。
invoke()    //开始执行任务，如果必要，等待计算完成。

public final ForkJoinTask<V> fork() {
    Thread t;
    if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
        ((ForkJoinWorkerThread)t).workQueue.push(this);
    else
        //将给定的任务添加到提交者当前队列的提交队列中，如果为 null 或存在竞争，则创建一个。
        ForkJoinPool.common.externalPush(this);
    return this;
}

//推送任务。 仅由非共享队列中的所有者调用。
final void push(ForkJoinTask<?> task) {
    ForkJoinTask<?>[] a;
    int s = top, d, cap, m;
    ForkJoinPool p = pool;
    if ((a = array) != null && (cap = a.length) > 0) {
        QA.setRelease(a, (m = cap - 1) & s, task);
        top = s + 1;
        if (((d = s - (int)BASE.getAcquire(this)) & ~1) == 0 &&
            p != null) {                 // size 0 or 1
            VarHandle.fullFence();
            // signalWork方法的意义在于，如果运行的工作程序太少，则尝试创建或释放工作程序。
            p.signalWork(); 
        }

        // 如果array的剩余空间不够了，则进行增加
        else if (d == m)
            growArray(false); 
    }
}

public final V join() {
    int s;
    if ((s = doJoin() & DONE_MASK) != NORMAL)
        //假如任务状态是抛出异常状态，就直接抛出对应的异常
        //若是任务状态是被取消状态，则直接抛出CancellationException异常。
        reportException(s);
    //如若任务状态是已经完成，则直接立马返回任务结果。
    //即使此任务异常完成，如果不知道此任务已完成，则返回null 
    return getRawResult();
}

private int doJoin() {
    int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w;
    //1. 首先通过查看任务的状态，看任务是否已经执行完成，如果执行完成，则直接返回任务状态；
	//2. 如果没有执行完任务，则从任务数组里取出任务并执行。
    //3. 如果任务顺利执行完成，则设置任务状态为 NORMAL，假如出现异常，则记录异常，并将任务状态设置为 EXCEPTIONAL。
    return (s = status) < 0 ?
        s : ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
        (w = (wt = (ForkJoinWorkerThread)t).workQueue).
        tryUnpush(this) && (s = doExec()) < 0 ? s :
    //帮助和/或阻塞，直到给定的任务完成或超时
    wt.pool.awaitJoin(w, this, 0L) :
    externalAwaitDone();
}

//如果此任务引发异常或被取消，则返回true 。 通常用来判断任务情况
public final boolean isCompletedAbnormally() {
    return (status & ABNORMAL) != 0;
}

//getException 方法返回 Throwable 对象，如果任务被取消了则返回CancellationException,如果任务没有完成或者没有抛出异常则返回 null。
public final Throwable getException() {
    int s = status;
    return ((s & ABNORMAL) == 0 ? null :
            (s & THROWN)   == 0 ? new CancellationException() :
            getThrowableException());
}

import java.util.concurrent.RecursiveTask;
/**
* 递归累加
*/
public class TaskExample extends RecursiveTask<Long> {
    private int start;
    private int end;
    private long sum;
    /**
	* 构造函数
	* @param start
	* @param end
	*/
    public TaskExample(int start, int end){
        this.start = start;
        this.end = end;
    }

   
    @Override
    protected Long compute() {
        System.out.println("任务" + start + "=========" + end + "累加开始");
        //大于 100 个数相加切分,小于直接加
        if(end - start <= 100){
            for (int i = start; i <= end; i++) {
                //累加
                sum += i;
            }
        }else {
            //切分为 2 块
            int middle = start + 100;
            //递归调用,切分为 2 个小任务
            TaskExample taskExample1 = new TaskExample(start, middle);
            TaskExample taskExample2 = new TaskExample(middle + 1, end);
            //执行:异步
            taskExample1.fork();
            taskExample2.fork();
            //同步阻塞获取执行结果
            sum = taskExample1.join() + taskExample2.join();
        }
        //加完返回
        return sum;
    }
}

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
/**
* 分支合并案例
*/
public class ForkJoinPoolDemo {
    /**
 * 生成一个计算任务，计算 1+2+3.........+1000
 * @param args
 */
    public static void main(String[] args) {
        //定义任务
        TaskExample taskExample = new TaskExample(1, 1000);
        //定义执行对象
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        //加入任务执行
        ForkJoinTask<Long> result = forkJoinPool.submit(taskExample);
        //输出结果
        try {
            System.out.println(result.get());
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            forkJoinPool.shutdown();
        }
    }
}