Java中J.U.C扩展组件之ForkJoinTask和ForkJoinPool

public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
    /** ForkJoinTask运行状态 */
    volatile int status; // 直接被ForkJoin池和工作线程访问
    static final int DONE_MASK   = 0xf0000000;  // mask out non-completion bits
    static final int NORMAL      = 0xf0000000;  // must be negative
    static final int CANCELLED   = 0xc0000000;  // must be < NORMAL
    static final int EXCEPTIONAL = 0x80000000;  // must be < CANCELLED
    static final int SIGNAL      = 0x00010000;  // must be >= 1 << 16
    static final int SMASK       = 0x0000ffff;  // short bits for tags
    
    /**
     * @Ruturn 任务是否扔出异常或被取消
     */
    public final boolean isCompletedAbnormally() {
        return status < NORMAL;
    }
    
    /**
     * 如果计算扔出异常，则返回异常
     * 如果任务被取消了则返回CancellationException。如果任务没有完成或者没有抛出异常则返回null
     */
    public final Throwable getException() {
        int s = status & DONE_MASK;
        return ((s >= NORMAL)    ? null :
                (s == CANCELLED) ? new CancellationException() :
                getThrowableException());
    }
}

public class ForkJoinPool extends AbstractExecutorService {
    /**
     * ForkJoinPool，它同ThreadPoolExecutor一样，也实现了Executor和ExecutorService接口。它使用了
     * 一个无限队列来保存需要执行的任务，而线程的数量则是通过构造函数传入，如果没有向构造函数中传入希
     * 望的线程数量，那么当前计算机可用的CPU数量会被设置为线程数量作为默认值。
     */
    public ForkJoinPool() {
        this(Math.min(MAX_CAP,Runtime.getRuntime().availableProcessors()),
             defaultForkJoinWorkerThreadFactory, null, false);
    }
    public ForkJoinPool(int parallelism) {
        this(parallelism, defaultForkJoinWorkerThreadFactory, null, false);
    }
    //有多个构造器，这里省略
    
    volatile WorkQueue[] workQueues;     // main registry
    static final class WorkQueue {
        final ForkJoinWorkerThread owner; // 工作线程
        ForkJoinTask<?>[] array;   // 任务
        
        //传入的是ForkJoinPool与指定的一个工作线程
        WorkQueue(ForkJoinPool pool, ForkJoinWorkerThread owner) {
            this.pool = pool;
            this.owner = owner;
            // Place indices in the center of array (that is not yet allocated)
            base = top = INITIAL_QUEUE_CAPACITY >>> 1;
        }

    }
}

public class ForkJoinPool extends AbstractExecutorService {
    public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {}
    public <T> ForkJoinTask<T> submit(Callable<T> task) {}
    public <T> ForkJoinTask<T> submit(Runnable task, T result) {}
    public ForkJoinTask<?> submit(Runnable task) {}
}

public abstract class ForkJoinTask<V> implements Future<V>, Serializable {
    /**
     * 在当前任务正在运行的池中异步执行此任务（如果适用）
     * 或使用ForkJoinPool.commonPool()（如果不是ForkJoinWorkerThread实例）进行异步执行 
     */
    public final ForkJoinTask<V> fork() {
        Thread t;
        if ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread)
            ((ForkJoinWorkerThread)t).workQueue.push(this);
        else
            ForkJoinPool.common.externalPush(this);
        return this;
    }
    
    public final V join() {
        int s;
        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 :
            ((t = Thread.currentThread()) instanceof ForkJoinWorkerThread) ?
            (w = (wt = (ForkJoinWorkerThread)t).workQueue).
            tryUnpush(this) && (s = doExec()) < 0 ? s :
            wt.pool.awaitJoin(w, this, 0L) :
            externalAwaitDone();
    }
}

public static void main(String[] args) throws ExecutionException, InterruptedException {
        //生成一个池
        ForkJoinPool forkJoinPool=new ForkJoinPool();
        ForkJoinTask task=new ForkJoinExample(1, 100000);
        ForkJoinTask<Integer> submit = forkJoinPool.submit(task);
        Integer sum = submit.get();
        System.out.println("最后的结果是:"+sum);

}