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

更新日志

導(dǎo)入

深度學(xué)習(xí)Java Future 系列：

• 深度學(xué)習(xí)Java Future （一）

第一篇文章基于FutureTask的Future基本實(shí)現(xiàn)來分析了Java Future的基本原理拍谐，F(xiàn)utureTask只是Future接口的一個基本實(shí)現(xiàn)秧廉，并且是作為一個Task對象存在的膛壹，F(xiàn)utureTask本身并不管理執(zhí)行線程池相關(guān)的內(nèi)容，我們生成一個FutureTask對象的動機(jī)是我們希望將我們的task包裝成一個FutureTask對象赵刑，使得我們可以借助FutureTask的特性來控制我們的任務(wù)分衫。雖然FutureTask較為簡單，但是可以從FutureTask的具體實(shí)現(xiàn)中學(xué)習(xí)一些Future的知識料睛，至少對于Future的定位應(yīng)該是更進(jìn)一步的丐箩，在進(jìn)行接下來的內(nèi)容之前摇邦，需要再次重申的是，F(xiàn)uture是一個可以代表異步計(jì)算結(jié)果的對象屎勘，并且Future提供了一些方法來讓調(diào)用者控制任務(wù)施籍，比如可以取消任務(wù)的執(zhí)行（當(dāng)然可能取消會失敗）概漱，或者設(shè)置超時時間來取得我們的任務(wù)的運(yùn)行結(jié)果丑慎。本文是深度學(xué)習(xí)Java Future 系列的第二篇文章，和第一篇文章借助FutureTask的具體實(shí)現(xiàn)來學(xué)習(xí)一樣瓤摧，本文也將借助一個具體的Future實(shí)現(xiàn)來分析總結(jié)竿裂，因?yàn)镃ompletableFuture在平時的開發(fā)中使用的頻率較高，所以本文將選擇使用CompletableFuture的具體實(shí)現(xiàn)來繼續(xù)分析Future照弥，試圖通過分析CompletableFuture的某些方法的實(shí)現(xiàn)來學(xué)習(xí)關(guān)于Future更為深層次的知識腻异。

下面的圖片展示了CompletableFuture的類圖關(guān)系：

可以看到，CompletableFuture同時實(shí)現(xiàn)了兩個接口这揣，分別為Future和CompletionStage悔常，CompletionStage是CompletableFuture提供的一些非常豐富的接口，可以借助這些接口來實(shí)現(xiàn)非常復(fù)雜的異步計(jì)算工作给赞，基于本文的主題是Future机打，所以本文不會過多的分析關(guān)于CompletionStage的內(nèi)容，如果想要了解CompletableFuture中關(guān)于CompletionStage的一些細(xì)節(jié)內(nèi)容片迅，可以參考文章Java CompletableFuture残邀，該文章詳細(xì)完整的描述了CompletableFuture中關(guān)于CompletionStage接口的實(shí)現(xiàn)情況。

CompletableFuture

首先來分析一下CompletableFuture的get方法的實(shí)現(xiàn)細(xì)節(jié)柑蛇，CompletableFuture實(shí)現(xiàn)了Future的所有接口芥挣，包括兩個get方法，一個是不帶參數(shù)的get方法耻台，一個是可以設(shè)置等待時間的get方法九秀，首先來看一下CompletableFuture中不帶參數(shù)的get方法的具體實(shí)現(xiàn)：

    public T get() throws InterruptedException, ExecutionException {
        Object r;
        return reportGet((r = result) == null ? waitingGet(true) : r);
    }

result字段代表任務(wù)的執(zhí)行結(jié)果，所以首先判斷是否為null粘我，為null則表示任務(wù)還沒有執(zhí)行結(jié)束，那么就會調(diào)用waitingGet方法來等待任務(wù)執(zhí)行完成痹换，如果result不為null征字，那么說明任務(wù)已經(jīng)成功執(zhí)行結(jié)束了，那么就調(diào)用reportGet來返回結(jié)果娇豫，下面先來看一下waitingGet方法的具體實(shí)現(xiàn)細(xì)節(jié)：

   /**
     * Returns raw result after waiting, or null if interruptible and
     * interrupted.
     */
    private Object waitingGet(boolean interruptible) {
        Signaller q = null;
        boolean queued = false;
        int spins = -1;
        Object r;
        while ((r = result) == null) {
            if (spins < 0)
                spins = (Runtime.getRuntime().availableProcessors() > 1) ?
                    1 << 8 : 0; // Use brief spin-wait on multiprocessors
            else if (spins > 0) {
                if (ThreadLocalRandom.nextSecondarySeed() >= 0)
                    --spins;
            }
            else if (q == null)
                q = new Signaller(interruptible, 0L, 0L);
            else if (!queued)
                queued = tryPushStack(q);
            else if (interruptible && q.interruptControl < 0) {
                q.thread = null;
                cleanStack();
                return null;
            }
            else if (q.thread != null && result == null) {
                try {
                    ForkJoinPool.managedBlock(q);
                } catch (InterruptedException ie) {
                    q.interruptControl = -1;
                }
            }
        }
        if (q != null) {
            q.thread = null;
            if (q.interruptControl < 0) {
                if (interruptible)
                    r = null; // report interruption
                else
                    Thread.currentThread().interrupt();
            }
        }
        postComplete();
        return r;
    }

這個方法的實(shí)現(xiàn)時比較復(fù)雜的匙姜，方法中有幾個地方需要特別注意，下面先來看一下spins是做什么的冯痢，根據(jù)注釋氮昧，可以知道spins是用來在多核心環(huán)境下的自旋操作的框杜，所謂自旋就是不斷循環(huán)等待判斷，從代碼可以看出在多核心環(huán)境下袖肥，spins會被初始化為1 << 8咪辱，然后在自旋的過程中如果發(fā)現(xiàn)spins大于0，那么就通過一個關(guān)鍵方法ThreadLocalRandom.nextSecondarySeed()來進(jìn)行spins的更新操作椎组，如果ThreadLocalRandom.nextSecondarySeed()返回的結(jié)果大于0油狂，那么spins就減1，否則不更新spins寸癌。ThreadLocalRandom.nextSecondarySeed()方法其實(shí)是一個類似于并發(fā)環(huán)境下的random专筷，是線程安全的。

接下來還需要注意的一個點(diǎn)是Signaller蒸苇，從Signaller的實(shí)現(xiàn)上可以發(fā)現(xiàn)磷蛹，Signaller實(shí)現(xiàn)了ForkJoinPool.ManagedBlocker，下面是ForkJoinPool.ManagedBlocker的接口定義：

    public static interface ManagedBlocker {
        /**
         * Possibly blocks the current thread, for example waiting for
         * a lock or condition.
         *
         * @return {@code true} if no additional blocking is necessary
         * (i.e., if isReleasable would return true)
         * @throws InterruptedException if interrupted while waiting
         * (the method is not required to do so, but is allowed to)
         */
        boolean block() throws InterruptedException;

        /**
         * Returns {@code true} if blocking is unnecessary.
         * @return {@code true} if blocking is unnecessary
         */
        boolean isReleasable();
    }

ForkJoinPool.ManagedBlocker的目的是為了保證ForkJoinPool的并行性溪烤，具體分析還需要更為深入的學(xué)習(xí)Fork/Join框架味咳。繼續(xù)回到waitingGet方法中，在自旋過程中會調(diào)用ForkJoinPool.managedBlock(ForkJoinPool.ManagedBlocker)來進(jìn)行阻塞工作氛什，實(shí)際的效果就是讓線程等任務(wù)執(zhí)行完成莺葫，CompletableFuture中與Fork/Join的交叉部分內(nèi)容不再本文的描述范圍，日后再進(jìn)行分析總結(jié)枪眉∞嗝剩總得看起來，waitingGet實(shí)現(xiàn)的功能就是等待任務(wù)執(zhí)行完成贸铜，執(zhí)行完成返回結(jié)果并做一些收尾工作堡纬。

現(xiàn)在來看reportGet方法的實(shí)現(xiàn)細(xì)節(jié)，在判斷任務(wù)執(zhí)行完成之后蒿秦，get方法會調(diào)用reportGet方法來獲取結(jié)果：

    /**
     * Reports result using Future.get conventions.
     */
    private static <T> T reportGet(Object r)
        throws InterruptedException, ExecutionException {
        if (r == null) // by convention below, null means interrupted
            throw new InterruptedException();
        if (r instanceof AltResult) {
            Throwable x, cause;
            if ((x = ((AltResult)r).ex) == null)
                return null;
            if (x instanceof CancellationException)
                throw (CancellationException)x;
            if ((x instanceof CompletionException) &&
                (cause = x.getCause()) != null)
                x = cause;
            throw new ExecutionException(x);
        }
        @SuppressWarnings("unchecked") T t = (T) r;
        return t;
    }

如果result為null烤镐，說明任務(wù)時被中斷的，拋出中斷異常棍鳖，如果result類型為AltResult炮叶，代表執(zhí)行過程中出現(xiàn)異常了，那么就拋出相應(yīng)的異常渡处，否則镜悉，返回result。

分析完了不帶參數(shù)的get方法（阻塞等待）之后医瘫，現(xiàn)在來分析一下帶超時參數(shù)的get方法的具體實(shí)現(xiàn)細(xì)節(jié)：

    public T get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException {
        Object r;
        long nanos = unit.toNanos(timeout);
        return reportGet((r = result) == null ? timedGet(nanos) : r);
    }

和不帶參數(shù)的get方法一樣侣肄，還是會判斷任務(wù)是否已經(jīng)執(zhí)行完成了，如果完成了會調(diào)用reportGet方法來返回最終的執(zhí)行結(jié)果（或者拋出異常）醇份，否則稼锅，會調(diào)用timedGet來進(jìn)行超時等待吼具，timedGet會等待一段時間，然后拋出超時異常（或者執(zhí)行結(jié)束返回正常結(jié)果）矩距，下面是timedGet方法的具體細(xì)節(jié)：

    private Object timedGet(long nanos) throws TimeoutException {
        if (Thread.interrupted())
            return null;
        if (nanos <= 0L)
            throw new TimeoutException();
        long d = System.nanoTime() + nanos;
        Signaller q = new Signaller(true, nanos, d == 0L ? 1L : d); // avoid 0
        boolean queued = false;
        Object r;
        // We intentionally don't spin here (as waitingGet does) because
        // the call to nanoTime() above acts much like a spin.
        while ((r = result) == null) {
            if (!queued)
                queued = tryPushStack(q);
            else if (q.interruptControl < 0 || q.nanos <= 0L) {
                q.thread = null;
                cleanStack();
                if (q.interruptControl < 0)
                    return null;
                throw new TimeoutException();
            }
            else if (q.thread != null && result == null) {
                try {
                    ForkJoinPool.managedBlock(q);
                } catch (InterruptedException ie) {
                    q.interruptControl = -1;
                }
            }
        }
        if (q.interruptControl < 0)
            r = null;
        q.thread = null;
        postComplete();
        return r;
    }

在timedGet中不再使用spins來進(jìn)行自旋拗盒，因?yàn)楝F(xiàn)在可以確定需要等待多少時間了。timedGet的邏輯和waitingGet的邏輯類似剩晴，畢竟都是在等待任務(wù)的執(zhí)行結(jié)果锣咒。

除了兩個get方法之前，CompletableFuture還提供了一個方法getNow赞弥，代表需要立刻返回不進(jìn)行阻塞等待毅整，下面是getNow的實(shí)現(xiàn)細(xì)節(jié)：

    public T getNow(T valueIfAbsent) {
        Object r;
        return ((r = result) == null) ? valueIfAbsent : reportJoin(r);
    }

getNow很簡單，判斷result是否為null绽左，如果不為null則直接返回悼嫉，否則返回參數(shù)中傳遞的默認(rèn)值。

分析完了get部分的內(nèi)容拼窥，下面開始分析CompletableFuture最為重要的一個部分戏蔑，就是如何開始一個任務(wù)的執(zhí)行。下文中將分析supplyAsync的具體執(zhí)行流程鲁纠，supplyAsync有兩個版本总棵，一個是不帶Executor的，還有一個是指定Executor的改含，下面首先分析一下不指定Executor的supplyAsync版本的具體實(shí)現(xiàn)流程：

    public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
        return asyncSupplyStage(asyncPool, supplier);
    }


    static <U> CompletableFuture<U> asyncSupplyStage(Executor e,
                                                     Supplier<U> f) {
        if (f == null) throw new NullPointerException();
        CompletableFuture<U> d = new CompletableFuture<U>();
        e.execute(new AsyncSupply<U>(d, f));
        return d;
    }

可以看到supplyAsync會調(diào)用asyncSupplyStage情龄，并且指定一個默認(rèn)的asyncPool來執(zhí)行任務(wù)，CompletableFuture是管理執(zhí)行任務(wù)的線程池的捍壤，這一點(diǎn)是和FutureTask的區(qū)別骤视，F(xiàn)utureTask只是一個可以被執(zhí)行的task，而CompletableFuture本身就管理者線程池鹃觉，可以由CompletableFuture本身來管理任務(wù)的執(zhí)行专酗。這個默認(rèn)的線程池是什么？

    private static final boolean useCommonPool =
        (ForkJoinPool.getCommonPoolParallelism() > 1);

    /**
     * Default executor -- ForkJoinPool.commonPool() unless it cannot
     * support parallelism.
     */
    private static final Executor asyncPool = useCommonPool ?
        ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();

首先會做一個判斷盗扇，如果條件滿足就使用ForkJoinPool的commonPool作為默認(rèn)的Executor祷肯，否則會使用一個ThreadPerTaskExecutor來作為CompletableFuture來做默認(rèn)的Executor。

接著看asyncSupplyStage疗隶，我們提交的任務(wù)會被包裝成一個AsyncSupply對象躬柬，然后交給CompletableFuture發(fā)現(xiàn)的Executor來執(zhí)行，那AsyncSupply是什么呢抽减？

   static final class AsyncSupply<T> extends ForkJoinTask<Void>
            implements Runnable, AsynchronousCompletionTask {
        CompletableFuture<T> dep; Supplier<T> fn;
        AsyncSupply(CompletableFuture<T> dep, Supplier<T> fn) {
            this.dep = dep; this.fn = fn;
        }

        public final Void getRawResult() { return null; }
        public final void setRawResult(Void v) {}
        public final boolean exec() { run(); return true; }

        public void run() {
            CompletableFuture<T> d; Supplier<T> f;
            if ((d = dep) != null && (f = fn) != null) {
                dep = null; fn = null;
                if (d.result == null) {
                    try {
                        d.completeValue(f.get());
                    } catch (Throwable ex) {
                        d.completeThrowable(ex);
                    }
                }
                d.postComplete();
            }
        }
    }

觀察到AsyncSupply實(shí)現(xiàn)了Runnable，而Executor會執(zhí)行Runnable的run方法來獲得結(jié)構(gòu)橄碾，所以主要看AsyncSupply的run方法的具體細(xì)節(jié)卵沉，可以看到颠锉，run方法中會試圖去獲取任務(wù)的結(jié)果，如果不拋出異常史汗，那么會調(diào)用CompletableFuture的completeValue方法琼掠，否則會調(diào)用CompletableFuture的completeThrowable方法，最后會調(diào)用CompletableFuture的postComplete方法來做一些收尾工作停撞，主要來看前兩個方法的細(xì)節(jié)瓷蛙，首先是completeValue方法：

    /** Completes with a non-exceptional result, unless already completed. */
    final boolean completeValue(T t) {
        return UNSAFE.compareAndSwapObject(this, RESULT, null,
                                           (t == null) ? NIL : t);
    }

completeValue方法會調(diào)用UNSAFE.compareAndSwapObject來講任務(wù)的結(jié)果設(shè)置到CompletableFuture的result字段中去。如果在執(zhí)行任務(wù)的時候拋出異常戈毒，會調(diào)用completeThrowable方法艰猬，下面是completeThrowable方法的細(xì)節(jié)：

    /** Completes with an exceptional result, unless already completed. */
    final boolean completeThrowable(Throwable x) {
        return UNSAFE.compareAndSwapObject(this, RESULT, null,
                                           encodeThrowable(x));
    }

指定Executor的supplyAsync方法和沒有指定Executor參數(shù)的supplyAsync方法的唯一區(qū)別就是執(zhí)行任務(wù)的Executor，所以不再贅述埋市。

到這里冠桃，可以知道Executor實(shí)際執(zhí)行的代碼到底是什么了，回到asyncSupplyStage方法道宅，接著就會執(zhí)行Executor.execute方法來執(zhí)行任務(wù)食听，需要注意的是，asyncSupplyStage方法返回的是一個CompletableFuture污茵，并且立刻返回的樱报，具體的任務(wù)處理邏輯是有Executor來執(zhí)行的，當(dāng)任務(wù)處理完成的時候泞当，Executor中負(fù)責(zé)處理的線程會將任務(wù)的執(zhí)行結(jié)果設(shè)置到CompletableFuture的result字段中去迹蛤。

本文的內(nèi)容到此也就結(jié)束了，上文中提到零蓉，CompletableFuture提供了大量實(shí)用的方法來支持我們的異步任務(wù)笤受，具體提供的方法可以參考上文中提供的鏈接，或者直接參考jdk源碼敌蜂、javadoc來獲取更為詳細(xì)的內(nèi)容箩兽，本文的目的是解析CompletableFuture的任務(wù)處理流程，并且試圖分析Future在CompletableFuture中的使用章喉，以更深入的理解Future汗贫，結(jié)合第一篇深度學(xué)習(xí)Java Future系列的文章，希望可以更加深入的理解Future秸脱，并且知道Future在java并發(fā)編程落包、異步計(jì)算中的重要作用。