Future原始碼一觀-JUC系列-有解無憂

背景介紹

在程式中，主執行緒啟動一個子執行緒進行異步計算，主執行緒是不阻塞繼續執行的，這點看起來是非常自然的，都已經選擇啟動子執行緒去異步執行了，主執行緒如果是阻塞的話，那還不如主執行緒自己去執行不就好了，那會不會有一種場景，異步執行緒執行的結果主執行緒是需要使用的，或者說主執行緒先做一些作業，然后需要確認子執行緒執行情況來進行后續的操作，那么這里就需要子執行緒異步執行完任務能把結果告訴主執行緒，并且主執行緒還能訪問到子執行緒執行任務的狀態，比如是否執行完成或正在執行中，

Future就是上面概念的抽象，按照原始碼中的注釋，它代表著一個異步計算的結果，提供的方法中可以通過get方法獲取異步執行緒計算的結果，如果計算還沒結束，就會阻塞等待回傳成功；也可以通過cancel方法取消異步計算任務；還可以通過isCancelled和isDone獲得異步執行的狀態；如果一個異步執行的內容并沒有回傳值，但是希望可以使用Future來獲得取消異步計算任務的能力，可以回傳null，

FutureTask

FutureTask提供了對Future的基礎實作，在進入FutureTask原始碼之前，我們先考慮下如果要實作Future的功能可以怎么設計呢？

1，異步執行緒進入執行任務的時候，主執行緒先阻塞住，等到一步執行緒任務完成有回傳結果了，喚醒主執行緒，把回傳結果給它，

2，需要有個標記，記錄異步執行緒有沒有執行結束，異步執行緒任務執行一結束，這個標記就要變更，通過這個標記就可以知道執行狀態，

3，能獲取異步執行緒，在執行還沒完成先，對異步執行緒可以中斷，這樣就可以取消異步執行緒執行的任務了，

4，異步執行緒執行和取消操作是有并發競爭的，所以應該對標記的更新做鎖保護處理，

對照Future的API，大致能想到這些，實際還有大量關鍵細節組合才能實作，可以帶這個實作思路進入原始碼的學習，

Task

FutureTask本身就是繼承Runnable，Runnable的run方法是沒有回傳引數的，那么既然FutureTask需要把異步執行緒執行結果回傳，就意味著需要把結果拿到記錄，

建構式

public FutureTask(Callable<V> callable) {
    if (callable == null)
        throw new NullPointerException();
    this.callable = callable;
    this.state = NEW;       // ensure visibility of callable
}
public FutureTask(Runnable runnable, V result) {
    this.callable = Executors.callable(runnable, result);
    this.state = NEW;       // ensure visibility of callable
}

當建構式傳的是Runnable的時候，會適配成Callable，所以對于自己的run方法需要回傳結果的事那么就好辦了，就是呼叫callable的run方法就行，我們再衍生開去看下Executors.callable(runnable, result);的實作，

public static <T> Callable<T> callable(Runnable task, T result) {
    if (task == null)
        throw new NullPointerException();
    return new RunnableAdapter<T>(task, result);
}	
static final class RunnableAdapter<T> implements Callable<T> {
    final Runnable task;
    final T result;
    RunnableAdapter(Runnable task, T result) {
        this.task = task;
        this.result = result;
    }
    public T call() {
        task.run();
        return result;
    }
}

這個適配沒什么特殊，把一個result參考作為引數傳入，然后作為結果回傳，所以其實很少用這種方式來獲取result，更多的是傳一個null進來，因為更多的時候是想知道異步執行緒是否執行結束了，而不是要結果，

run方法

FutureTask既然本身就是Runnable，把它作為task提交給執行緒池執行，就是呼叫它的run方法，根據前面的分析，這個run方法需要呼叫內部屬性callable的run獲得result，然后保存result，以備get方法來獲取的時候能直接回傳，另外肯定也是要處理例外的場景，

以下是run方法的原始碼，再加上仔細關注一下狀態的流轉，就可以比較好的理解這個核心代碼了，

public void run() {
  	// 【1】
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                     null, Thread.currentThread()))
        return;
    try {
        Callable<V> c = callable;
        if (c != null && state == NEW) {
            V result;
            boolean ran;
            try {
                result = c.call();
                ran = true;
            } catch (Throwable ex) {
                result = null;
                ran = false;
              	// 【2】
                setException(ex);
            }
            if (ran)
              	// 【3】
                set(result);
        }
    } finally {
      	//【4】
        // runner must be non-null until state is settled to
        // prevent concurrent calls to run()
        runner = null;
        // state must be re-read after nulling runner to prevent
        // leaked interrupts
        int s = state;
        if (s >= INTERRUPTING)
          	//【5】
            handlePossibleCancellationInterrupt(s);
    }
}

protected void set(V v) {
  if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
    outcome = v;
    UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
    finishCompletion();
  }
}
protected void setException(Throwable t) {
  if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
    outcome = t;
    UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
    finishCompletion();
  }
}
private void handlePossibleCancellationInterrupt(int s) {
  if (s == INTERRUPTING)
    while (state == INTERRUPTING)
      Thread.yield(); // wait out pending interrupt
}

【1】執行的起始狀態必須是NEW，初始化FutureTask的時候設定的NEW狀態，如果不是NEW狀態，就退出run方法；并且CAS設定runner欄位為當前執行執行緒，設定失敗表示已經設定過，就退出run方法，根據狀態和CAS設定runner欄位判斷，確保了FutureTask實體同時只能有一個一個執行緒在執行，
【2】執行callable的run方法例外，進行setException操作，先把狀態從NEW設定成COMPLETING，設定成功后把outcome欄位設定成例外結果，然后將狀態設定成EXCEPTIONAL，finishCompletion方法在狀態進入終態（final state）的時候都會被呼叫，他會喚醒等待的執行緒節點，是流程中的關鍵一環，在后續中詳細分析，
【3】正常執行callable的run方法會獲得結果，進行set操作，老規矩，先把狀態從NEW設定成COMPLIETING，設定成功后把outcome欄位設定成回傳結果result，以備等待執行緒來獲取，然后把狀態設定成NORMAL，NORMAL作為終態，也會呼叫finishCompletion方法，
【4】finally代碼塊，前面有通過判斷runner是否為空來避免并發執行，所以最后把runner設定成null，這個注釋好理解，在狀態確定之前，Runner必須是非空的，以防止對run()的并發呼叫，這一點結合【1】就可以解釋，第二步的注釋說，狀態重新讀取必須在將runner設定為null之后，以防止泄漏中斷，這里需要結合cancel方法分析，cancel方法中執行的順序是先將state修改成INTERRUPTING成功后再使用runner，這里就保證了先設定runner為null后再獲取state的最新值，
【5】handlePossibleCancellationInterrupt方法中用一個while回圈加Thread.yield()來等待state在INTERRUPTING下變成INTERRUPTED，也就是說當cancel方法把state改成INTERRUPTING后，run方法就會等待cancel方法執行結束后自己才執行結束，

直到網上找到了這篇文章why outcome object in FutureTask is non-volatile?

這里有個很巧妙的設計，就是利用java的happends before中的傳遞原則，使得在不使用鎖的情況下，保證其他執行緒讀到state=NORMAL時，該執行緒一定能讀到outcome的最新值

Task State

前面提到需要一個標記來記錄任務的執行狀態，原始碼實作中有一個volatile修飾的int型別state欄位（和AQS一樣的配方的感覺來了），

    /**
     * NEW -> COMPLETING -> NORMAL
     * NEW -> COMPLETING -> EXCEPTIONAL
     * NEW -> CANCELLED
     * NEW -> INTERRUPTING -> INTERRUPTED
    **/
		private volatile int state;
    private static final int NEW          = 0;
    private static final int COMPLETING   = 1;
    private static final int NORMAL       = 2;
    private static final int EXCEPTIONAL  = 3;
    private static final int CANCELLED    = 4;
    private static final int INTERRUPTING = 5;
    private static final int INTERRUPTED  = 6;

注釋提供了全部狀態流轉路徑，核心邏輯就是一步步變更狀態來進行的，

Treiber Stack

我們需要了解清楚這個Treiber Stack的概念，因為這在JUC原始碼很多地方有使用，有助于我們理解JUC其他組件代碼的實作，

對于一個堆疊，我們并發往堆疊里放節點的時候如何處理競爭呢？比較簡單的方式就是使用鎖，放的時候鎖，取的時候鎖，

有個大佬他不想用鎖，而是利用CAS并發原語設計了一個無鎖堆疊，并發表了論文，他名字就叫Treiber，這個就是Treiber Stack的由來，在FutureTask的原始碼注釋中專門提到，很多JDK原始碼中都用到了類似這種參考，表達這個實作是有堅實理論依據的，有一種做學術的專業氛圍，

直接看《Java Concurrency in Practice》中提供的實作代碼：

public class TreiberStack <E> {
    AtomicReference<Node<E>> top = new AtomicReference<Node<E>>();

    public void push(E item) {
        Node<E> newHead = new Node<E>(item);
        Node<E> oldHead;
        do {
            oldHead = top.get();
            newHead.next = oldHead;
        } while (!top.compareAndSet(oldHead, newHead));
    }

    public E pop() {
        Node<E> oldHead;
        Node<E> newHead;
        do {
            oldHead = top.get();
            if (oldHead == null)
                return null;
            newHead = oldHead.next;
        } while (!top.compareAndSet(oldHead, newHead));
        return oldHead.item;
    }

    private static class Node <E> {
        public final E item;
        public Node<E> next;

        public Node(E item) {
            this.item = item;
        }
    }
}

這個佇列在入隊和出隊的時候都沒有進行鎖操作，而是CAS設定頭節點是否成功，如果設定成功表示頭節點沒有被修改過，就沒有競爭發生，直接設定頭節點，如果CAS設定失敗表示有競爭發生，則欄位繼續，知道設定頭節點成功，

其實只要記住一點，操作這個資料結構的入口集中在頭節點上，原子操作頭節點保證不會發生并發引起的讀寫資料例外的問題，

下面看一下FutureTask是如何定義這個鏈表節點的：

WaitNode

使用WaitNode來表示鏈表節點，內部有記錄阻塞等待的執行緒和下一個節點的參考，

static final class WaitNode {
    volatile Thread thread;
    volatile WaitNode next;
    WaitNode() { thread = Thread.currentThread(); }
}

以下是FutureTask中實作的Treiber Stack結構圖：

get方法

前面已經提過，get方法是阻塞執行緒等待的，怎么阻塞的？多個執行緒都呼叫get方法阻塞的時候如何維護這些執行緒？帶著這兩問題繼續閱讀原始碼，

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    if (s <= COMPLETING)
        s = awaitDone(false, 0L);
    return report(s);
}

public V get(long timeout, TimeUnit unit)
    throws InterruptedException, ExecutionException, TimeoutException {
    if (unit == null)
        throw new NullPointerException();
    int s = state;
    if (s <= COMPLETING &&
        (s = awaitDone(true, unit.toNanos(timeout))) <= COMPLETING)
        throw new TimeoutException();
    return report(s);
}

【1】狀態不是終態情況下呼叫awaitDone方法，是終態時呼叫report方法，對于有超時時間需求的情況，在到達超時時間時awaitDone方法就會回傳state結果，如果還不是終態就拋出TimeoutException，

awaitDone

這個方法里實作了如果異步執行緒還未執行結束的時候，來呼叫get方法阻塞等待的能力，

private int awaitDone(boolean timed, long nanos)
    throws InterruptedException {
    final long deadline = timed ? System.nanoTime() + nanos : 0L;
    WaitNode q = null;
    boolean queued = false;
    for (;;) {
        if (Thread.interrupted()) {
            removeWaiter(q);
            throw new InterruptedException();
        }

        int s = state;
      	// 【1】
        if (s > COMPLETING) {
            if (q != null)
                q.thread = null;
            return s;
        }
        else if (s == COMPLETING) // cannot time out yet
            Thread.yield();
      	// 【2】
        else if (q == null)
            q = new WaitNode();
        else if (!queued)
            queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                 q.next = waiters, q);
      	// 【3】
        else if (timed) {
            nanos = deadline - System.nanoTime();
            if (nanos <= 0L) {
                removeWaiter(q);
                return state;
            }
            LockSupport.parkNanos(this, nanos);
        }
        else
            LockSupport.park(this);
    }
}

【1】首先，判斷狀態，如果狀態大于COMPLETING，執行全部結束，是可以拿到結果了的，就直接回傳狀態，如果自己執行緒的節點已經產生，需要把節點中的執行緒設定為null，注意這里并沒有執行洗掉節點的操作，如果剛好處于COMPLETING狀態，說明計算已經結束，正在進行結果或執行例外的設定，這個操作非常快，那就再等等（Thread.yield()），另外，這里可以想象COMPLETING狀態是一個非常短暫的狀態，所以是放在后面判斷的，一般代碼都以主意這種細節，
【2】通過前面兩個判斷表示還未執行結束，那么就需要進入等待了，進入等待前，先要往鏈表里放節點，如果鏈表還沒節點，就new WaitNode()初始化一個節點，然后再下次回圈的時候放入鏈表，放入的方式就是CAS比對頭節點(waiters)是否變化設定，
【3】阻塞執行緒就是呼叫LockSupport.park方法阻塞執行緒，有阻塞就會有喚醒，正常喚醒執行緒的時候就是計算結束的時候，那么就會執行【1】的邏輯，退出回圈；例外的喚醒有可能是執行緒發生中斷，前面代碼中對執行緒中斷標記的處理，會移除節點（removeWaiter）并拋出例外，另外，超時情況發生的時候，也會移除節點，

finishCompletion

這個方法在任務執行結束或取消的時候執行，前面提到過的其中執行結束的兩種情況是正常執行結束和例外結束，它需要把等待的節點中的執行緒全部喚醒，在了解了鏈表結構后，我們看一下這個喚醒操作的代碼：

private void finishCompletion() {
    // assert state > COMPLETING;
    for (WaitNode q; (q = waiters) != null;) {
        if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
            for (;;) {
                Thread t = q.thread;
                if (t != null) {
                    q.thread = null;
                    LockSupport.unpark(t);
                }
                WaitNode next = q.next;
                if (next == null)
                    break;
                q.next = null; // unlink to help gc
                q = next;
            }
            break;
        }
    }

    done();

    callable = null;        // to reduce footprint
}

遍歷節點前會先用CAS的方式將頭設定成null，成功設定才能繼續，所以這里有兩個for回圈，第二個for回圈是遍歷鏈表，找出Thread不為空的節點，用LockSupport.unpark喚醒，被喚醒的執行緒會從awaitDone的park處醒來繼續執行，

其中留了一個done()方法提供給子類擴展，很多字類實作了這個擴展，比如說guava的ListenableFutureTask，

removeWaiter

在awaitDone方法中的回圈中，判斷出執行緒有中斷標記的時候會執行removeWaiter，還有就是get超時也會觸發，

private void removeWaiter(WaitNode node) {
    if (node != null) {
        node.thread = null;
        retry:
        for (;;) {          // restart on removeWaiter race
            for (WaitNode pred = null, q = waiters, s; q != null; q = s) {
                s = q.next;
                if (q.thread != null)
                    pred = q;
                else if (pred != null) {
                    pred.next = s;
                    if (pred.thread == null) // check for race
                        continue retry;
                }
                else if (!UNSAFE.compareAndSwapObject(this, waitersOffset,
                                                      q, s))
                    continue retry;
            }
            break;
        }
    }
}

一個鏈表中并發洗掉隨機節點自然會有沖突問題，比如同時洗掉的節點為相鄰節點，前面的節點的next可能只想null導致鏈表斷裂，那么這里是如何避免這種問題的呢？

首先，這個方法進入的時候第一步就會把節點的thread設定為null，實際這個操作是可以作為當前執行緒正在洗掉這個節點的標記，其他執行緒只要判斷節點是否為null就可以推算出可能有執行緒正在洗掉這個節點，

其次，每個節點都會先判斷thread是否為空，不為空則會設定給pred，也就是說pred只要有節點這個節點在從節點移除前thread都是不為空的，如果判斷出節點的thread為空，那么就跳過判斷進入下一個節點的判斷，那么這個節點就自然鏈表中移除了，因為上一個節點的next會指向到thread不為空的下一個節點（pred.next = s），當next指向后，會再判斷pred的thread是否為空，如果是為空就表示可能有執行緒并發操作，這里就直接從頭回圈鏈表，

最后，前兩個判斷都不成立的情況只有一種那就是頭節點的thread為空的情況，此時就要用cas的方式來處理如果設定失敗，和前面操作鏈表一樣自旋即可，

cancel

public boolean cancel(boolean mayInterruptIfRunning) {
  	// 【1】
    if (!(state == NEW &&
          UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
              mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
        return false;
  	// 【2】
    try {    // in case call to interrupt throws exception
        if (mayInterruptIfRunning) {
            try {
                Thread t = runner;
                if (t != null)
                    t.interrupt();
            } finally { // final state
                UNSAFE.putOrderedInt(this, stateOffset, INTERRUPTED);
            }
        }
    } finally {
      	//【3】
        finishCompletion();
    }
    return true;
}

【1】第一個判斷就是要求狀態必須是NEW，如果任務已經開始執行，那么直接就回傳false，如果呼叫cancel方法時狀態是NEW，那么直接對這個狀態進行CAS修改，如果傳參值mayInterruptIfRunning未true，那么狀態先改成INTERRUPTING，然后改成INTERRUPTED；如果傳參值是true，狀態修改為CANCELLED，直接進入終態，這一步修改動作也可能失敗，失敗意味著裝已經從剛剛的NEW發生了變化，那么就不能在進行cancel操作了，直接回傳false，
【2】上面的代碼執行成功，意味這狀態成功從NEW改成了INTERRUPTING或CANCELLED，
- 如果傳參為true，此時狀態必然已經是INTERRUPTING，然后就開始進行執行緒中斷操作，并最終將狀態變更為INTERRUPTED，
- 如果傳參為false，此時狀態為CANCELLED，已是終態，回傳true即可
【3】無論是INTERRUPTED還是CANCELLED的結果，都會執行finishCompletion方法，該方法前面已詳細決議，

后記

《Netty實戰》中有寫到JDK中Future所提供的實作只允許手動檢查對應的操作是否完成，或者一直阻塞知道它完成，這是非常煩瑣的，所以Netty提供了自己的實作，所以下一站，ChannelFuture，

參考

https://yangsanity.me/2021/07/27/FutureTask/

https://en.wikipedia.org/wiki/Treiber_stack

https://www.cnblogs.com/iwehdio/p/14285282.html

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/498429.html

標籤：Java

上一篇：聊聊 Netty 那些事兒之 Reactor 在 Netty 中的實作(創建篇)

下一篇：查詢效率提升10倍！3種優化方案，幫你解決MySQL深分頁問題