面试官: Fork/Join 有了解过吗？说说看（含源码分析）

public class ForkJoinTest {
    // 首先继承 RecursiveTask 并重写compute方法
    public static class TaskDemo extends RecursiveTask<Integer> {
        private Integer i = 0;
        private Integer num;
        public TaskDemo(Integer num) {
            this.num = num;
        }
        @Override
        protected Integer compute() {
            // 判断任务是否拆分到合适数量
            if(num > 20) {
                // 任务拆分完成后进行计算
                for(int index=0; index < 200; index++) {
                    i ++;
                }
                return i;
            }else {
                // 拆分成两个任务
                TaskDemo t1 = new TaskDemo(num + 1);
                t1.fork();
                TaskDemo t2 = new TaskDemo(num + 2);
                t2.fork();
                return t1.join() + t2.join();
            }
        }
    }
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Long start = System.currentTimeMillis();
        // 创建线程池
        ForkJoinPool pool = new ForkJoinPool();
        // 捕获返回的最终结果
        ForkJoinTask<Integer> taskFuture = pool.submit(new TaskDemo(0));
        Integer result = taskFuture.get();
        System.out.println("最终结果 >>>> " + result);
        Long end = System.currentTimeMillis();
        System.out.println("耗时 >>> " + (end -start) + "ms");
    }
}
复制代码

最终结果 >>>> 5731400
耗时 >>> 170ms
复制代码

public final ForkJoinTask<V> fork() {
    Thread t;
    // ForkJoinWorkerThread是执行ForkJoinTask的专有线程，由ForkJoinPool管理
    // 先判断当前线程是否是ForkJoin专有线程，如果是，则将任务push到当前线程所负责的队列里去
    if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
        ((ForkJoinWorkerThread)t).workQueue.push(this);
    else
      // 如果不是则将线程加入队列
        // 没有显式创建ForkJoinPool的时候走这里，提交任务到默认的common线程池中,之前的例子是手动创建了一个
        ForkJoinPool.common.externalPush(this);
    return this;
}
复制代码

public final V join() {
    int s;
    // doJoin()来获取当前任务的执行状态
    if ((s = doJoin() & DONE_MASK) != NORMAL)
        reportException(s);
    // 最后返回结果
    return getRawResult();
}
复制代码

private int doJoin() {
    int s; Thread t; ForkJoinWorkerThread wt; ForkJoinPool.WorkQueue w;
    // 先判断任务是否执行完毕，执行完毕直接返回结果（执行状态）
    return (s = status) < 0 ? s :
    // 如果没有执行完毕，先判断是否是ForkJoinWorkThread线程
        ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
        // 如果是，先判断任务是否处于工作队列顶端（意味着下一个就执行它）
        // tryUnpush()方法判断任务是否处于当前工作队列顶端，是返回true
        // doExec()方法执行任务
        (w = (wt = (ForkJoinWorkerThread)t).workQueue).
        // 如果是处于顶端并且任务执行完毕，返回结果
        tryUnpush(this) && (s = doExec()) < 0 ? s :
        // 如果不在顶端或者在顶端却没执行完毕，那就调用awitJoin()执行任务
        // awaitJoin()：使用自旋使任务执行完成，返回结果
        wt.pool.awaitJoin(w, this, 0L) :
        // 如果不是ForkJoinWorkThread线程，执行externalAwaitDone()返回任务结果
        externalAwaitDone();
}
复制代码