搞定 CompletableFuture，并發異步編程和撰寫串行程式還有什么區別？你們要的多圖長文-有解無憂

你有一個思想，我有一個思想，我們交換后，一個人就有兩個思想

If you can NOT explain it simply, you do NOT understand it well enough

現陸續將Demo代碼和技術文章整理在一起 Github實踐精選，方便大家閱讀查看，本文同樣收錄在此，覺得不錯，還請Star??

前言

上一篇文章不會用Java Future，我懷疑你泡茶沒我快全面分析了 Future，通過它我們可以獲取執行緒的執行結果，它雖然解決了 Runnable 的 “三無” 短板，但是它自身還是有短板：

不能手動完成計算

假設你使用 Future 運行子執行緒呼叫遠程 API 來獲取某款產品的最新價格，服務器由于洪災宕機了，此時如果你想手動結束計算，而是想回傳上次快取中的價格，這是 Future 做不到的

呼叫 get() 方法會阻塞程式

Future 不會通知你它的完成，它提供了一個get()方法，程式呼叫該方法會阻塞直到結果可用為止，沒有辦法利用回呼函式附加到Future，并在Future的結果可用時自動呼叫它

不能鏈式執行

燒水泡茶中，通過建構式傳參做到多個任務的鏈式執行，萬一有更多的任務，或是任務鏈的執行順序有變，對原有程式的影響都是非常大的

整合多個 Future 執行結果方式笨重

假設有多個 Future 并行執行，需要在這些任務全部執行完成之后做后續操作，Future 本身是做不到的，需要借助工具類 Executors 的方法

<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
<T> T invokeAny(Collection<? extends Callable<T>> tasks)

沒有例外處理

Future 同樣沒有提供很好的例外處理方案

上一篇文章看 Future 覺得是發現了新天地，這么一說有感徑訓到了解放前

對于 Java 后端的同學，在 Java1.8 之前想實作異步編程，還想避開上述這些煩惱，ReactiveX 應該是一個常見解決方案（做Android 的應該會有了解），如果熟悉前端同學， ES6 Promise（男朋友的承諾）也解決了異步編程的煩惱

天下語言都在彼此借鑒相應優點，Java 作為老牌勁旅自然也要解決上述問題，又是那個男人，并發大師 Doug Lea 憂天下程式員之憂，解天下程式員之困擾，在 Java1.8 版本（Lambda 橫空出世）中，新增了一個并發工具類 CompletableFuture，它的出現，讓人在泡茶程序中，品嘗到了不一樣的味道......

幾個重要 Lambda 函式

CompletableFuture 在 Java1.8 的版本中出現，自然也得搭上 Lambda 的順風車，為了更好的理解 CompletableFuture，這里我需要先介紹一下幾個 Lambda 函式，我們只需要關注它們的以下幾點就可以：

引數接受形式
回傳值形式
函式名稱

Runnable

Runnable 我們已經說過無數次了，無引數，無回傳值

@FunctionalInterface
public interface Runnable {
    public abstract void run();
}

Function

Function<T, R> 接受一個引數，并且有回傳值

@FunctionalInterface
public interface Function<T, R> {
    R apply(T t);
}

Consumer

Consumer 接受一個引數，沒有回傳值

@FunctionalInterface
public interface Consumer<T> {   
    void accept(T t);
}

Supplier

Supplier 沒有引數，有一個回傳值

@FunctionalInterface
public interface Supplier<T> {
    T get();
}

BiConsumer

BiConsumer<T, U> 接受兩個引數（Bi，英文單詞詞根，代表兩個的意思），沒有回傳值

@FunctionalInterface
public interface BiConsumer<T, U> {
    void accept(T t, U u);

好了，我們做個小匯總

有些同學可能有疑問，為什么要關注這幾個函式式介面，因為 CompletableFuture 的函式命名以及其作用都是和這幾個函式式介面高度相關的，一會你就會發現了

前戲做足，終于可以進入正題了 CompletableFuture

CompletableFuture

類結構

老規矩，先從類結構看起：

實作了 Future 介面

實作了 Future 介面，那就具有 Future 介面的相關特性，請腦補 Future 那少的可憐的 5 個方法，這里不再贅述，具體請查看不會用Java Future，我懷疑你泡茶沒我快

實作了 CompletionStage 介面

CompletionStage 這個介面還是挺陌生的，中文直譯過來是【竣工階段】，如果將燒水泡茶比喻成一項大工程，他們的竣工階段體現是不一樣的

單看執行緒1 或單看執行緒 2 就是一種串行關系，做完一步之后做下一步
一起看執行緒1 和執行緒 2，它們彼此就是并行關系，兩個執行緒做的事彼此獨立互補干擾
泡茶就是執行緒1 和執行緒 2 的匯總/組合，也就是執行緒 1 和執行緒 2 都完成之后才能到這個階段（當然也存在執行緒1 或執行緒 2 任意一個執行緒竣工就可以開啟下一階段的場景）

所以，CompletionStage 介面的作用就做了這點事，所有函式都用于描述任務的時序關系，總結起來就是這個樣子：

CompletableFuture 既然實作了兩個介面，自然也就會實作相應的方法充分利用其介面特性，我們走進它的方法來看一看

CompletableFuture 大約有50種不同處理串行，并行，組合以及處理錯誤的方法，小弟螢屏不爭氣，方法之多，一個螢屏裝不下，看到這么多方法，是不是瞬間要直接 收藏——>吃灰 2連走人？別擔心，我們按照相應的命名和作用進行分類，分分鐘搞定50多種方法

串行關系

then 直譯【然后】，也就是表示下一步，所以通常是一種串行關系體現, then 后面的單詞（比如 run /apply/accept）就是上面說的函式式介面中的抽象方法名稱了，它的作用和那幾個函式式介面的作用是一樣一樣滴

CompletableFuture<Void> thenRun(Runnable action)
CompletableFuture<Void> thenRunAsync(Runnable action)
CompletableFuture<Void> thenRunAsync(Runnable action, Executor executor)
  
<U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)
<U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn)
<U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)
  
CompletableFuture<Void> thenAccept(Consumer<? super T> action) 
CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action)
CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action, Executor executor)
  
<U> CompletableFuture<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn)  
<U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn)
<U> CompletableFuture<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn, Executor executor)

聚合 And 關系

combine... with... 和 both...and... 都是要求兩者都滿足，也就是 and 的關系了

<U,V> CompletableFuture<V> thenCombine(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
<U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
<U,V> CompletableFuture<V> thenCombineAsync(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn, Executor executor)

<U> CompletableFuture<Void> thenAcceptBoth(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action)
<U> CompletableFuture<Void> thenAcceptBothAsync(CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action)
<U> CompletableFuture<Void> thenAcceptBothAsync( CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action, Executor executor)
  
CompletableFuture<Void> runAfterBoth(CompletionStage<?> other, Runnable action)
CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action)
CompletableFuture<Void> runAfterBothAsync(CompletionStage<?> other, Runnable action, Executor executor)

聚合 Or 關系

Either...or... 表示兩者中的一個，自然也就是 Or 的體現了

<U> CompletableFuture<U> applyToEither(CompletionStage<? extends T> other, Function<? super T, U> fn)
<U> CompletableFuture<U> applyToEitherAsync(、CompletionStage<? extends T> other, Function<? super T, U> fn)
<U> CompletableFuture<U> applyToEitherAsync(CompletionStage<? extends T> other, Function<? super T, U> fn, Executor executor)

CompletableFuture<Void> acceptEither(CompletionStage<? extends T> other, Consumer<? super T> action)
CompletableFuture<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action)
CompletableFuture<Void> acceptEitherAsync(CompletionStage<? extends T> other, Consumer<? super T> action, Executor executor)

CompletableFuture<Void> runAfterEither(CompletionStage<?> other, Runnable action)
CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action)
CompletableFuture<Void> runAfterEitherAsync(CompletionStage<?> other, Runnable action, Executor executor)

例外處理

CompletableFuture<T> exceptionally(Function<Throwable, ? extends T> fn)
CompletableFuture<T> exceptionallyAsync(Function<Throwable, ? extends T> fn)
CompletableFuture<T> exceptionallyAsync(Function<Throwable, ? extends T> fn, Executor executor)
        
CompletableFuture<T> whenComplete(BiConsumer<? super T, ? super Throwable> action)
CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action)
CompletableFuture<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action, Executor executor)
        
       
<U> CompletableFuture<U> handle(BiFunction<? super T, Throwable, ? extends U> fn)
<U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn)
<U> CompletableFuture<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn, Executor executor)

這個例外處理看著還挺嚇人的，拿傳統的 try/catch/finally 做個對比也就瞬間秒懂了

whenComplete 和 handle 的區別如果你看接受的引數函式式介面名稱你也就能看出差別了，前者使用Comsumer, 自然也就不會有回傳值；后者使用 Function，自然也就會有回傳值

這里并沒有全部列舉，不過相信很多同學已經發現了規律：

CompletableFuture 提供的所有回呼方法都有兩個異步（Async）變體，都像這樣

// thenApply() 的變體
<U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn)
<U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn)
<U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor)

另外,方法的名稱也都與前戲中說的函式式介面完全匹配，按照這中規律分類之后，這 50 多個方法看起來是不是很輕松了呢？

基本方法已經羅列的差不多了，接下來我們通過一些例子來實際演示一下：

案例演示

創建一個 CompletableFuture 物件

創建一個 CompletableFuture 物件并沒有什么稀奇的，依舊是通過建構式構建

CompletableFuture<String> completableFuture = new CompletableFuture<String>();

這是最簡單的 CompletableFuture 物件創建方式，由于它實作了 Future 介面，所以自然就可以通過 get() 方法獲取結果

String result = completableFuture.get();

文章開頭已經說過，get()方法在任務結束之前將一直處在阻塞狀態，由于上面創建的 Future 沒有回傳，所以在這里呼叫 get() 將會永久性的堵塞

這時就需要我們呼叫 complete() 方法手動的結束一個 Future

completableFuture.complete("Future's Result Here Manually");

這時，所有等待這個 Future 的 client 都會回傳手動結束的指定結果

runAsync

使用 runAsync 進行異步計算

CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
    try {
        TimeUnit.SECONDS.sleep(3);
    } catch (InterruptedException e) {
        throw new IllegalStateException(e);
    }
    System.out.println("運行在一個單獨的執行緒當中");
});

future.get();

由于使用的是 Runnable 函式式運算式，自然也不會獲取到結果

supplyAsync

使用 runAsync 是沒有回傳結果的，我們想獲取異步計算的回傳結果需要使用 supplyAsync() 方法

		CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
			try {
				TimeUnit.SECONDS.sleep(3);
			} catch (InterruptedException e) {
				throw new IllegalStateException(e);
			}
			log.info("運行在一個單獨的執行緒當中");
			return "我有回傳值";
		});

		log.info(future.get());

由于使用的是 Supplier 函式式運算式，自然可以獲得回傳結果

我們已經多次說過，get() 方法在Future 計算完成之前會一直處在 blocking 狀態下，對于真正的異步處理，我們希望的是可以通過傳入回呼函式，在Future 結束時自動呼叫該回呼函式，這樣，我們就不用等待結果

CompletableFuture<String> comboText = CompletableFuture.supplyAsync(() -> {
  		//可以注釋掉做快速回傳 start
			try {
				TimeUnit.SECONDS.sleep(3);
			} catch (InterruptedException e) {
				throw new IllegalStateException(e);
			}
			log.info("??");
  		//可以注釋掉做快速回傳 end
			return "贊";
		})
				.thenApply(first -> {
					log.info("在看");
					return first + ", 在看";
				})
				.thenApply(second -> second + ", 轉發");

		log.info("三連有沒有？");
		log.info(comboText.get());

對 thenApply 的呼叫并沒有阻塞程式列印log，也就是前面說的通過回呼通知機制，這里你看到 thenApply 使用的是supplyAsync所用的執行緒，如果將supplyAsync 做快速回傳，我們再來看一下運行結果：

thenApply 此時使用的是主執行緒，所以：

串行的后續操作并不一定會和前序操作使用同一個執行緒

thenAccept

如果你不想從回呼函式中回傳任何結果，那可以使用 thenAccept

		final CompletableFuture<Void> voidCompletableFuture = CompletableFuture.supplyAsync(
				// 模擬遠端API呼叫，這里只回傳了一個構造的物件
				() -> Product.builder().id(12345L).name("頸椎/腰椎治療儀").build())
				.thenAccept(product -> {
					log.info("獲取到遠程API產品名稱 " + product.getName());
				});
		voidCompletableFuture.get();

thenRun

thenAccept 可以從回呼函式中獲取前序執行的結果，但thenRun 卻不可以，因為它的回呼函式式運算式定義中沒有任何引數

CompletableFuture.supplyAsync(() -> {
    //前序操作
}).thenRun(() -> {
    //串行的后需操作，無引數也無回傳值
});

我們前面同樣說過了，每個提供回呼方法的函式都有兩個異步（Async）變體，異步就是另外起一個執行緒

		CompletableFuture<String> stringCompletableFuture = CompletableFuture.supplyAsync(() -> {
			log.info("前序操作");
			return "前需操作結果";
		}).thenApplyAsync(result -> {
			log.info("后續操作");
			return "后續操作結果";
		});

到這里，相信你串行的操作你已經非常熟練了

thenCompose

日常的任務中，通常定義的方法都會回傳 CompletableFuture 型別，這樣會給后續操作留有更多的余地，假如有這樣的業務（X唄是不是都有這樣的業務呢？）：

//獲取用戶資訊詳情
	CompletableFuture<User> getUsersDetail(String userId) {
		return CompletableFuture.supplyAsync(() -> User.builder().id(12345L).name("日拱一兵").build());
	}

	//獲取用戶信用評級
	CompletableFuture<Double> getCreditRating(User user) {
		return CompletableFuture.supplyAsync(() -> CreditRating.builder().rating(7.5).build().getRating());
	}

這時，如果我們還是使用 thenApply() 方法來描述串行關系，回傳的結果就會發生 CompletableFuture 的嵌套

		CompletableFuture<CompletableFuture<Double>> result = completableFutureCompose.getUsersDetail(12345L)
				.thenApply(user -> completableFutureCompose.getCreditRating(user));

顯然這不是我們想要的，如果想“拍平” 回傳結果，thenCompose 方法就派上用場了

CompletableFuture<Double> result = completableFutureCompose.getUsersDetail(12345L)
				.thenCompose(user -> completableFutureCompose.getCreditRating(user));

這個和 Lambda 的map 和 flatMap 的道理是一樣一樣滴

thenCombine

如果要聚合兩個獨立 Future 的結果，那么 thenCombine 就會派上用場了

		CompletableFuture<Double> weightFuture = CompletableFuture.supplyAsync(() -> 65.0);
		CompletableFuture<Double> heightFuture = CompletableFuture.supplyAsync(() -> 183.8);
		
		CompletableFuture<Double> combinedFuture = weightFuture
				.thenCombine(heightFuture, (weight, height) -> {
					Double heightInMeter = height/100;
					return weight/(heightInMeter*heightInMeter);
				});

		log.info("身體BMI指標 - " + combinedFuture.get());

當然這里多數時處理兩個 Future 的關系，如果超過兩個Future，如何處理他們的一些聚合關系呢？

allOf ｜ anyOf

相信你看到方法的簽名，你已經明白他的用處了，這里就不再介紹了

static CompletableFuture<Void>	 allOf(CompletableFuture<?>... cfs)
static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs)

接下來就是例外的處理了

exceptionally

		Integer age = -1;

		CompletableFuture<String> maturityFuture = CompletableFuture.supplyAsync(() -> {
			if( age < 0 ) {
				throw new IllegalArgumentException("何方神圣？");
			}
			if(age > 18) {
				return "大家都是成年人";
			} else {
				return "未成年禁止入內";
			}
		}).thenApply((str) -> {
			log.info("游戲開始");
			return str;
		}).exceptionally(ex -> {
			log.info("必有蹊蹺，來者" + ex.getMessage());
			return "Unknown!";
		});

		log.info(maturityFuture.get());

exceptionally 就相當于 catch，出現例外，將會跳過 thenApply 的后續操作，直接捕獲例外，進行一場處理

handle

用多執行緒，良好的習慣是使用 try/finally 范式，handle 就可以起到 finally 的作用，對上述程式做一個小小的更改， handle 接受兩個引數，一個是正常回傳值，一個是例外

注意：handle的寫法也算是范式的一種

		Integer age = -1;

		CompletableFuture<String> maturityFuture = CompletableFuture.supplyAsync(() -> {
			if( age < 0 ) {
				throw new IllegalArgumentException("何方神圣？");
			}
			if(age > 18) {
				return "大家都是成年人";
			} else {
				return "未成年禁止入內";
			}
		}).thenApply((str) -> {
			log.info("游戲開始");
			return str;
		}).handle((res, ex) -> {
			if(ex != null) {
				log.info("必有蹊蹺，來者" + ex.getMessage());
				return "Unknown!";
			}
			return res;
		});

		log.info(maturityFuture.get());

到這里，關于 CompletableFuture 的基本使用你已經了解的差不多了，不知道你是否注意，我們前面說的帶有 Sync 的方法是單獨起一個執行緒來執行，但是我們并沒有創建執行緒，這是怎么實作的呢？

細心的朋友如果仔細看每個變種函式的第三個方法也許會發現里面都有一個 Executor 型別的引數，用于指定執行緒池，因為實際業務中我們是嚴謹手動創建執行緒的，這在我會手動創建執行緒，為什么要使用執行緒池?文章中明確說明過；如果沒有指定執行緒池，那自然就會有一個默認的執行緒池，也就是 ForkJoinPool

private static final Executor ASYNC_POOL = USE_COMMON_POOL ?
    ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();

ForkJoinPool 的執行緒數默認是 CPU 的核心數，但是，在前序文章中明確說明過：

不要所有業務共用一個執行緒池，因為，一旦有任務執行一些很慢的 I/O 操作，就會導致執行緒池中所有執行緒都阻塞在 I/O 操作上，從而造成執行緒饑餓，進而影響整個系統的性能

總結

CompletableFuture 的方法并沒有全部介紹完全，也沒必要全部介紹，相信大家按照這個思路來理解 CompletableFuture 也不會有什么大問題了，剩下的就交給實踐/時間以及自己的體會了

后記

你以為 JDK1.8 CompletableFuture 已經很完美了是不是，但追去完美的道路上永無止境，Java 9 對CompletableFuture 又做了部分升級和改造

添加了新的工廠方法
支持延遲和超時處理
```
orTimeout()
completeOnTimeout()
```
改進了對子類的支持

詳情可以查看： Java 9 CompletableFuture API Improvements. 怎樣快速的切換不同 Java 版本來嘗鮮？SDKMAN 統一靈活管理多版本Java 這篇文章的方法送給你

最后咱們再泡一壺茶，感受一下新變化吧

靈魂追問

聽說 ForkJoinPool 執行緒池效率更高，為什么呢？
如果批量處理異步程式，有什么可用的方案嗎？

參考

Java 并發編程實戰
Java 并發編程的藝術
Java 并發編程之美
https://www.baeldung.com/java-completablefuture
https://www.callicoder.com/java-8-completablefuture-tutorial/
個人博客：https://dayarch.top
加我微信好友, 進群娛樂學習交流，備注「進群」

歡迎持續關注公眾號：「日拱一兵」

前沿 Java 技術干貨分享

高效工具匯總 | 回復「工具」

面試問題分析與解答

技術資料領取 | 回復「資料」

以讀偵探小說思維輕松趣味學習 Java 技術堆疊相關知識，本著將復雜問題簡單化，抽象問題具體化和圖形化原則逐步分解技術問題，技術持續更新，請持續關注......

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/138746.html

標籤：Java

上一篇：Spring Boot + MyBatis 多模塊搭建教程

下一篇：Java 在PDF中添加騎縫章