淺談CompletableFuture

作者簡介：大家好，我是smart哥，前中興通訊、美團架構師，現某互聯網公司CTO

聯系qq：184480602，加我進群，大家一起學習，一起進步，一起對抗互聯網寒冬

回顧FutureTask

之前我們已經學習過FutureTask以及線程池繼承體系，里面介紹了線程池是如何借助FutureTask返回異步結果的：

而FutureTask#run()的作用就是執行任務，并把最終結果設置到FutureTask.outcome中：

FutureTask#get()為了能獲取到最終結果，內部會阻塞，直到outcome被賦值：

基于以上原因，實際編程中如果期望得到異步結果，一般有兩種方式：

FutureTask#get()阻塞等待
判斷FutureTask#isDone()，如果為true則返回

第一種大家都比較熟悉，下面演示第二種：

@RunWith(SpringRunner.class)
@SpringBootTest
public class CompletableFutureTest {private final ExecutorService executor = Executors.newFixedThreadPool(5);/*** 輪詢異步結果并獲取** @throws ExecutionException* @throws InterruptedException*/@Testpublic void testFutureAsk() throws ExecutionException, InterruptedException {// 任務1Future<String> runnableFuture = executor.submit(new Runnable() {@Overridepublic void run() {try {System.out.println("Runnable異步線程開始...");TimeUnit.SECONDS.sleep(3);System.out.println("Runnable異步線程結束...");} catch (InterruptedException e) {e.printStackTrace();}}}, "fakeRunnableResult");// 任務2Future<String> callableFuture = executor.submit(new Callable<String>() {@Overridepublic String call() throws Exception {System.out.println("Callable異步線程開始...");TimeUnit.SECONDS.sleep(3);System.out.println("Callable異步線程結束...");return "callableResult";}});boolean runnableDone = false;boolean callableDone = false;// 不斷輪詢，直到所有任務結束while (true) {TimeUnit.MILLISECONDS.sleep(500);System.out.println("輪詢異步結果...");if (runnableFuture.isDone()) {System.out.println("Runnable執行結果:" + runnableFuture.get());runnableDone = true;}if (callableFuture.isDone()) {System.out.println("Callable執行結果:" + callableFuture.get());callableDone = true;}if (runnableDone && callableDone) {break;}}System.out.println("任務全部結束");}
}

結果

Runnable異步線程開始...

Callable異步線程開始...

輪詢異步結果...

Runnable異步線程結束...

Callable異步線程結束...

輪詢異步結果...

Runnable執行結果:fakeRunnableResult

Callable執行結果:callableResult

任務全部結束

FutureTask的不足

FutureTask其實各方面都比較完美，初見時甚至讓人驚艷，因為它允許我們獲取異步執行的結果！但FutureTask#get()本身是阻塞的，假設當前有三個下載任務在執行：

task1（預計耗時5秒）
task2（預計耗時1秒）
task3（預計耗時1秒）

如果阻塞獲取時不湊巧把task1.get()排在最前面，那么會造成一定的資源浪費，因為task2和task3早就已經準備好了，可以先拿出來處理，以獲得最佳的用戶體驗。

我們固然可以像上面的Demo一樣，結合輪詢+isDone()的方式改進，但仍存在以下問題：

輪詢間隔多少合適？
為了避免while(true)阻塞主線程邏輯，可能需要開啟單獨的線程輪詢，浪費一個線程
仍然無法處理復雜的任務依賴關系

特別是第三點，使用FutureTask幾乎難以編寫...也就是說FutureTask很難處理異步編排問題。

CompletableFuture：基于異步回調的Future

CompletableFuture VS FutureTask

廢話不多說，直接上代碼：

@Test
public void testCallBack() throws InterruptedException, ExecutionException {// 提交一個任務，返回CompletableFutureCompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(new Supplier<String>() {@Overridepublic String get() {System.out.println("=============>異步線程開始...");System.out.println("=============>異步線程為：" + Thread.currentThread().getName());try {TimeUnit.SECONDS.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("=============>異步線程結束...");return "supplierResult";}});// 阻塞獲取結果System.out.println("異步結果是:" + completableFuture.get());System.out.println("main結束");
}

結果

=============>異步線程開始...

=============>異步線程為：ForkJoinPool.commonPool-worker-9

=============>異步線程結束...

異步結果是:supplierResult

main結束

整個過程看起來和同步沒啥區別，因為我們在main線程中使用了CompletableFuture#get()，直接阻塞了。當然你也可以使用CompletableFuture#isDone()改進，但我們并不推薦你把CompletableFuture當成FutureTask使用。

兩者如此相似，有什么區別嗎？

CompletableFuture和FutureTask的異同點：

相同：都實現了Future接口，所以都可以使用諸如Future#get()、Future#isDone()、Future#cancel()等方法
不同：

- FutureTask實現了Runnable，所以它可以作為任務被執行，且內部維護outcome，可以存儲結果
- CompletableFuture沒有實現Runnable，無法作為任務被執行，所以你無法把它直接丟給線程池執行，相反地，你可以把Supplier#get()這樣的函數式接口實現類丟給它執行
- CompletableFuture實現了CompletionStage，支持異步回調

總的來說，FutureTask和CompletableFuture最大的區別在于，FutureTask需要我們主動阻塞獲取，而CompletableFuture支持異步回調（后面演示）。

如果大家拿上面的代碼和之前的線程池+Runnable/Callable對比，就會發現CompletableFuture好像承擔的其實是線程池的角色，而Supplier#get()則對應Runnable#run()、Callable#call()。但我們在分析線程池繼承體系時從未見過CompletableFuture，Supplier也只是Java8預置的函數式接口而已，并不是任務類。

也就是說，不是線程池的CompletableFuture + 不是任務類的函數式接口實例，竟然把異步任務搞定了！

所以：

CompletableFuture底層到底做了什么？
它為什么能把函數式接口的實例作為任務執行？明明既不是Runnable也不是Callable！
CompletionStage和異步回調之間有什么關系？

CompletableFuture與CompletionStage

大家可能對于CompletionStage比較陌生，沒關系，先看代碼：

@Test
public void testCallBack() throws InterruptedException, ExecutionException {// 提交一個任務，返回CompletableFuture（注意，并不是把CompletableFuture提交到線程池，它沒有實現Runnable）CompletableFuture<String> completableFuture = CompletableFuture.supplyAsync(new Supplier<String>() {@Overridepublic String get() {System.out.println("=============>異步線程開始...");System.out.println("=============>異步線程為：" + Thread.currentThread().getName());try {TimeUnit.SECONDS.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("=============>異步線程結束...");return "supplierResult";}});// 異步回調：上面的Supplier#get()返回結果后，異步線程會回調BiConsumer#accept()completableFuture.whenComplete(new BiConsumer<String, Throwable>() {@Overridepublic void accept(String s, Throwable throwable) {System.out.println("=============>異步任務結束回調...");System.out.println("=============>回調線程為：" + Thread.currentThread().getName());}});// CompletableFuture的異步線程是守護線程，一旦main結束就沒了，為了看到打印結果，需要讓main休眠一會兒System.out.println("main結束");TimeUnit.SECONDS.sleep(15);
}

結果

=============>異步線程開始...

=============>異步線程為：ForkJoinPool.commonPool-worker-9

main結束

=============>異步線程結束...

=============>異步任務結束回調...

=============>回調線程為：ForkJoinPool.commonPool-worker-9

你可以暫時把每一部分的方法理解為提交一個任務。

到這里，大家應該會有兩個疑問：

CompletionStage是什么？
它和異步回調有啥關系？

本小節先回答第一個問題，第二個問題留待后續章節闡述（見右側目錄<異步回調的實現機制>）。

主線程調用了CompletableFuture#whenComplete()：

// 異步回調：上面的Supplier#get()返回結果后，異步線程會回調BiConsumer#accept()
completableFuture.whenComplete(new BiConsumer<String, Throwable>() {@Overridepublic void accept(String s, Throwable throwable) {System.out.println("=============>異步任務結束回調...");}
});

實際上這個方法定義在CompletionStage接口中（方法超級多）：

public interface CompletionStage<T> {// 省略其他方法.../*** Returns a new CompletionStage with the same result or exception as* this stage, that executes the given action when this stage completes.** <p>When this stage is complete, the given action is invoked with the* result (or {@code null} if none) and the exception (or {@code null}* if none) of this stage as arguments.  The returned stage is completed* when the action returns.  If the supplied action itself encounters an* exception, then the returned stage exceptionally completes with this* exception unless this stage also completed exceptionally.** @param action the action to perform* @return the new CompletionStage*/public CompletionStage<T> whenComplete(BiConsumer<? super T, ? super Throwable> action);// 省略其他方法...
}

而CompletableFuture實現了whenComplete()：

public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {// 省略其他方法...public CompletableFuture<T> whenComplete(BiConsumer<? super T, ? super Throwable> action) {return uniWhenCompleteStage(null, action);}private CompletableFuture<T> uniWhenCompleteStage(Executor e, BiConsumer<? super T, ? super Throwable> f) {if (f == null) throw new NullPointerException();CompletableFuture<T> d = new CompletableFuture<T>();if (e != null || !d.uniWhenComplete(this, f, null)) {UniWhenComplete<T> c = new UniWhenComplete<T>(e, d, this, f);push(c);c.tryFire(SYNC);}return d;}// 省略其他方法...
}

所以，CompletionStage是什么呢？

我的回答是：

它是一個“很簡單”的接口。完全獨立，沒有繼承任何其他接口，所有方法都是它自己定義的。

public interface CompletionStage<T> {// 定義了超級多類似whenComplete()的方法
}

它是個不簡單的接口。因為CompletableFuture實現Future的同時，還實現了它。Future方法就6、7個，而CompletionStage的方法超級多，所以如果你打開CompletableFuture的源碼，目之所及幾乎都是它對CompletionStage的實現。

public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {// 一些字段// 實現Future的方法// 實現CompletionStage的方法// 一些私有方法，配合CompletionStage// 一些內部類，配合CompletionStage
}

異步回調其實和CompletionStage有著很大的關系（廢話，FutureTask也實現了Future，但不能異步回調）

總而言之，CompletionStage是一個接口，定義了一些方法，CompletableFuture實現了這些方法并設計出了異步回調的機制。

具體怎么做到的，稍后見分曉。

一個小細節

有一個細節，很多人應該都不曾注意：

上面注釋說的是：

異步線程會回調BiConsumer#accept()

咋一看沒什么神奇，但你如果停下來理一下自己的思路，就會發現剛才你以為的是：

異步線程會回調CompletableFuture#whenComplete()

上面介紹CompletionStage時，我說的是“主線程調用了CompletableFuture#whenComplete()”，而不是異步線程調用。

換句話說，CompletionStage中定義的諸如whenComplete()等方法雖然和異步回調有關系，但并不是最終被回調的方法，最終被回調的其實是whenComplete(BiConsumer)傳進去的BiConsumer#accept()。

到這里，你可能懵逼了，那這個CompletionStage定義了那么多方法，到底是干啥用的？異步線程又為何會回調它傳入的函數接口的方法呢，BiConsumer#accept()明明不是Runnable#run()、Callable#call()呀！

異步回調的實現機制

上面種種表述，似乎都在暗示CompletableFuture#supplyAsync()會開啟一個異步線程，然后才有后面的一系列異步回調操作。

為了更好地理解CompletableFuture的異步回調，我們對現有的疑問進行進一步的拆分：

左邊為主線程，右邊為異步線程。

異步線程哪來的，Supplier如何被執行？

之前分析過CompletableFuture和FutureTask的異同點，其中有一點提到：

CompletableFuture沒有實現Runnable，無法作為任務被執行，所以你無法把它直接丟給線程池執行，相反地，你可以把Supplier#get()這樣的函數式接口實現類丟給它執行

那么CompletableFuture為什么能執行“任務”，異步線程又是哪來的，為什么Supplier沒有實現Runnable/Callable也能被執行？

跟隨主線程進入CompletableFuture#supplyAsync()，我們會發現：

注意看Doug Lea大佬寫的注釋：

返回一個新的CompletableFuture，該future是由運行在{@link ForkJoinPool#commonPool()}中的任務異步完成的，其值是通過調用給定的Supplier獲得的。

總的來說，和FutureTask有點像，都是傳入某種參數，然后返回Future。

從Doug Lea大佬的注釋中，我們可以窺見部分重要的信息：

異步線程來自ForkJoinPool線程池
通過CompletableFuture#supplyAsync(supplier)傳入Supplier，返回CompletableFuture對象，它包含一個未來的value，且這個value會在稍后由異步線程執行Supplier#get()產生

如何驗證大佬說的是不是正確的呢？

開個玩笑，后端之事，哪輪得到前端插嘴。

來，我們一起看源碼~

我們可以看到CompletableFuture#supplyAsync(supplier)內部調用了asyncSupplyStage(asyncPool, supplier)，此時傳入了一個線程池asyncPool，它是CompletableFuture的成員變量：

useCommonPool為true時會使用ForkJoinPool，而useCommonPool取決于運行當前程序的硬件是否支持多核CPU，具體大家可以自己看源碼。

現在我們已經確定了異步線程來自ForkJoinPool，剩下的問題是，主線程傳進來的Supplier壓根沒有實現Runnable/Callable接口，怎么被異步線程執行呢？

哦~和ExecutorService#submit()一樣的套路：包裝成Task再執行。只不過這次被包裝成了AsyncSupply，而不是FutureTask：

AsyncSupply名字雖然怪異，但和當初的FutureTask頗為相似，都實現了Future和Runnable，具備任務+結果雙重屬性：

然后就是熟悉的配方：

等線程池分配出線程，最終會執行AsyncSupply#run()：

異步線程會執行AsyncSupply#run()并在方法內調用f.get()，也就是Supplier#get()，阻塞獲取結果并通過d.completeValue(v)把值設置到CompletableFuture中，而CompletableFuture d已經在上一步asyncSupplyStage()中被返回。最終效果和線程池+FutureTask是一樣的，先返回Future實例，再通過引用把值放進去。

所以，completableFuture.get()可以阻塞得到result：

至此，我們搞明白了異步線程是怎么來的，以及Supplier是如何被執行的。

從這個層面上來看，CompletableFuture相當于一個自帶線程池的Future，而CompletableFuture#supplyAsync(Supplier)倒像是ExecutorService#submit(Runnable/Callable)，內部也會包裝任務，最終丟給Executor#execute(Task)。只不過ExecutorService是把Runnable#run()/Callable#call()包裝成FutureTask，而CompletableFuture則把亂七八糟的Supplier#get()等函數式接口的方法包裝成ForkJoinTask。

異步回調的原理

阻塞get()已經不足為奇，關鍵是回調機制如何實現？

在介紹CompletableFuture的回調機制之前，先跟大家說明一下，回調并沒有大家想的那么神奇，尤其CompletableFuture的回調機制，其實本質上是對多個CompletableFuture內部函數的順序執行，只不過發起者是異步線程而不是主線程：

現在第1個問題已經解決：

第2、3兩個問題其實是同一個問題，放在一起講。

為了能更好地說明問題，我們把原本第二部分的CompletableFuture#whenComplete()換成CompletableFuture#thenApply()，本質是一樣的，順便熟悉熟悉其他方法（也是CompletableFuture對CompletionStage的實現）：

@RunWith(SpringRunner.class)
@SpringBootTest
public class CompletableFutureTest {@Testpublic void testCallBack() throws InterruptedException {// 任務一：把第一個任務推進去，順便開啟異步線程CompletableFuture<String> completableFuture1 = CompletableFuture.supplyAsync(new Supplier<String>() {@Overridepublic String get() {System.out.println("=============>異步線程開始...");try {TimeUnit.SECONDS.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("=============>completableFuture1任務結束...");System.out.println("=============>執行completableFuture1的線程為：" + Thread.currentThread().getName());return "supplierResult";}});System.out.println("completableFuture1:" + completableFuture1);// 任務二：把第二個任務推進去，等待異步回調CompletableFuture<String> completableFuture2 = completableFuture1.thenApply(new Function<String, String>() {@Overridepublic String apply(String s) {try {TimeUnit.SECONDS.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("=============>completableFuture2任務結束 result=" + s);System.out.println("=============>執行completableFuture2的線程為：" + Thread.currentThread().getName());return s;}});System.out.println("completableFuture2:" + completableFuture2);// 任務三：把第三個任務推進去，等待異步回調CompletableFuture<String> completableFuture3 = completableFuture2.thenApply(new Function<String, String>() {@Overridepublic String apply(String s) {try {TimeUnit.SECONDS.sleep(10);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("=============>completableFuture3任務結束 result=" + s);System.out.println("=============>執行completableFuture3的線程為：" + Thread.currentThread().getName());return s;}});System.out.println("completableFuture3:" + completableFuture3);System.out.println("主線程結束");TimeUnit.SECONDS.sleep(40);}
}

結果

completableFuture1:java.util.concurrent.CompletableFuture@76e4212[Not completed]

=============>異步線程開始...

completableFuture2:java.util.concurrent.CompletableFuture@23121d14[Not completed]

completableFuture3:java.util.concurrent.CompletableFuture@72af90e8[Not completed]

主線程結束

=============>completableFuture1任務結束...

=============>執行completableFuture1的線程為：ForkJoinPool.commonPool-worker-9

=============>completableFuture2任務結束 result=supplierResult

=============>執行completableFuture2的線程為：ForkJoinPool.commonPool-worker-9

=============>completableFuture3任務結束 result=supplierResult

=============>執行completableFuture3的線程為：ForkJoinPool.commonPool-worker-9

分析主線程的主干：

CompletableFuture#supplyAsync(Supplier)：包裝Supplier為AsyncSupply，調用executor.execute()，等待異步線程回調Supplier#get()
CompletableFuture#thenApply(Function)
CompletableFuture#thenApply(Function)

在之前介紹CompletionStage時，我很對不住這位老哥，隆重介紹了一番，結果發現CompletionStage#whenComplete()、CompletionStage#thenApply()竟然是主線程調用的，而不是異步回調。那么，主線程調用whenComplete(BiConsumer)、thenApply(Function)時到底做了什么，導致異步線程最終會執行BiConsumer#accept()、Function#apply()呢？

請大家把上面代碼中的休眠時間統一改為100秒，方便留出分析時間。

然后再跟著我打幾個斷點：

點進CompletableFuture#thenApply(Function)：

OK，DEBUG模式啟動測試案例，多體會幾遍Main的執行流程，必要時停止程序熟悉一下代碼。

五分鐘后...

應該已經走過好幾遍了吧？讓我們一起來看看。

相信大家對于uniApply()印象很深，因為對于主線程而言這個方法幾乎相當于沒有執行，每次都返回了false。

CompletableFutureTest目前有三塊代碼：任務一、任務二、任務三。

主線程在執行“任務一”的CompletableFuture#supplyAsync(Supplier)時，將Supplier包裝成AsyncSupply任務，并開啟了異步線程，此后異步線程會阻塞在Supplier#get()：

也就是說，Supplier#get()是異步線程開啟后執行的第一站！

與此同時，主線程繼續執行后面的“任務二”、“任務三”，并且都會到達uniApply()，且都返回false，因為a.result==null。

以“任務二”為例，當主線程從任務二進來，調用thenApply()：

最終會到達uniApply()，通過控制臺的日志，我們發現a其實就是completableFuture1：

因為uniApply()的上一步傳入的this：

也就是說：

主線程 ---> completableFuture1.thenApply(Function#apply) ---> !d.uniApply(this, f#apply, null)

a.result就是completableFuture1.result，而completableFuture1的值來自Supplier#get()，此時確實還是null（異步線程阻塞100秒后才會）。

所以此時d.uniApply(this, f, null) 為false，那么!d.uniApply(this, f, null) 為true，就會進入if語句：

主要做了3件事：

傳入Executor e、新建的CompletableFuture d、當前completableFuture1、Function f，構建UniApply
push(uniApply)
uniApply.tryFire(SYNC)

任務一做了兩件事：

開啟異步線程
等待回調

由于要開啟線程，自己也要作為任務被執行，所以Supplier#get()被包裝成AsyncSupply，是一個Task。而后續的幾個任務其實只做了一件事：等待回調。只要能通過實例執行方法即可，和任務一有所不同，所以只是被包裝成UniApply對象。

push(uniApply)姑且認為是把任務二的Function#apply()包裝后塞到任務棧中。

但uniApply.tryFire(SYNC)是干嘛的呢？里面又調了一次uniApply()：

SYNC=0，所以最終判斷!d.uniApply(this, f, this) ==true，tryFire(SYNC)返回null，后面的d.postFire(a, mode)此時并不會執行，等后面異步線程復蘇后，帶著任務一的結果再次調用時，效果就截然不同了。

總結一下，“任務二”、“任務三”操作都是一樣的，都做了3件事：

主線程調用CompletableFuture#thenApply(Function f)傳入f，構建UniApply對象，包裝Function#apply()
把構建好的UniApply對象push到棧中
返回CompletableFuture d

綠色的是異步線程，此時阻塞等待Supplier#get()，但主線程沒閑著，正在努力構建任務棧。

等過了100秒，supplyAsync(Supplier)中的Supplier#get()返回結果后，異步線程繼續往下走：

看到了嗎，postComplete()也會走uniApply()，但這次已經有了異步結果result，所以流程不會被截斷，最終會調用Function#apply(s)，而這個s是上一個函數的執行結果。也就是說，新的CompletableFuture對象調用Function#apply()處理了上一個CompletableFuture產生的結果。

最后，為CompletionStage老哥扳回顏面，你是最棒的：

CompletableFuture#whenComplete(BiConsumer)、CompletableFuture#thenApply(Function)等方法的目的是把BiConsumer#accept()及Function#apply()等回調函數封裝成一個個UniApply對象被壓入棧，等異步線程執行時，再逐個彈棧并回調。

黑線先行，綠色的異步線程阻塞一會后再走，此時主線程已經成功構建任務棧，引導異步線程去執行即可。

所以，總的來說CompletionStage設計得很巧妙，Doug Lea老爺子不愧是獨立設計了JUC的男人，數據結構功底和對編程的理解舉世無雙。

當然，CompletableFuture還有其他很多的API，甚至回調任務過程中還可以再開異步線程，本文只分析了supplyAsync()+thenApply()，但原理大致相同。老實說，內部的實現機制比較復雜，個人不建議繼續深入研究源碼，意義不大。

CompletableFuture與FutureTask線程數對比

CompletableFuture和FutureTask耗費的線程數是一致的，但對于FutureTask來說，無論是輪詢還是阻塞get，都會導致主線程無法繼續其他任務，又或者主線程可以繼續其他任務，但要時不時check FutureTask是否已經完成任務，比較糟心。而CompletableFuture則會根據我們編排的順序逐個回調，是按照既定路線執行的。

其實無論是哪種方式，異步線程其實都需要阻塞等待結果，期間不能處理其他任務。但對于FutureTask而言，在異步線程注定無法復用的前提下，如果想要獲取最終結果，需要主線程主動查詢或者額外開啟一個線程查詢，并且可能造成阻塞，而CompletableFuture的異步任務執行、任務結果獲取都是異步線程獨立完成。

所以：

1個異步線程阻塞執行任務 + 回調異步結果 > 1個異步線程阻塞執行任務 + 1個線程阻塞查詢任務

問題

鑒于篇幅較長，內容較深，所以設置一些問題，強制大家去思考整理吧。

假設有以下代碼：

public class CompletableFutureTest {@Testpublic void testCallBack() throws InterruptedException {// 任務一CompletableFuture<String> completableFuture1 = CompletableFuture.supplyAsync(new Supplier<String>() {@Overridepublic String get() {return "supplierResult";}});System.out.println("completableFuture1:" + completableFuture1);// 任務二CompletableFuture<String> completableFuture2 = completableFuture1.thenApply(new Function<String, String>() {@Overridepublic String apply(String s) {return s;}});System.out.println("completableFuture2:" + completableFuture2);// 任務三CompletableFuture<String> completableFuture3 = completableFuture2.thenApply(new Function<String, String>() {@Overridepublic String apply(String s) {return s;}});System.out.println("completableFuture3:" + completableFuture3);System.out.println("主線程結束");TimeUnit.SECONDS.sleep(40);}
}