併發系列（二）——FutureTask類源碼簡析

背景

　　本文基於JDK 11，主要介紹FutureTask類中的run()、get()和cancel() 方法，沒有過多解析相應interface中的註釋，但閱讀源碼時建議先閱讀註釋，明白方法的主要的功能，再去看源碼會更快。

　　文中如有不正確的地方歡迎大夥留言指出，謝謝了！

一、FutureTask類圖

　　1.1 FutureTask簡介

　　FutureTask類圖以下（使用IDEA生成）。如圖所示，FutureTask實現了Future接口的全部方法，而且實現了Runnable接口，其中，Runnable接口的現實類用於被線程執行，而Future表明的是異步計算的結果。所以，FutureTask類能夠理解爲，執行run()（實現Runnable接口中的方法），經過Future的get()方法獲取結果。

　　1.2 FutureTask的屬性

 //任務線程總共有七中狀態以下：
    * Possible state transitions:
     * 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;

    /** The underlying callable; nulled out after running */
    //在run()方法中調用
    private Callable<V> callable;
    /** The result to return or exception to throw from get() */
    //任務執行結果，callable.call()正常執行的返回值
    private Object outcome; // non-volatile, protected by state reads/writes
    /** The thread running the callable; CASed during run() */
    //任務線程
    private volatile Thread runner;
    /** Treiber stack of waiting threads */
    //等待任務結果的線程組成的節點，放在鏈表對列中
    private volatile WaitNode waiters;

 二、源碼解析

　　2.1 run()方法

public void run() {
        //一、如果任務的狀態不是NEW，且使用CAS將runner置爲當前線程則直接返回
        if (state != NEW ||
            !RUNNER.compareAndSet(this, null, Thread.currentThread()))
            return;
        try {
            Callable<V> c = callable;
            //二、任務不爲null，且state的狀態爲NEW的狀況下才執行任務
            if (c != null && state == NEW) {
                V result;
                boolean ran;
                try {
                    //執行任務並接收執行結果
                    result = c.call();
                    //正常執行結果則將標識置爲true
                    ran = true;
                } catch (Throwable ex) {
                    //三、任務發生異常，執行或cancel()，則結果置爲null，並記錄異常信息
                    result = null;
                    ran = false;
                    setException(ex);
                }
                //四、任務正常結束，則設置返回結果
                if (ran)
                    set(result);
            }
        } finally {
            // 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)
                handlePossibleCancellationInterrupt(s);
        }
    }

　　1）若任務的狀態不是NEW，或者使用CAS將runner置爲當前線程失敗，則直接返回的緣由是防止多線程調用；

　　2）再度確認任務執行的前置條件；

　　3）任務執行異常，將result置爲null，並記錄異常，setException()源碼以下：

protected void setException(Throwable t) {
        //使用CAS將狀態置爲中間態COMPLETING
        if (STATE.compareAndSet(this, NEW, COMPLETING)) {
            outcome = t;
            STATE.setRelease(this, EXCEPTIONAL); // final state
            //任務處於結束態時，遍歷喚醒等待result的線程
            finishCompletion();
        }
    }

　　任務的狀態變化爲NEW  - >  COMPLETING  ->  EXCEPTIONAL

　　4）任務正常結果則會設置result以後，喚醒waitNode的鏈表對列中等待任務結果的線程；

　　5）異常後的調用邏輯以下：

 //保證調用cancel在run方法返回以前中斷執行任務
    private void handlePossibleCancellationInterrupt(int s) {
        // It is possible for our interrupter to stall before getting a
        // chance to interrupt us.  Let's spin-wait patiently.
        if (s == INTERRUPTING)
            //自旋等待
            while (state == INTERRUPTING)
            //當前線程讓出CPU執行權
                Thread.yield(); // wait out pending interrupt
    }

 　　2.2  get()方法

　　源碼分析以下：

public V get() throws InterruptedException, ExecutionException {
        int s = state;
        if (s <= COMPLETING)
            //等待任務完成
            s = awaitDone(false, 0L);
        //返回結果
        return report(s);
    }

　　其中，等待過程分析以下：

private int awaitDone(boolean timed, long nanos)
        throws InterruptedException {
        // The code below is very delicate, to achieve these goals:
        // - call nanoTime exactly once for each call to park
        // - if nanos <= 0L, return promptly without allocation or nanoTime
        // - if nanos == Long.MIN_VALUE, don't underflow
        // - if nanos == Long.MAX_VALUE, and nanoTime is non-monotonic
        //   and we suffer a spurious wakeup, we will do no worse than
        //   to park-spin for a while
        long startTime = 0L;    // Special value 0L means not yet parked
        WaitNode q = null;
        boolean queued = false;
        for (;;) {
            int s = state;
            //一、任務的狀態已經處於最終的狀態，則將任務線程的引用置爲null，直接返回狀態
            if (s > COMPLETING) {
                if (q != null)
                    q.thread = null;
                return s;
            }
            //二、任務的狀態爲COMPLETING說明任務已經接近完成，則當前線程讓出CPU權限以便任務執行線程獲取到CPU執行權
            else if (s == COMPLETING)
                // We may have already promised (via isDone) that we are done
                // so never return empty-handed or throw InterruptedException
                Thread.yield();
            //三、當前線程被中斷，則將當前線程從等待任務結果的對列中移除，並拋出異常
            else if (Thread.interrupted()) {
                removeWaiter(q);
                throw new InterruptedException();
            }
            //四、任務線程的狀態小於COMPLETING，則將當前調用get()方法的線程新建一個Node
            else if (q == null) {
                if (timed && nanos <= 0L)
                    return s;
                q = new WaitNode();
            }
            //五、若由當前線程構成的Node未加入鏈表中，則加入
            else if (!queued)
                queued = WAITERS.weakCompareAndSet(this, q.next = waiters, q);
            //六、是否開啓了超時獲取結果
            else if (timed) {
                final long parkNanos;
                if (startTime == 0L) { // first time
                    startTime = System.nanoTime();
                    if (startTime == 0L)
                        startTime = 1L;
                    parkNanos = nanos;
                } else {
                    long elapsed = System.nanoTime() - startTime;
                    //七、超時則從棧中移除當前線程
                    if (elapsed >= nanos) {
                        removeWaiter(q);
                        return state;
                    }
                    parkNanos = nanos - elapsed;
                }
                // nanoTime may be slow; recheck before parking
                //當前線程掛起
                if (state < COMPLETING)
                    LockSupport.parkNanos(this, parkNanos);
            }
            else
                LockSupport.park(this);
        }
    }

　　獲取到返回的狀態值後，根據其狀態值判斷是返回結果仍是拋出異常。

　　2.2 cancel()方法

public boolean cancel(boolean mayInterruptIfRunning) {
        //一、若任務線程的狀態爲NEW，則將其狀態從NEW置爲INTERRUPTING、CANCELLED
        if (!(state == NEW && STATE.compareAndSet
              (this, NEW, mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
            //CAS改變任務線程的狀態失敗，則直接返回false，表示cancel失敗
            return false;
        try {    // in case call to interrupt throws exception
            //二、改變任務線程的狀態成功後，根據是否中斷running的任務線程的標識位，決定是否中斷正在運行的任務線程
            if (mayInterruptIfRunning) {
                try {
                    Thread t = runner;
                    //任務線程不爲null，則使用interrupt()中斷
                    if (t != null)
                        t.interrupt();
                } finally { // final state
                    //設置狀態
                    STATE.setRelease(this, INTERRUPTED);
                }
            }
        } finally {
            //三、清理等待任務結果的等待線程
            finishCompletion();
        }
        return true;
    }

 三、總結

　　1）執行run()方法，是在調用在Callable的call()方法，其實在初始化時被指定；

　　2）調用get()方法，如果任務線程還在執行，則會把調用get的線程封裝成waitNode塞入到FutureTask類內部的阻塞鏈表對列中，能夠有多個線程同時調用get()方法；

　　3）cancel()方法是經過對任務線程調用interrupt()實現；