CompletableFuture學(xué)習(xí)總結(jié)

簡介

CompletableFuture結(jié)合了Future的優(yōu)點，提供了非常強大的Future的擴展功能，可以幫助我們簡化異步編程的復(fù)雜性，提供了函數(shù)式編程的能力，可以通過回調(diào)的方式處理計算結(jié)果，并且提供了轉(zhuǎn)換和組合CompletableFuture的方法。

CompletableFuture被設(shè)計在Java中進行異步編程。異步編程意味著在主線程之外創(chuàng)建一個獨立的線程，與主線程分隔開，并在上面運行一個非阻塞的任務(wù)，然后通知主線程進展，成功或者失敗。

CompletableFuture是由Java8引入的，在Java8之前我們一般通過Future實現(xiàn)異步。
Future用于表示異步計算的結(jié)果，只能通過阻塞或者輪詢的方式獲取結(jié)果，而且不支持設(shè)置回調(diào)方法，Java8之前若要設(shè)置回調(diào)一般會使用guava的ListenableFuture。 CompletableFuture對Future進行了擴展，可以通過設(shè)置回調(diào)的方式處理計算結(jié)果，同時也支持組合操作，支持進一步的編排，同時一定程度解決了回調(diào)地獄的問題。

核心概念

CompletableFuture 是一個非常強大的并發(fā)工具類，它實現(xiàn)了 Future 和 CompletionStage 接口，用于表示某個異步計算的結(jié)果，與傳統(tǒng)的 Future 不同， CompletableFuture 提供了函數(shù)式編程的方法，可以更容易地組織異步代碼，處理回調(diào)和組合多個異步操作。

假設(shè)，有一個電商網(wǎng)站，用戶瀏覽產(chǎn)品詳情頁時，需要展示產(chǎn)品的基本信息、價格、庫存、用戶評價等多個方面的數(shù)據(jù)，這些數(shù)據(jù)可能來自不同的數(shù)據(jù)源或服務(wù)，比如：

1. 產(chǎn)品基本信息 可能來自一個主數(shù)據(jù)庫。
2. 價格 和 庫存 可能需要實時從另一個庫存服務(wù)獲取。
3. 用戶評價 可能存儲在另一個專門用于用戶反饋的系統(tǒng)中。

為了提升用戶體驗，希望這些數(shù)據(jù)的獲取能夠并行進行，而不是一個接一個地串行獲取，這就是 CompletableFuture 的經(jīng)典場景。

CompletableFuture 類在主要用來解決異步編程和并發(fā)執(zhí)行的問題，在傳統(tǒng)的同步編程模型中，代碼的執(zhí)行通常是阻塞的，即一行代碼執(zhí)行完成后，下一行代碼才能開始執(zhí)行，這種模型在處理耗時操作時，如 I/O 操作、數(shù)據(jù)庫訪問或網(wǎng)絡(luò)請求，會導(dǎo)致線程長時間閑置，等待操作完成，從而降低系統(tǒng)的吞吐量和響應(yīng)能力。

因此， CompletableFuture 類提供了一種非阻塞的、基于回調(diào)的編程方式，可以在等待某個長時間運行的任務(wù)完成時，同時執(zhí)行其他任務(wù)，這樣，就可以更充分地利用系統(tǒng)資源，提高程序的并發(fā)性和響應(yīng)速度。

使用 CompletableFuture 通常用于解決以下類似場景的問題：

1. 發(fā)起異步請求 ：當(dāng)用戶請求一個產(chǎn)品詳情頁時，后端服務(wù)可以同時發(fā)起對三個數(shù)據(jù)源的異步請求，這可以通過創(chuàng)建三個 CompletableFuture 實例來實現(xiàn)，每個實例負責(zé)一個數(shù)據(jù)源的請求。
2. 處理異步結(jié)果 ：一旦這些異步請求發(fā)出，它們就可以獨立地執(zhí)行，主線程可以繼續(xù)處理其他任務(wù)，當(dāng)某個 CompletableFuture 完成時，它會包含一個結(jié)果（或者是執(zhí)行過程中的異常）。
3. 組合異步結(jié)果 ：使用 CompletableFuture 的組合方法（如 thenCombine 、 thenAcceptBoth 或 allOf ），可以等待所有異步操作完成，并將它們的結(jié)果組合在一起，比如，可以等待產(chǎn)品基本信息、價格和庫存以及用戶評價都返回后，再將這些數(shù)據(jù)整合到一個響應(yīng)對象中，返回給前端。
4. 異常處理 ：如果在獲取某個數(shù)據(jù)源時發(fā)生異常， CompletableFuture 允許以異步的方式處理這些異常，比如通過 exceptionally 方法提供一個默認的備選結(jié)果或執(zhí)行一些清理操作。
5. 最終響應(yīng) ：一旦所有數(shù)據(jù)源的數(shù)據(jù)都成功獲取并組合在一起，或者某個數(shù)據(jù)源發(fā)生異常并得到了妥善處理，服務(wù)就可以將最終的產(chǎn)品詳情頁響應(yīng)發(fā)送給前端用戶。

使用 CompletableFuture 可以高效的并發(fā)數(shù)據(jù)獲取，提升系統(tǒng)的響應(yīng)速度和整體性能。

核心API

CompletableFuture 列用于表示某個異步計算的結(jié)果，它提供了函數(shù)式編程的方法來處理異步計算，允許以非阻塞的方式編寫并發(fā)代碼，并且可以鏈接多個異步操作，以下是一些常用方法的含義：

1、靜態(tài)工廠方法

• CompletableFuture.supplyAsync(Supplier supplier) : 異步執(zhí)行給定的 Supplier ，并返回一個表示結(jié)果的新 CompletableFuture 。
• CompletableFuture.supplyAsync(Supplier supplier, Executor executor) : 使用指定的執(zhí)行器異步執(zhí)行給定的 Supplier 。
• CompletableFuture.runAsync(Runnable runnable) : 異步執(zhí)行給定的 Runnable ，并返回一個表示其完成的新 CompletableFuture 。
• CompletableFuture.runAsync(Runnable runnable, Executor executor) : 使用指定的執(zhí)行器異步執(zhí)行給定的 Runnable 。

2、完成時的處理

• thenApply(Function fn) : 當(dāng)此 CompletableFuture 完成時，對其結(jié)果應(yīng)用給定的函數(shù)。
• thenAccept(Consumer action) : 當(dāng)此 CompletableFuture 完成時，執(zhí)行給定的操作。
• thenRun(Runnable action) : 當(dāng)此 CompletableFuture 完成時，執(zhí)行給定的無參數(shù)操作。

3、異常處理

• exceptionally(Function fn) : 當(dāng)此 CompletableFuture 異常完成時，對其異常應(yīng)用給定的函數(shù)。

4、組合多個 CompletableFuture

• thenCombine(CompletableFuture other, BiFunction fn) : 當(dāng)此 CompletableFuture 和另一個都完成時，使用給定的函數(shù)組合它們的結(jié)果。
• thenAcceptBoth(CompletableFuture other, BiConsumer action) : 當(dāng)此 CompletableFuture 和另一個都完成時，對它們的結(jié)果執(zhí)行給定的操作。
• runAfterBoth(CompletableFuture other, Runnable action) : 當(dāng)此 CompletableFuture 和另一個都完成時，執(zhí)行給定的操作。
• applyToEither(CompletableFuture other, Function fn) : 當(dāng)此 CompletableFuture 或另一個完成時（哪個先完成），對其結(jié)果應(yīng)用給定的函數(shù)。
• acceptEither(CompletableFuture other, Consumer action) : 當(dāng)此 CompletableFuture 或另一個完成時（哪個先完成），對其結(jié)果執(zhí)行給定的操作。
• runAfterEither(CompletableFuture other, Runnable action) : 當(dāng)此 CompletableFuture 或另一個完成時（哪個先完成），執(zhí)行給定的操作。

5、等待和獲取結(jié)果

• get() : 等待計算完成，然后獲取其結(jié)果。
• get(long timeout, TimeUnit unit) : 等待計算在給定的時間內(nèi)完成，并獲取其結(jié)果。
• join() : 類似于 get() ，但是會在計算未完成時拋出未檢查的異常。
• complete(T value) : 如果尚未完成，則設(shè)置此 CompletableFuture 的結(jié)果。
• completeExceptionally(Throwable ex) : 如果尚未完成，則使此 CompletableFuture 異常完成。

6、取消

• cancel(boolean mayInterruptIfRunning) : 嘗試取消此 CompletableFuture 。
• isCancelled() : 如果此 CompletableFuture 被取消，則返回 true 。

7、查詢

• isDone() : 如果此 CompletableFuture 完成（無論是正常完成還是異常完成），則返回 true 。

封裝計算邏輯的CompletableFuture

上面的代碼允許我們選擇任何并發(fā)執(zhí)行的機制，但是如果我們想跳過這個樣板文件，簡單地異步執(zhí)行一些代碼呢？

靜態(tài)方法runAsync和supplyAsync允許我們相應(yīng)地使用Runnable和Supplier函數(shù)類型創(chuàng)建一個可完成的未來實例。

Runnable和Supplier都是函數(shù)接口，由于新的java8特性，它們允許將實例作為lambda表達式傳遞。

Runnable接口與線程中使用的舊接口相同，不允許返回值。

Supplier接口是一個通用函數(shù)接口，它有一個方法，該方法沒有參數(shù)，并且返回一個參數(shù)化類型的值。

這允許我們提供一個供應(yīng)商實例作為lambda表達式來執(zhí)行計算并返回結(jié)果。簡單到：

CompletableFuture?future
??=?CompletableFuture.supplyAsync(()?->?"Hello");

//?...

assertEquals("Hello",?future.get());

異步計算的處理結(jié)果

處理計算結(jié)果的最通用的方法是將其提供給函數(shù)。thenApply方法正是這樣做的；它接受一個函數(shù)實例，用它來處理結(jié)果，并返回一個包含函數(shù)返回值的Future：

CompletableFuture?completableFuture
??=?CompletableFuture.supplyAsync(()?->?"Hello");

CompletableFuture?future?=?completableFuture
??.thenApply(s?->?s?+?"?World");

assertEquals("Hello?World",?future.get());

如果我們不需要在Future中返回值，我們可以使用Consumer函數(shù)接口的實例。它的單個方法接受一個參數(shù)并返回void。

在可完成的將來，有一種方法可以解決這個用例。thenAccept方法接收使用者并將計算結(jié)果傳遞給它。最后一個future.get（）調(diào)用返回Void類型的實例：

CompletableFuture?completableFuture
??=?CompletableFuture.supplyAsync(()?->?"Hello");

CompletableFuture?future?=?completableFuture
??.thenAccept(s?->?System.out.println("Computation?returned:?"?+?s));

future.get();

最后，如果我們既不需要計算的值，也不想返回值，那么我們可以將一個可運行的lambda傳遞給thenRun方法。在下面的示例中，我們只需在調(diào)用future.get（）后在控制臺中打印一行：

CompletableFuture?completableFuture?
??=?CompletableFuture.supplyAsync(()?->?"Hello");

CompletableFuture?future?=?completableFuture
??.thenRun(()?->?System.out.println("Computation?finished."));

future.get();

組合CompletableFuture

CompletableFuture API最好的部分是能夠在一系列計算步驟中組合CompletableFuture實例。

這種鏈接的結(jié)果本身就是一個完整的Future，允許進一步的鏈接和組合。這種方法在函數(shù)語言中普遍存在，通常被稱為享元模式。

在下面的示例中，我們使用thenCompose方法按順序鏈接兩個Future。

請注意，此方法接受一個返回CompletableFuture實例的函數(shù)。此函數(shù)的參數(shù)是上一計算步驟的結(jié)果。這允許我們在下一個CompletableFuture的lambda中使用此值：

CompletableFuture?completableFuture?
??=?CompletableFuture.supplyAsync(()?->?"Hello")
????.thenCompose(s?->?CompletableFuture.supplyAsync(()?->?s?+?"?World"));

assertEquals("Hello?World",?completableFuture.get());

thenCompose方法與thenApply一起實現(xiàn)了享元模式的基本構(gòu)建塊。它們與流的map和flatMap方法以及java8中的可選類密切相關(guān)。

兩個方法都接收一個函數(shù)并將其應(yīng)用于計算結(jié)果，但是thencomose（flatMap）方法接收一個返回另一個相同類型對象的函數(shù)。這種功能結(jié)構(gòu)允許將這些類的實例組合為構(gòu)建塊。

如果我們想執(zhí)行兩個獨立的未來，并對它們的結(jié)果進行處理，我們可以使用thenCombine方法，該方法接受一個未來和一個具有兩個參數(shù)的函數(shù)來處理這兩個結(jié)果：

CompletableFuture?completableFuture?
??=?CompletableFuture.supplyAsync(()?->?"Hello")
????.thenCombine(CompletableFuture.supplyAsync(
??????()?->?"?World"),?(s1,?s2)?->?s1?+?s2));

assertEquals("Hello?World",?completableFuture.get());

一個簡單的例子是，當(dāng)我們想處理兩個CompletableFuture的結(jié)果時，但不需要將任何結(jié)果值傳遞給CompletableFuture的鏈。thenAcceptBoth方法可以幫助：

CompletableFuture?future?=?CompletableFuture.supplyAsync(()?->?"Hello")
??.thenAcceptBoth(CompletableFuture.supplyAsync(()?->?"?World"),
????(s1,?s2)?->?System.out.println(s1?+?s2));

thenApply（）和thenCompose（）方法之間的區(qū)別

在前面的部分中，我們展示了有關(guān)thenApply（）和thenCompose（）的示例。兩個api都有助于鏈接不同的CompletableFuture調(diào)用，但這兩個函數(shù)的用法不同。

thenApply()

我們可以使用此方法處理上一次調(diào)用的結(jié)果。但是，需要記住的一點是，返回類型將由所有調(diào)用組合而成。

因此，當(dāng)我們要轉(zhuǎn)換CompletableFuture調(diào)用的結(jié)果時，此方法非常有用：

CompletableFuture?finalResult?=?compute().thenApply(s->?s?+?1);

thenCompose()

thenCompose（）方法與thenApply（）類似，因為兩者都返回一個新的完成階段。但是，thencose（）使用前一階段作為參數(shù)。它將展平并直接返回一個帶有結(jié)果的CompletableFuture，而不是我們在thenApply（）中觀察到的嵌套CompletableFuture：

CompletableFuture?computeAnother(Integer?i){
????return?CompletableFuture.supplyAsync(()?->?10?+?i);
}
CompletableFuture?finalResult?=?compute().thenCompose(this::computeAnother);

因此，如果要鏈接可完成的CompletableFuture方法，那么最好使用thenCompose（）。

另外，請注意，這兩個方法之間的差異類似于map（）和flatMap（）之間的差異。

并行運行多個CompletableFuture

當(dāng)我們需要并行執(zhí)行多個期貨時，我們通常希望等待所有Supplier執(zhí)行，然后處理它們的組合結(jié)果。

CompletableFuture.allOf靜態(tài)方法允許等待的所有Supplier的完成：

CompletableFuture?future1??
??=?CompletableFuture.supplyAsync(()?->?"Hello");
CompletableFuture?future2??
??=?CompletableFuture.supplyAsync(()?->?"Beautiful");
CompletableFuture?future3??
??=?CompletableFuture.supplyAsync(()?->?"World");

CompletableFuture?combinedFuture?
??=?CompletableFuture.allOf(future1,?future2,?future3);

//?...

combinedFuture.get();

assertTrue(future1.isDone());
assertTrue(future2.isDone());
assertTrue(future3.isDone());

注意CompletableFuture.allOf（）的返回類型是CompletableFuture。這種方法的局限性在于它不能返回所有Supplier的組合結(jié)果。相反，我們必須從未來手動獲取結(jié)果。幸運的是，CompletableFuture.join（）方法和Java 8 Streams API使它變得簡單：

String?combined?=?Stream.of(future1,?future2,?future3)
??.map(CompletableFuture::join)
??.collect(Collectors.joining("?"));

assertEquals("Hello?Beautiful?World",?combined);

join（）方法類似于get方法，但是如果Future不能正常完成，它會拋出一個未檢查的異常。這樣就可以將其用作Stream.map（）方法中的方法引用。

具體使用簡單demo案例

返回結(jié)果：