摘要：前言在前面的文章和響應(yīng)式編程中提到了和后者毫無疑問是一個(gè)線程池前者則是一個(gè)類似經(jīng)典定義的概念官方有一個(gè)非常無語的解釋就是運(yùn)行在的一個(gè)任務(wù)抽象就是運(yùn)行的線程池框架包含和若干的子類它的核心在于分治和工作竅取最大程度利用線程池中的工作線程避免忙的

在前面的文章"CompletableFuture和響應(yīng)式編程"中提到了ForkJoinTask和ForkJoinPool,后者毫無疑問是一個(gè)線程池,前者則是一個(gè)類似FutureTask經(jīng)典定義的概念.

官方有一個(gè)非常無語的解釋:ForkJoinTask就是運(yùn)行在ForkJoinPool的一個(gè)任務(wù)抽象,ForkJoinPool就是運(yùn)行ForkJoinTask的線程池.

ForkJoin框架包含F(xiàn)orkJoinTask,ForkJoinWorkerThread,ForkJoinPool和若干ForkJoinTask的子類,它的核心在于分治和工作竅取,最大程度利用線程池中的工作線程,避免忙的忙死,餓的餓死.

ForkJoinTask可以理解為類線程但比線程輕量的實(shí)體,在ForkJoinPool中運(yùn)行的少量ForkJoinWorkerThread可以持有大量的ForkJoinTask和它的子任務(wù).ForkJoinTask同時(shí)也是一個(gè)輕量的Future,使用時(shí)應(yīng)避免較長阻塞和io.

ForkJoinTask在JAVA8中應(yīng)用廣泛,但它是一個(gè)抽象類,它的子類派生了各種用途,如后續(xù)計(jì)劃多帶帶介紹的CountedCompleter,以及若干JAVA8中stream?api定義的與并行流有關(guān)的各種操作(ops).

首先看ForkJoinTask的簽名.

public abstract class ForkJoinTask implements Future, Serializable 

從簽名上看,ForkJoinTask實(shí)現(xiàn)了future,也可以序列化,但它不是一個(gè)Runnable或Callable.

ForkJoinTask雖然可以序列化,但它只對(duì)運(yùn)行前和后敏感,對(duì)于執(zhí)行過程中不敏感.

先來看task的運(yùn)行字段:

//volatie修飾的任務(wù)狀態(tài)值,由ForkJoinPool或工作線程修改.
volatile int status; 
static final int DONE_MASK   = 0xf0000000;//用于屏蔽完成狀態(tài)位. 
static final int NORMAL      = 0xf0000000;//表示正常完成,是負(fù)值.
static final int CANCELLED   = 0xc0000000;//表示被取消,負(fù)值,且小于NORMAL
static final int EXCEPTIONAL = 0x80000000;//異常完成,負(fù)值,且小于CANCELLED
static final int SIGNAL      = 0x00010000;//用于signal,必須不小于1<<16,默認(rèn)為1<<16.
static final int SMASK       = 0x0000ffff;//后十六位的task標(biāo)簽

很顯然,DONE_MASK能夠過濾掉所有非NORMAL,非CANCELLED,非EXCEPTIONAL的狀態(tài),字段的含義也很直白,后面的SIGNAL和SMASK還不明確,后面再看.

//標(biāo)記當(dāng)前task的completion狀態(tài),同時(shí)根據(jù)情況喚醒等待該task的線程.
private int setCompletion(int completion) {
    for (int s;;) {
        //開啟一個(gè)循環(huán),如果當(dāng)前task的status已經(jīng)是各種完成(小于0),則直接返回status,這個(gè)status可能是某一次循環(huán)前被其他線程完成.
        if ((s = status) < 0)
            return s;
        //嘗試將原來的status設(shè)置為它與completion按位或的結(jié)果.
        if (U.compareAndSwapInt(this, STATUS, s, s | completion)) {
            if ((s >>> 16) != 0)
                //此處體現(xiàn)了SIGNAL的標(biāo)記作用,很明顯,只要task完成(包含取消或異常),或completion傳入的值不小于1<<16,
                //就可以起到喚醒其他線程的作用.
                synchronized (this) { notifyAll(); }
            //cas成功,返回參數(shù)中的completion.
            return completion;
        }
    }
}

前面用注釋解釋了這個(gè)方法的邏輯,顯然該方法是阻塞的,如果傳入的參數(shù)不能將status設(shè)置為負(fù)值會(huì)如何?

顯然,可能會(huì)有至多一次的成功cas,并且若滿足喚醒的條件,會(huì)嘗試去喚醒線程,甚至可能因?yàn)闉榱藛拘哑渌€程而被阻塞在synchonized代碼塊外;也可能沒有一次成功的cas,直到其他線程成功將status置為完成.

//final修飾,運(yùn)行ForkJoinTask的核心方法.
final int doExec() {
    int s; boolean completed;
    //僅未完成的任務(wù)會(huì)運(yùn)行,其他情況會(huì)忽略.
    if ((s = status) >= 0) {
        try {
            //調(diào)用exec
            completed = exec();
        } catch (Throwable rex) {
            //發(fā)生異常,用setExceptionalCompletion設(shè)置結(jié)果
            return setExceptionalCompletion(rex);
        }
        if (completed)
            //正常完成,調(diào)用前面說過的setCompletion,參數(shù)為normal,并將返回值作為結(jié)果s.
            s = setCompletion(NORMAL);
    }
    //返回s
    return s;
}

//記錄異常并且在符合條件時(shí)傳播異常行為
private int setExceptionalCompletion(Throwable ex) {
    //首先記錄異常信息到結(jié)果
    int s = recordExceptionalCompletion(ex);
    if ((s & DONE_MASK) == EXCEPTIONAL)
        //status去除非完成態(tài)標(biāo)志位(只保留前4位),等于EXCEPTIONAL.內(nèi)部傳播異常
        internalPropagateException(ex);
    return s;
}
//internalPropagateException方法是一個(gè)空方法,留給子類實(shí)現(xiàn),可用于completer之間的異常傳遞
void internalPropagateException(Throwable ex) {}
//記錄異常完成
final int recordExceptionalCompletion(Throwable ex) {
    int s;
    if ((s = status) >= 0) {
        //只能是異常態(tài)的status可以記錄.
        //hash值禁止重寫,不使用子類的hashcode函數(shù).
        int h = System.identityHashCode(this);
        final ReentrantLock lock = exceptionTableLock;
        //異常鎖,加鎖
        lock.lock();
        try {
            //抹除臟異常,后面敘述
            expungeStaleExceptions();
            //異常表數(shù)組.ExceptionNode后面敘述.
            ExceptionNode[] t = exceptionTable;//exceptionTable是一個(gè)全局的靜態(tài)常量,后面敘述
            //用hash值和數(shù)組長度進(jìn)行與運(yùn)算求一個(gè)初始的索引
            int i = h & (t.length - 1);
            for (ExceptionNode e = t[i]; ; e = e.next) {
                //找到空的索引位,就創(chuàng)建一個(gè)新的ExceptionNode,保存this,異常對(duì)象并退出循環(huán)
                if (e == null) {
                    t[i] = new ExceptionNode(this, ex, t[i]);//(1)
                    break;
                }
                if (e.get() == this) //已設(shè)置在相同的索引位置的鏈表中,退出循環(huán).//2
                    break;
            //否則e指向t[i]的next,進(jìn)入下個(gè)循環(huán),直到發(fā)現(xiàn)判斷包裝this這個(gè)ForkJoinTask的ExceptionNode已經(jīng)出現(xiàn)在t[i]這個(gè)鏈表并break(2),
            //或者直到e是null,意味著t[i]出發(fā)開始的鏈表并無包裝this的ExceptionNode,則將構(gòu)建一個(gè)新的ExceptionNode并置換t[i],
            //將原t[i]置為它的next(1).整個(gè)遍歷判斷和置換過程處在鎖中進(jìn)行.
            }
        } finally {
            lock.unlock();
        }
        //記錄成功,將當(dāng)前task設(shè)置為異常完成.
        s = setCompletion(EXCEPTIONAL);
    }
    return s;
}

//exceptionTable聲明
private static final ExceptionNode[] exceptionTable;//全局異常node表
private static final ReentrantLock exceptionTableLock;//上面用到的鎖,就是一個(gè)普通的可重入鎖.
private static final ReferenceQueue