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

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

官方有一個非常無語的解釋:ForkJoinTask就是運行在ForkJoinPool的一個任務抽象,ForkJoinPool就是運行ForkJoinTask的線程池.

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

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

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

首先看ForkJoinTask的簽名.

public abstract class ForkJoinTask implements Future, Serializable 

從簽名上看,ForkJoinTask實現了future,也可以序列化,但它不是一個Runnable或Callable.

ForkJoinTask雖然可以序列化,但它只對運行前和后敏感,對于執行過程中不敏感.

先來看task的運行字段:

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

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

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

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

顯然,可能會有至多一次的成功cas,并且若滿足喚醒的條件,會嘗試去喚醒線程,甚至可能因為為了喚醒其他線程而被阻塞在synchonized代碼塊外;也可能沒有一次成功的cas,直到其他線程成功將status置為完成.

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

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

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