作者： 一字馬胡
轉(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接口的一個(gè)基本實(shí)現(xiàn)，并且是作為一個(gè)Task對(duì)象存在的，F(xiàn)utureTask本身并不管理執(zhí)行線程池相關(guān)的內(nèi)容，我們生成一個(gè)FutureTask對(duì)象的動(dòng)機(jī)是我們希望將我們的task包裝成一個(gè)FutureTask對(duì)象，使得我們可以借助FutureTask的特性來控制我們的任務(wù)。雖然FutureTask較為簡(jiǎn)單，但是可以從FutureTask的具體實(shí)現(xiàn)中學(xué)習(xí)一些Future的知識(shí)，至少對(duì)于Future的定位應(yīng)該是更進(jìn)一步的，在進(jìn)行接下來的內(nèi)容之前，需要再次重申的是，F(xiàn)uture是一個(gè)可以代表異步計(jì)算結(jié)果的對(duì)象，并且Future提供了一些方法來讓調(diào)用者控制任務(wù)，比如可以取消任務(wù)的執(zhí)行（當(dāng)然可能取消會(huì)失?。蛘咴O(shè)置超時(shí)時(shí)間來取得我們的任務(wù)的運(yùn)行結(jié)果。本文是深度學(xué)習(xí)Java Future 系列的第二篇文章，和第一篇文章借助FutureTask的具體實(shí)現(xiàn)來學(xué)習(xí)一樣，本文也將借助一個(gè)具體的Future實(shí)現(xiàn)來分析總結(jié)，因?yàn)镃ompletableFuture在平時(shí)的開發(fā)中使用的頻率較高，所以本文將選擇使用CompletableFuture的具體實(shí)現(xiàn)來繼續(xù)分析Future，試圖通過分析CompletableFuture的某些方法的實(shí)現(xiàn)來學(xué)習(xí)關(guān)于Future更為深層次的知識(shí)。

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

可以看到，CompletableFuture同時(shí)實(shí)現(xiàn)了兩個(gè)接口，分別為Future和CompletionStage，CompletionStage是CompletableFuture提供的一些非常豐富的接口，可以借助這些接口來實(shí)現(xiàn)非常復(fù)雜的異步計(jì)算工作，基于本文的主題是Future，所以本文不會(huì)過多的分析關(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的所有接口，包括兩個(gè)get方法，一個(gè)是不帶參數(shù)的get方法，一個(gè)是可以設(shè)置等待時(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é)束，那么就會(huì)調(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;
    }

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

接下來還需要注意的一個(gè)點(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方法中，在自旋過程中會(huì)調(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方法會(huì)調(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ù)時(shí)被中斷的，拋出中斷異常，如果result類型為AltResult，代表執(zhí)行過程中出現(xiàn)異常了，那么就拋出相應(yīng)的異常，否則，返回result。

分析完了不帶參數(shù)的get方法（阻塞等待）之后，現(xiàn)在來分析一下帶超時(shí)參數(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方法一樣，還是會(huì)判斷任務(wù)是否已經(jīng)執(zhí)行完成了，如果完成了會(huì)調(diào)用reportGet方法來返回最終的執(zhí)行結(jié)果（或者拋出異常），否則，會(huì)調(diào)用timedGet來進(jìn)行超時(shí)等待，timedGet會(huì)等待一段時(shí)間，然后拋出超時(shí)異常（或者執(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)在可以確定需要等待多少時(shí)間了。timedGet的邏輯和waitingGet的邏輯類似，畢竟都是在等待任務(wù)的執(zhí)行結(jié)果。

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

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

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

分析完了get部分的內(nèi)容，下面開始分析CompletableFuture最為重要的一個(gè)部分，就是如何開始一個(gè)任務(wù)的執(zhí)行。下文中將分析supplyAsync的具體執(zhí)行流程，supplyAsync有兩個(gè)版本，一個(gè)是不帶Executor的，還有一個(gè)是指定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會(huì)調(diào)用asyncSupplyStage，并且指定一個(gè)默認(rèn)的asyncPool來執(zhí)行任務(wù)，CompletableFuture是管理執(zhí)行任務(wù)的線程池的，這一點(diǎn)是和FutureTask的區(qū)別，F(xiàn)utureTask只是一個(gè)可以被執(zhí)行的task，而CompletableFuture本身就管理者線程池，可以由CompletableFuture本身來管理任務(wù)的執(zhí)行。這個(gè)默認(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();

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

接著看asyncSupplyStage，我們提交的任務(wù)會(huì)被包裝成一個(gè)AsyncSupply對(duì)象，然后交給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會(huì)執(zhí)行Runnable的run方法來獲得結(jié)構(gòu)，所以主要看AsyncSupply的run方法的具體細(xì)節(jié)，可以看到，run方法中會(huì)試圖去獲取任務(wù)的結(jié)果，如果不拋出異常，那么會(huì)調(diào)用CompletableFuture的completeValue方法，否則會(huì)調(diào)用CompletableFuture的completeThrowable方法，最后會(huì)調(diào)用CompletableFuture的postComplete方法來做一些收尾工作，主要來看前兩個(gè)方法的細(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方法會(huì)調(diào)用UNSAFE.compareAndSwapObject來講任務(wù)的結(jié)果設(shè)置到CompletableFuture的result字段中去。如果在執(zhí)行任務(wù)的時(shí)候拋出異常，會(huì)調(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方法，接著就會(huì)執(zhí)行Executor.execute方法來執(zhí)行任務(wù)，需要注意的是，asyncSupplyStage方法返回的是一個(gè)CompletableFuture，并且立刻返回的，具體的任務(wù)處理邏輯是有Executor來執(zhí)行的，當(dāng)任務(wù)處理完成的時(shí)候，Executor中負(fù)責(zé)處理的線程會(huì)將任務(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ì)算中的重要作用。