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

簡(jiǎn)介

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

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

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

核心概念

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

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

1. 產(chǎn)品基本信息 可能來(lái)自一個(gè)主數(shù)據(jù)庫(kù)。
2. 價(jià)格 和 庫(kù)存 可能需要實(shí)時(shí)從另一個(gè)庫(kù)存服務(wù)獲取。
3. 用戶(hù)評(píng)價(jià) 可能存儲(chǔ)在另一個(gè)專(zhuān)門(mén)用于用戶(hù)反饋的系統(tǒng)中。

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

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

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

使用 CompletableFuture 通常用于解決以下類(lèi)似場(chǎng)景的問(wèn)題：

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

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

核心API

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

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

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

2、完成時(shí)的處理

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

3、異常處理

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

4、組合多個(gè) CompletableFuture

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

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

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

6、取消

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

7、查詢(xún)

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

封裝計(jì)算邏輯的CompletableFuture

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

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

Runnable和Supplier都是函數(shù)接口，由于新的java8特性，它們?cè)试S將實(shí)例作為lambda表達(dá)式傳遞。

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

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

這允許我們提供一個(gè)供應(yīng)商實(shí)例作為lambda表達(dá)式來(lái)執(zhí)行計(jì)算并返回結(jié)果。簡(jiǎn)單到：

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

//?...

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

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

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

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

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

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

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

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

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

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

future.get();

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

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

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

future.get();

組合CompletableFuture

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

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

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

請(qǐng)注意，此方法接受一個(gè)返回CompletableFuture實(shí)例的函數(shù)。此函數(shù)的參數(shù)是上一計(jì)算步驟的結(jié)果。這允許我們?cè)谙乱粋�(gè)CompletableFuture的lambda中使用此值：

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

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

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

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

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

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

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

一個(gè)簡(jiǎn)單的例子是，當(dāng)我們想處理兩個(gè)CompletableFuture的結(jié)果時(shí)，但不需要將任何結(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（）的示例。兩個(gè)api都有助于鏈接不同的CompletableFuture調(diào)用，但這兩個(gè)函數(shù)的用法不同。

thenApply()

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

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

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

thenCompose()

thenCompose（）方法與thenApply（）類(lèi)似，因?yàn)閮烧叨挤祷匾粋�(gè)新的完成階段。但是，thencose（）使用前一階段作為參數(shù)。它將展平并直接返回一個(gè)帶有結(jié)果的CompletableFuture，而不是我們?cè)趖henApply（）中觀察到的嵌套CompletableFuture：

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

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

另外，請(qǐng)注意，這兩個(gè)方法之間的差異類(lèi)似于map（）和flatMap（）之間的差異。

并行運(yùn)行多個(gè)CompletableFuture

當(dāng)我們需要并行執(zhí)行多個(gè)期貨時(shí)，我們通常希望等待所有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（）的返回類(lèi)型是CompletableFuture。這種方法的局限性在于它不能返回所有Supplier的組合結(jié)果。相反，我們必須從未來(lái)手動(dòng)獲取結(jié)果。幸運(yùn)的是，CompletableFuture.join（）方法和Java 8 Streams API使它變得簡(jiǎn)單：

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

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

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

具體使用簡(jiǎn)單demo案例

返回結(jié)果：