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結構圖：
image

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。

參考

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

//en.wikipedia.org/wiki/Treiber_stack

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

Tags:

VirMach 便宜 VPS

QNews

QNews