作者： 一字馬胡
轉(zhuǎn)載標(biāo)志 【2017-12-07】

更新日志

Future

 * A {@code Future} represents the result of an asynchronous
 * computation.  Methods are provided to check if the computation is
 * complete, to wait for its completion, and to retrieve the result of
 * the computation.  The result can only be retrieved using method
 * {@code get} when the computation has completed, blocking if
 * necessary until it is ready.  Cancellation is performed by the
 * {@code cancel} method.  Additional methods are provided to
 * determine if the task completed normally or was cancelled. Once a
 * computation has completed, the computation cannot be cancelled.
 * If you would like to use a {@code Future} for the sake
 * of cancellability but not provide a usable result, you can
 * declare types of the form {@code Future<?>} and
 * return {@code null} as a result of the underlying task.

上面這段文字已經(jīng)說明了Future的本質(zhì)甘苍，一個Future代表一個異步計算的結(jié)果，并且它提供一些方法來讓調(diào)用者檢測異步過程是否完成宅楞，或者取得異步計算的結(jié)果刻获，或者取消正在執(zhí)行的異步任務(wù)。本文將分析總結(jié)Future的一些實現(xiàn)細(xì)節(jié)定嗓，希望可以弄明白Future的原理枯怖。

為了有一定的目標(biāo)性注整，本文將選取Future的一個基本實現(xiàn)FutureTask來進(jìn)行分析總結(jié)，其他更為復(fù)雜豐富的Future實現(xiàn)日后再進(jìn)行分析總結(jié)度硝。下面的圖片展示了FutureTask的類關(guān)系圖设捐，下文會對FutureTask進(jìn)行詳細(xì)的分析：

從類圖可以看出，F(xiàn)utureTask實現(xiàn)了Runnable接口和Future接口塘淑，下面的圖片展示了FutureTask提供的一些接口萝招，下文中將對其中的一些接口做詳細(xì)分析。

FutureTask

首先來看一下FutureTask的一些關(guān)鍵字段存捺，第一個需要注意的是state字段槐沼，看下面的代碼：

    /*
     * 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;

state代表當(dāng)前任務(wù)的狀態(tài)曙蒸，NEW代表當(dāng)前以及獲取到任務(wù)，準(zhǔn)備開始執(zhí)行任務(wù)岗钩。COMPLETING狀態(tài)代表正在進(jìn)行任務(wù)處理中纽窟，NORMAL表示任務(wù)執(zhí)行結(jié)束，并且任務(wù)處理結(jié)果正常兼吓，沒有異常出現(xiàn)臂港，EXCEPTIONAL則表示執(zhí)行任務(wù)的過程中出現(xiàn)了異常，CANCELLED表示任務(wù)被取消了视搏，INTERRUPTING表示任務(wù)在執(zhí)行過程中被中斷审孽，是一個中間狀態(tài)，INTERRUPTED表示中斷結(jié)束浑娜。這些狀態(tài)的可能轉(zhuǎn)換關(guān)系在上面的注釋中可以看到佑力，可以發(fā)現(xiàn)總共只有四種狀態(tài)轉(zhuǎn)移路徑，在下文的某些方法分析中還會提到state筋遭。

第二個需要關(guān)注的字段是callable字段打颤，這就是實際需要執(zhí)行的任務(wù)，而結(jié)果將被設(shè)置到outcome字段中去漓滔，runner字段代表了運行任務(wù)的線程编饺。waiters字段則代表阻塞在該Future上的線程鏈表，可以看一下waiters的數(shù)據(jù)結(jié)構(gòu)：

    /**
     * Simple linked list nodes to record waiting threads in a Treiber
     * stack.  See other classes such as Phaser and SynchronousQueue
     * for more detailed explanation.
     */
    static final class WaitNode {
        volatile Thread thread;
        volatile WaitNode next;
        WaitNode() { thread = Thread.currentThread(); }
    }

可以看到是一個非常簡單的單鏈表數(shù)據(jù)結(jié)構(gòu)响驴。下面來看一下FutureTask的構(gòu)造函數(shù)透且，F(xiàn)utureTask有兩個構(gòu)造函數(shù)，分別如下：

    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
    }

可以看到踏施，第一個構(gòu)造函數(shù)傳遞了一個Callable類型的參數(shù)，構(gòu)造函數(shù)將該Callable參數(shù)初始化給類的callback字段罕邀，并且初始化state為NEW畅形。第二個構(gòu)造函數(shù)傳遞了兩個參數(shù)，一個是Runnable類型的runnable诉探，代表需要執(zhí)行的任務(wù)日熬，以及任務(wù)的返回結(jié)果，其實還是使用這兩個參數(shù)來構(gòu)造出一個callback肾胯，并且執(zhí)行和第一個構(gòu)造函數(shù)一樣的邏輯竖席，下面是如何使用Runnable和result來構(gòu)造出一個callback的剩下細(xì)節(jié)：

    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;
        }
    }

下面來看一下Future的一個核心方法get的實現(xiàn)細(xì)節(jié)，下面是get方法的具體代碼：

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

首先獲取到任務(wù)的當(dāng)前狀態(tài)敬肚，如果狀態(tài)小于等于COMPLETING毕荐，那么根據(jù)最開始的定義，可以知道目前的狀態(tài)只可能是NEW或者COMPLETING艳馒，也就是任務(wù)還沒有開始執(zhí)行憎亚，或者還在繼續(xù)執(zhí)行沒有結(jié)束员寇，那么就調(diào)用awaitDone方法來進(jìn)行等待任務(wù)執(zhí)行完成，否則第美，也就是說蝶锋，當(dāng)任務(wù)的當(dāng)前狀態(tài)大于COMPLETING的時候，那么當(dāng)前狀態(tài)可能為：

• NORMAL 正常結(jié)束
• EXCEPTIONAL 異常結(jié)束
• CANCELLED 被取消
• INTERRUPTING 正在中斷
• INTERRUPTED 中斷結(jié)束

先來看第一條分支什往，也就是調(diào)用awaitDone的具體流程扳缕。

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

這個方法還是很復(fù)雜的，下面根據(jù)分支來分析一下這個方法都在做什么事情：

• 如果當(dāng)前線程被中斷别威，那么就將所有等待在該Future上的線程都從阻塞鏈表移除
• 如果發(fā)現(xiàn)任務(wù)的狀態(tài)變?yōu)槟撤Nfinal態(tài)的時候躯舔，也就是state大于COMPLETING的時候，說明任務(wù)執(zhí)行已經(jīng)結(jié)束了（無論是怎么結(jié)束的都不再運行中）兔港，那么返回任務(wù)的狀態(tài)庸毫，并且清除等待在該Future上的線程
• 如果發(fā)現(xiàn)任務(wù)的當(dāng)前狀態(tài)為COMPLETING，那么說明任務(wù)正在執(zhí)行過程中衫樊，需要等待一下
• 否則飒赃，就需要等待了，如果配置了不等待科侈，那么不會將當(dāng)前線程添加到等待鏈表中载佳，否則將當(dāng)前線程添加到等待鏈表中去
• 如果配置了等待時間，那么就需要判斷是否超時了臀栈，如果超時了蔫慧，那么就阻塞等待，如果沒有設(shè)置超時時間权薯，那么就會一直阻塞等待下去直到任務(wù)處理完成（不管怎么完成）

上面分析完了當(dāng)狀態(tài)小于等于COMPLETING的時候的處理流程姑躲，下面來看一下當(dāng)任務(wù)的狀態(tài)大于COMPLETING的時候的處理流程：

    private V report(int s) throws ExecutionException {
        Object x = outcome;
        if (s == NORMAL)
            return (V)x;
        if (s >= CANCELLED)
            throw new CancellationException();
        throw new ExecutionException((Throwable)x);
    }

需要判斷任務(wù)的當(dāng)前狀態(tài)，其實現(xiàn)在可以知道目前的狀態(tài)可能為什么盟蚣，任務(wù)肯定已經(jīng)不再運行了黍析，要么正常結(jié)束，要么異常結(jié)束屎开，要么被中斷阐枣，根據(jù)不同的情況進(jìn)行不同的處理，比如當(dāng)發(fā)現(xiàn)狀態(tài)為NORMAL的時候奄抽，就判斷為任務(wù)正常結(jié)束蔼两，處理結(jié)果應(yīng)該保存在outcome中，所以返回outcom就可以了逞度，當(dāng)發(fā)現(xiàn)任務(wù)時被取消的時候额划，get操作會拋出異常，其他情況也會拋出異常來告知調(diào)用者發(fā)生的情況档泽。Future除了提供不帶參數(shù)的get方法外锁孟，還提供了一個帶參數(shù)的get方法彬祖，也就是可以設(shè)置等待時間，超時會拋出異常品抽，下面是帶參數(shù)的get方法的細(xì)節(jié)：

    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);
    }

該方法的實現(xiàn)細(xì)節(jié)與不帶參數(shù)的get方法一樣储笑，只是增加了超時機制，等待時間超過了設(shè)定的時間之后就會拋出TimeoutException異常圆恤。該方法和不帶參數(shù)的get方法是共用一個awaitDone方法來實現(xiàn)任務(wù)結(jié)果的等待獲取的突倍，所以就不再往下贅述了。

下面再來看一個Future的比較重要的方法cancel盆昙，也就是取消任務(wù)的執(zhí)行羽历，它的具體實現(xiàn)細(xì)節(jié)看下面的代碼：

    public boolean cancel(boolean mayInterruptIfRunning) {
        if (!(state == NEW &&
              UNSAFE.compareAndSwapInt(this, stateOffset, NEW,
                  mayInterruptIfRunning ? INTERRUPTING : CANCELLED)))
            return false;
        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 {
            finishCompletion();
        }
        return true;
    }

首先判斷任務(wù)的當(dāng)前狀態(tài)，如果不為NEW或者試圖將任務(wù)的狀態(tài)設(shè)置為NEW失敗之后淡喜，就會返回false告訴用戶cancel失敗了秕磷，否則，就調(diào)用負(fù)責(zé)執(zhí)行任務(wù)的線程的interrupt方法來結(jié)束任務(wù)的運行炼团，并且會更新任務(wù)的狀態(tài)澎嚣。在finally中會調(diào)用一個方法finishCompletion，下面是這個方法的具體實現(xiàn)細(xì)節(jié)：

    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
    }

這個方法做的事情就是告訴所有等待在該Future上的線程瘟芝，讓他們別等了易桃，任務(wù)已經(jīng)被cancel了，再等下去也不會有結(jié)果了锌俱。

文章開頭給出了FutureTask的類關(guān)系圖晤郑，并且知道了FutureTask繼承了Runnable，我們在創(chuàng)建了一個FutureTask之后贸宏，會使用線程池來執(zhí)行這個FutureTask造寝，最后會執(zhí)行FutureTask的run方法，所以最為重要的就是FutureTask的run方法吭练，下面開始分析FutureTask的run方法的具體實現(xiàn)細(xì)節(jié)：

   public void run() {
        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;
                    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);
        }
    }

首選诫龙，如果任務(wù)的當(dāng)前狀態(tài)不是NEW，或者試圖將任務(wù)的狀態(tài)變?yōu)镹EW失敗的時候线脚，或者試圖設(shè)置runner字段為當(dāng)前的線程的時候遇到失敗赐稽，也就是獲取到了執(zhí)行任務(wù)的具體Thread叫榕，但是設(shè)置字段失敗浑侥，run方法都將直接返回，這也就說明了為什么說Future是不可逆的晰绎，只能執(zhí)行一次寓落。接著，run方法獲取到了具體的任務(wù)荞下，并且再次判斷該任務(wù)的狀態(tài)是否為NEW伶选，以及判斷任務(wù)是否為null史飞，如果這些判斷都通過的話，那么就可以執(zhí)行任務(wù)了仰税，具體的執(zhí)行就是調(diào)用Callable的call方法來獲取結(jié)果构资，如果在執(zhí)行過程中拋出異常，那么就需要調(diào)用setException來設(shè)置具體的異常陨簇，否則調(diào)用set方法來設(shè)置任務(wù)的執(zhí)行結(jié)果吐绵，下面先來看setException的具體細(xì)節(jié)：

    protected void setException(Throwable t) {
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = t;
            UNSAFE.putOrderedInt(this, stateOffset, EXCEPTIONAL); // final state
            finishCompletion();
        }
    }

這個方法會設(shè)置任務(wù)的狀態(tài)為EXCEPTIONAL，并且調(diào)用finishCompletion來做一些任務(wù)的收尾工作河绽。下面來看一下正常結(jié)束時候的set細(xì)節(jié)：

    protected void set(V v) {
        if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
            outcome = v;
            UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
            finishCompletion();
        }
    }

和setException一樣己单，只是將任務(wù)的狀態(tài)設(shè)置為了NORMAL而不是EXCEPTIONAL，這樣調(diào)用線程在進(jìn)行g(shù)et方法調(diào)用的時候就可以獲取到正常的結(jié)果了耙饰。

到此纹笼，本文分析了Future的基本實現(xiàn)，并且基于FutureTask來進(jìn)行具體的分析苟跪，思路更加清晰廷痘，再次需要說明的是，一個Future代表一個異步計算的結(jié)果削咆，我們可以取消任務(wù)牍疏，可以等待任務(wù)，并且可以設(shè)置一個超時時間以控制等待時間拨齐，當(dāng)然鳞陨，本文的目的是初步理解Future的原理，為了深刻理解Future的原理瞻惋，需要做更為復(fù)雜豐富的分析總結(jié)厦滤，下一步可以借助CompletableFuture來深入學(xué)習(xí)Future，關(guān)于CompletableFuture的一些基礎(chǔ)知識歼狼，可以參考文章Java CompletableFuture掏导，對于CompletableFuture的更為深入的學(xué)習(xí)總結(jié)將在未來適宜的時候進(jìn)行。