異步神器：CompletableFuture實作原理和使用場景

1.概述

CompletableFuture是jdk1.8引入的實作類。擴充了Future和CompletionStage，是一個可以在任務完成階段觸發一些操作Future。簡單的來講就是可以實作異步回調。

2.為什麼引入CompletableFuture

對于jdk1.5的Future，雖然提供了異步處理任務的能力，但是擷取結果的方式很不優雅，還是需要通過阻塞（或者輪訓）的方式。如何避免阻塞呢？其實就是注冊回調。

業界結合觀察者模式實作異步回調。也就是當任務執行完成後去通知觀察者。比如Netty的ChannelFuture，可以通過注冊監聽實作異步結果的處理。

Netty的ChannelFuture

public Promise<V> addListener(GenericFutureListener<? extends Future<? super V>> listener){  
    checkNotNull(listener, "listener");  
    synchronized (this) {  
        addListener0(listener);  
    }  
    if (isDone()) {  
        notifyListeners();  
    }  
    return this;  
}  
private boolean setValue0(Object objResult){  
    if (RESULT_UPDATER.compareAndSet(this, null, objResult) ||  
        RESULT_UPDATER.compareAndSet(this, UNCANCELLABLE, objResult)) {  
        if (checkNotifyWaiters()) {  
            notifyListeners();  
        }  
        return true;  
    }  
    return false;  
}

通過addListener方法注冊監聽。如果任務完成，會調用notifyListeners通知。

CompletableFuture通過擴充Future，引入函數式程式設計，通過回調的方式去處理結果。

3.功能

CompletableFuture的功能主要展現在他的CompletionStage。

可以實作如下等功能

轉換（thenCompose）
組合（thenCombine）
消費（thenAccept）
運作（thenRun）。
帶傳回的消費（thenApply）

消費和運作的差別：

消費使用執行結果。運作則隻是運作特定任務。具體其他功能大家可以根據需求自行檢視。

CompletableFuture借助CompletionStage的方法可以實作鍊式調用。并且可以選擇同步或者異步兩種方式。

這裡舉個簡單的例子來體驗一下他的功能。

public static void thenApply(){  
    ExecutorService executorService = Executors.newFixedThreadPool(2);  
    CompletableFuture cf = CompletableFuture.supplyAsync(() -> {  
        try {  
            //  Thread.sleep(2000);  
        } catch (Exception e) {  
            e.printStackTrace();  
        }  
        System.out.println("supplyAsync " + Thread.currentThread().getName());  
        return "hello";  
    }, executorService).thenApplyAsync(s -> {  
        System.out.println(s + "world");  
        return "hhh";  
    }, executorService);  
    cf.thenRunAsync(() -> {  
        System.out.println("ddddd");  
    });  
    cf.thenRun(() -> {  
        System.out.println("ddddsd");  
    });  
    cf.thenRun(() -> {  
        System.out.println(Thread.currentThread());  
        System.out.println("dddaewdd");  
    });  
}

執行結果

supplyAsync pool-1-thread-1  
helloworld  
ddddd  
ddddsd  
Thread[main,5,main]  
dddaewdd

根據結果我們可以看到會有序執行對應任務。

注意：

如果是同步執行cf.thenRun。他的執行線程可能main線程，也可能是執行源任務的線程。如果執行源任務的線程在main調用之前執行完了任務。那麼cf.thenRun方法會由main線程調用。

這裡說明一下，如果是同一任務的依賴任務有多個：

如果這些依賴任務都是同步執行。那麼假如這些任務被目前調用線程（main）執行，則是有序執行，假如被執行源任務的線程執行，那麼會是倒序執行。因為内部任務資料結構為LIFO。
如果這些依賴任務都是異步執行，那麼他會通過異步線程池去執行任務。不能保證任務的執行順序。

上面的結論是通過閱讀源代碼得到的。下面我們深入源代碼。

4.源碼追蹤

建立CompletableFuture

建立的方法有很多，甚至可以直接new一個。我們來看一下supplyAsync異步建立的方法。

public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,  
                                                   Executor executor){  
    return asyncSupplyStage(screenExecutor(executor), supplier);  
}  
static Executor screenExecutor(Executor e){  
    if (!useCommonPool && e == ForkJoinPool.commonPool())  
        return asyncPool;  
    if (e == null) throw new NullPointerException();  
    return e;  
}

入參Supplier，帶傳回值的函數。如果是異步方法，并且傳遞了執行器，那麼會使用傳入的執行器去執行任務。否則采用公共的ForkJoin并行線程池，如果不支援并行，建立一個線程去執行。

這裡我們需要注意ForkJoin是通過守護線程去執行任務的。是以必須有非守護線程的存在才行。

asyncSupplyStage方法

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

這裡會建立一個用于傳回的CompletableFuture。

然後構造一個AsyncSupply，并将建立的CompletableFuture作為構造參數傳入。

那麼，任務的執行完全依賴AsyncSupply。

AsyncSupply#run

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

該方法會調用Supplier的get方法。并将結果設定到CompletableFuture中。我們應該清楚這些操作都是在異步線程中調用的。
d.postComplete 方法就是通知任務執行完成。觸發後續依賴任務的執行，也就是實作CompletionStage的關鍵點。

在看postComplete方法之前我們先來看一下建立依賴任務的邏輯。

thenAcceptAsync方法

public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action){  
    return uniAcceptStage(asyncPool, action);  
}  
private CompletableFuture<Void> uniAcceptStage(Executor e,  
                                               Consumer<? super T> f){  
    if (f == null) throw new NullPointerException();  
    CompletableFuture<Void> d = new CompletableFuture<Void>();  
    if (e != null || !d.uniAccept(this, f, null)) {  
        # 1  
        UniAccept<T> c = new UniAccept<T>(e, d, this, f);  
        push(c);  
        c.tryFire(SYNC);  
    }  
    return d;  
}

上面提到過。thenAcceptAsync是用來消費CompletableFuture的。該方法調用uniAcceptStage。

uniAcceptStage邏輯：

構造一個CompletableFuture，主要是為了鍊式調用。
如果為異步任務，直接傳回。因為源任務結束後會觸發異步線程執行對應邏輯。
如果為同步任務（e==null），會調用d.uniAccept方法。這個方法在這裡邏輯：如果源任務完成，調用f，傳回true。否則進入if代碼塊（Mark 1）。
如果是異步任務直接進入if（Mark 1）。

Mark1邏輯：

構造一個UniAccept，将其push入棧。這裡通過CAS實作樂觀鎖實作。
調用c.tryFire方法。

final CompletableFuture<Void> tryFire(int mode){  
    CompletableFuture<Void> d; CompletableFuture<T> a;  
    if ((d = dep) == null ||  
        !d.uniAccept(a = src, fn, mode > 0 ? null : this))  
        return null;  
    dep = null; src = null; fn = null;  
    return d.postFire(a, mode);  
}

會調用d.uniAccept方法。其實該方法判斷源任務是否完成，如果完成則執行依賴任務，否則傳回false。
如果依賴任務已經執行，調用d.postFire，主要就是Fire的後續處理。根據不同模式邏輯不同。

這裡簡單說一下，其實mode有同步異步，和疊代。疊代為了避免無限遞歸。

這裡強調一下d.uniAccept方法的第三個參數。

如果是異步調用（mode>0），傳入null。否則傳入this。

差別看下面代碼。c不為null會調用c.claim方法。

try {  
    if (c != null && !c.claim())  
        return false;  
    @SuppressWarnings("unchecked") S s = (S) r;  
    f.accept(s);  
    completeNull();  
} catch (Throwable ex) {  
    completeThrowable(ex);  
}  
  
final boolean claim(){  
    Executor e = executor;  
    if (compareAndSetForkJoinTaskTag((short)0, (short)1)) {  
        if (e == null)  
            return true;  
        executor = null; // disable  
        e.execute(this);  
    }  
    return false;  
}

claim方法是邏輯：

如果異步線程為null。說明同步，那麼直接傳回true。最後上層函數會調用f.accept(s)同步執行任務。
如果異步線程不為null，那麼使用異步線程去執行this。

this的run任務如下。也就是在異步線程同步調用tryFire方法。達到其被異步線程執行的目的。

public final void run(){   
   tryFire(ASYNC);   
}

看完上面的邏輯，我們基本了解依賴任務的邏輯。

其實就是先判斷源任務是否完成，如果完成，直接在對應線程執行以來任務（如果是同步，則在目前線程處理，否則在異步線程處理）

如果任務沒有完成，直接傳回，因為等任務完成之後會通過postComplete去觸發調用依賴任務。

postComplete方法

final void postComplete(){  
    /*  
     * On each step, variable f holds current dependents to pop  
     * and run.  It is extended along only one path at a time,  
     * pushing others to avoid unbounded recursion.  
     */  
    CompletableFuture<?> f = this; Completion h;  
    while ((h = f.stack) != null ||  
           (f != this && (h = (f = this).stack) != null)) {  
        CompletableFuture<?> d; Completion t;  
        if (f.casStack(h, t = h.next)) {  
            if (t != null) {  
                if (f != this) {  
                    pushStack(h);  
                    continue;  
                }  
                h.next = null;    // detach  
            }  
            f = (d = h.tryFire(NESTED)) == null ? this : d;  
        }  
    }  
}

在源任務完成之後會調用。

其實邏輯很簡單，就是疊代堆棧的依賴任務。調用h.tryFire方法。NESTED就是為了避免遞歸死循環。因為FirePost會調用postComplete。如果是NESTED，則不調用。

堆棧的内容其實就是在依賴任務建立的時候加入進去的。上面我們已經提到過。

4.總結

基本上述源碼已經分析了邏輯。

因為涉及異步等操作，我們需要理一下（這裡針對全異步任務）：

建立CompletableFuture成功之後會通過異步線程去執行對應任務。
如果CompletableFuture還有依賴任務（異步），會将任務加入到CompletableFuture的堆棧儲存起來。以供後續完成後執行依賴任務。

當然，建立依賴任務并不隻是将其加入堆棧。如果源任務在建立依賴任務的時候已經執行完成，那麼目前線程會觸發依賴任務的異步線程直接處理依賴任務。并且會告訴堆棧其他的依賴任務源任務已經完成。

主要是考慮代碼的複用。是以邏輯相對難了解。

postComplete方法會被源任務線程執行完源任務後調用。同樣也可能被依賴任務線程後調用。

如果是目前依賴任務線程，那麼會執行依賴任務，并且會通知其他依賴任務。
如果是源任務線程，和其他依賴任務線程，則将任務轉換給依賴線程去執行。不需要通知其他依賴任務，避免死遞歸。

異步神器：CompletableFuture實作原理和使用場景

1.概述

2.為什麼引入CompletableFuture

Netty的ChannelFuture

3.功能

4.源碼追蹤

建立CompletableFuture

asyncSupplyStage方法

AsyncSupply#run

thenAcceptAsync方法

postComplete方法

4.總結

繼續閱讀

C++ 堆和棧

分析解決OOM與JVM參數調優

JVM調優案例-NewRatio使用

變量的含義

JavaScript 如何執行上下文

python 簡單實作堆棧原理

性能分析之CPU分析-從CPU調用到具體代碼行（C/C++）

JVM 參數監控之 jinfo

java Stack 類（棧）

LeetCode_394. Decode String 字元串解碼

Spring Boot：異步任務與定時任務

堆棧實作的數學運算表達式計算堆棧實作的數學運算表達式計算

Java資料結構和算法（五）棧介紹棧的應用場景棧的快速入門棧實作綜合電腦

cpu負載過高分析

dfs深度優先搜尋_圖的深度優先搜尋（DFS）

詳解STM32單片機的堆棧