Java 8 中的 CompletableFuture 太好用了!20 個示例全分享…
在 Java 中異步編程,不一定非要使用 rxJava, Java 本身的庫中的 CompletableFuture 可以很好的應對大部分的場景。
這篇文章介紹 Java 8 的 CompletionStage API 和它的標準庫的實現 CompletableFuture。API 通過例子的方式演示了它的行爲,每個例子演示一到兩個行爲。
既然CompletableFuture
類實現了CompletionStage
接口,首先我們需要理解這個接口的契約。它代表了一個特定的計算的階段,可以同步或者異步的被完成。你可以把它看成一個計算流水線上的一個單元,最終會產生一個最終結果,這意味着幾個CompletionStage
可以串聯起來,一個完成的階段可以觸發下一階段的執行,接着觸發下一次,接着……
除了實現CompletionStage
接口, CompletableFuture
也實現了future
接口, 代表一個未完成的異步事件。CompletableFuture
提供了方法,能夠顯式地完成這個 future, 所以它叫CompletableFuture
。
1、 創建一個完成的 CompletableFuture
最簡單的例子就是使用一個預定義的結果創建一個完成的 CompletableFuture, 通常我們會在計算的開始階段使用它。
static void completedFutureExample() {
CompletableFuture cf = CompletableFuture.completedFuture("message");
assertTrue(cf.isDone());
assertEquals("message", cf.getNow(null));
}
getNow(null)
方法在 future 完成的情況下會返回結果,就比如上面這個例子,否則返回 null (傳入的參數)。
2、運行一個簡單的異步階段
這個例子創建一個一個異步執行的階段:
static void runAsyncExample() {
CompletableFuture cf = CompletableFuture.runAsync(() -> {
assertTrue(Thread.currentThread().isDaemon());
randomSleep();
});
assertFalse(cf.isDone());
sleepEnough();
assertTrue(cf.isDone());
}
通過這個例子可以學到兩件事情:
CompletableFuture 的方法如果以Async
結尾,它會異步的執行 (沒有指定 executor 的情況下), 異步執行通過 ForkJoinPool
實現, 它使用守護線程去執行任務。注意這是 CompletableFuture 的特性, 其它 CompletionStage 可以 override 這個默認的行爲。
3、在前一個階段上應用函數
下面這個例子使用前面 #1 的完成的 CompletableFuture, #1 返回結果爲字符串message
, 然後應用一個函數把它變成大寫字母。
static void thenApplyExample() {
CompletableFuture cf = CompletableFuture.completedFuture("message").thenApply(s -> {
assertFalse(Thread.currentThread().isDaemon());
return s.toUpperCase();
});
assertEquals("MESSAGE", cf.getNow(null));
}
注意thenApply
方法名稱代表的行爲。
then
意味着這個階段的動作發生當前的階段正常完成之後。本例中,當前節點完成,返回字符串message
。
Apply
意味着返回的階段將會對結果前一階段的結果應用一個函數。
函數的執行會被阻塞,這意味着getNow()
只有打斜操作被完成後才返回。
另外,關注公衆號 Java 技術棧,在後臺回覆:面試,可以獲取我整理的 Java 併發多線程系列面試題和答案,非常齊全。
4、在前一個階段上異步應用函數
通過調用異步方法 (方法後邊加 Async 後綴),串聯起來的 CompletableFuture 可以異步地執行(使用 ForkJoinPool.commonPool())。
static void thenApplyAsyncExample() {
CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(s -> {
assertTrue(Thread.currentThread().isDaemon());
randomSleep();
return s.toUpperCase();
});
assertNull(cf.getNow(null));
assertEquals("MESSAGE", cf.join());
}
5、使用定製的 Executor 在前一個階段上異步應用函數
異步方法一個非常有用的特性就是能夠提供一個 Executor 來異步地執行 CompletableFuture。《線程池全面解析》推薦看下。
這個例子演示瞭如何使用一個固定大小的線程池來應用大寫函數。
static ExecutorService executor = Executors.newFixedThreadPool(3, new ThreadFactory() {
int count = 1;
@Override
public Thread newThread(Runnable runnable) {
return new Thread(runnable, "custom-executor-" + count++);
}
});
static void thenApplyAsyncWithExecutorExample() {
CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(s -> {
assertTrue(Thread.currentThread().getName().startsWith("custom-executor-"));
assertFalse(Thread.currentThread().isDaemon());
randomSleep();
return s.toUpperCase();
}, executor);
assertNull(cf.getNow(null));
assertEquals("MESSAGE", cf.join());
}
6、消費前一階段的結果
如果下一階段接收了當前階段的結果,但是在計算的時候不需要返回值 (它的返回類型是 void), 那麼它可以不應用一個函數,而是一個消費者, 調用方法也變成了thenAccept
:
static void thenAcceptExample() {
StringBuilder result = new StringBuilder();
CompletableFuture.completedFuture("thenAccept message")
.thenAccept(s -> result.append(s));
assertTrue("Result was empty", result.length() > 0);
}
本例中消費者同步地執行,所以我們不需要在 CompletableFuture 調用join
方法。
7、異步地消費遷移階段的結果
同樣,可以使用thenAcceptAsync
方法, 串聯的 CompletableFuture 可以異步地執行。
static void thenAcceptAsyncExample() {
StringBuilder result = new StringBuilder();
CompletableFuture cf = CompletableFuture.completedFuture("thenAcceptAsync message")
.thenAcceptAsync(s -> result.append(s));
cf.join();
assertTrue("Result was empty", result.length() > 0);
}
8、完成計算異常
現在我們來看一下異步操作如何顯式地返回異常,用來指示計算失敗。我們簡化這個例子,操作處理一個字符串,把它轉換成答謝,我們模擬延遲一秒。
我們使用thenApplyAsync(Function, Executor)
方法,第一個參數傳入大寫函數, executor 是一個 delayed executor, 在執行前會延遲一秒。
static void completeExceptionallyExample() {
CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(String::toUpperCase,
CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS));
CompletableFuture exceptionHandler = cf.handle((s, th) -> { return (th != null) ? "message upon cancel" : ""; });
cf.completeExceptionally(new RuntimeException("completed exceptionally"));
assertTrue("Was not completed exceptionally", cf.isCompletedExceptionally());
try {
cf.join();
fail("Should have thrown an exception");
} catch(CompletionException ex) { // just for testing
assertEquals("completed exceptionally", ex.getCause().getMessage());
}
assertEquals("message upon cancel", exceptionHandler.join());
}
讓我們看一下細節。
首先我們創建了一個 CompletableFuture, 完成後返回一個字符串message
, 接着我們調用thenApplyAsync
方法,它返回一個 CompletableFuture。這個方法在第一個函數完成後,異步地應用轉大寫字母函數。
這個例子還演示瞭如何通過delayedExecutor(timeout, timeUnit)
延遲執行一個異步任務。
我們創建了一個分離的handler
階段:exceptionHandler, 它處理異常異常,在異常情況下返回message upon cancel
。
下一步我們顯式地用異常完成第二個階段。在階段上調用join
方法,它會執行大寫轉換,然後拋出 CompletionException(正常的 join 會等待 1 秒,然後得到大寫的字符串。不過我們的例子還沒等它執行就完成了異常), 然後它觸發了 handler 階段。
9、取消計算
和完成異常類似,我們可以調用cancel(boolean mayInterruptIfRunning)
取消計算。對於 CompletableFuture 類,布爾參數並沒有被使用,這是因爲它並沒有使用中斷去取消操作,相反,cancel
等價於completeExceptionally(new CancellationException())
。
static void cancelExample() {
CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(String::toUpperCase,
CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS));
CompletableFuture cf2 = cf.exceptionally(throwable -> "canceled message");
assertTrue("Was not canceled", cf.cancel(true));
assertTrue("Was not completed exceptionally", cf.isCompletedExceptionally());
assertEquals("canceled message", cf2.join());
}
10、在兩個完成的階段其中之一上應用函數
下面的例子創建了CompletableFuture
, applyToEither
處理兩個階段, 在其中之一上應用函數 (包保證哪一個被執行)。本例中的兩個階段一個是應用大寫轉換在原始的字符串上, 另一個階段是應用小些轉換。
static void applyToEitherExample() {
String original = "Message";
CompletableFuture cf1 = CompletableFuture.completedFuture(original)
.thenApplyAsync(s -> delayedUpperCase(s));
CompletableFuture cf2 = cf1.applyToEither(
CompletableFuture.completedFuture(original).thenApplyAsync(s -> delayedLowerCase(s)),
s -> s + " from applyToEither");
assertTrue(cf2.join().endsWith(" from applyToEither"));
}
11、在兩個完成的階段其中之一上調用消費函數
和前一個例子很類似了,只不過我們調用的是消費者函數 (Function 變成 Consumer):
static void acceptEitherExample() {
String original = "Message";
StringBuilder result = new StringBuilder();
CompletableFuture cf = CompletableFuture.completedFuture(original)
.thenApplyAsync(s -> delayedUpperCase(s))
.acceptEither(CompletableFuture.completedFuture(original).thenApplyAsync(s -> delayedLowerCase(s)),
s -> result.append(s).append("acceptEither"));
cf.join();
assertTrue("Result was empty", result.toString().endsWith("acceptEither"));
}
12、在兩個階段都執行完後運行一個 Runnable
這個例子演示了依賴的 CompletableFuture 如果等待兩個階段完成後執行了一個 Runnable。
注意下面所有的階段都是同步執行的,第一個階段執行大寫轉換,第二個階段執行小寫轉換。
static void runAfterBothExample() {
String original = "Message";
StringBuilder result = new StringBuilder();
CompletableFuture.completedFuture(original).thenApply(String::toUpperCase).runAfterBoth(
CompletableFuture.completedFuture(original).thenApply(String::toLowerCase),
() -> result.append("done"));
assertTrue("Result was empty", result.length() > 0);
}
13、 使用 BiConsumer 處理兩個階段的結果
上面的例子還可以通過 BiConsumer 來實現:
static void thenAcceptBothExample() {
String original = "Message";
StringBuilder result = new StringBuilder();
CompletableFuture.completedFuture(original).thenApply(String::toUpperCase).thenAcceptBoth(
CompletableFuture.completedFuture(original).thenApply(String::toLowerCase),
(s1, s2) -> result.append(s1 + s2));
assertEquals("MESSAGEmessage", result.toString());
}
14、使用 BiFunction 處理兩個階段的結果
如果 CompletableFuture 依賴兩個前面階段的結果, 它複合兩個階段的結果再返回一個結果,我們就可以使用thenCombine()
函數。整個流水線是同步的,所以getNow()
會得到最終的結果,它把大寫和小寫字符串連接起來。
static void thenCombineExample() {
String original = "Message";
CompletableFuture cf = CompletableFuture.completedFuture(original).thenApply(s -> delayedUpperCase(s))
.thenCombine(CompletableFuture.completedFuture(original).thenApply(s -> delayedLowerCase(s)),
(s1, s2) -> s1 + s2);
assertEquals("MESSAGEmessage", cf.getNow(null));
}
15、異步使用 BiFunction 處理兩個階段的結果
類似上面的例子,但是有一點不同:依賴的前兩個階段異步地執行,所以thenCombine()
也異步地執行,即時它沒有Async
後綴。
Javadoc 中有註釋:
Actions supplied for dependent completions of non-async methods may be performed by the thread that completes the current CompletableFuture, or by any other caller of a completion method
所以我們需要join
方法等待結果的完成。
static void thenCombineAsyncExample() {
String original = "Message";
CompletableFuture cf = CompletableFuture.completedFuture(original)
.thenApplyAsync(s -> delayedUpperCase(s))
.thenCombine(CompletableFuture.completedFuture(original).thenApplyAsync(s -> delayedLowerCase(s)),
(s1, s2) -> s1 + s2);
assertEquals("MESSAGEmessage", cf.join());
}
16、組合 CompletableFuture
我們可以使用thenCompose()
完成上面兩個例子。這個方法等待第一個階段的完成 (大寫轉換), 它的結果傳給一個指定的返回 CompletableFuture 函數,它的結果就是返回的 CompletableFuture 的結果。
static void thenComposeExample() {
String original = "Message";
CompletableFuture cf = CompletableFuture.completedFuture(original).thenApply(s -> delayedUpperCase(s))
.thenCompose(upper -> CompletableFuture.completedFuture(original).thenApply(s -> delayedLowerCase(s))
.thenApply(s -> upper + s));
assertEquals("MESSAGEmessage", cf.join());
}
17、當幾個階段中的一個完成,創建一個完成的階段
下面的例子演示了當任意一個 CompletableFuture 完成後, 創建一個完成的 CompletableFuture.
待處理的階段首先創建, 每個階段都是轉換一個字符串爲大寫。因爲本例中這些階段都是同步地執行 (thenApply), 從anyOf
中創建的 CompletableFuture 會立即完成,這樣所有的階段都已完成,我們使用whenComplete(BiConsumer<? super Object, ? super Throwable> action)
處理完成的結果。
static void anyOfExample() {
StringBuilder result = new StringBuilder();
List messages = Arrays.asList("a", "b", "c");
List<CompletableFuture> futures = messages.stream()
.map(msg -> CompletableFuture.completedFuture(msg).thenApply(s -> delayedUpperCase(s)))
.collect(Collectors.toList());
CompletableFuture.anyOf(futures.toArray(new CompletableFuture[futures.size()])).whenComplete((res, th) -> {
if(th == null) {
assertTrue(isUpperCase((String) res));
result.append(res);
}
});
assertTrue("Result was empty", result.length() > 0);
}
18、當所有的階段都完成後創建一個階段
上一個例子是當任意一個階段完成後接着處理,接下來的兩個例子演示當所有的階段完成後才繼續處理, 同步地方式和異步地方式兩種。
static void allOfExample() {
StringBuilder result = new StringBuilder();
List messages = Arrays.asList("a", "b", "c");
List<CompletableFuture> futures = messages.stream()
.map(msg -> CompletableFuture.completedFuture(msg).thenApply(s -> delayedUpperCase(s)))
.collect(Collectors.toList());
CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()])).whenComplete((v, th) -> {
futures.forEach(cf -> assertTrue(isUpperCase(cf.getNow(null))));
result.append("done");
});
assertTrue("Result was empty", result.length() > 0);
}
19、當所有的階段都完成後異步地創建一個階段
使用thenApplyAsync()
替換那些單個的 CompletableFutures 的方法,allOf()
會在通用池中的線程中異步地執行。所以我們需要調用join
方法等待它完成。
static void allOfAsyncExample() {
StringBuilder result = new StringBuilder();
List messages = Arrays.asList("a", "b", "c");
List<CompletableFuture> futures = messages.stream()
.map(msg -> CompletableFuture.completedFuture(msg).thenApplyAsync(s -> delayedUpperCase(s)))
.collect(Collectors.toList());
CompletableFuture allOf = CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]))
.whenComplete((v, th) -> {
futures.forEach(cf -> assertTrue(isUpperCase(cf.getNow(null))));
result.append("done");
});
allOf.join();
assertTrue("Result was empty", result.length() > 0);
}
20、真實的例子
現在你已經瞭解了 CompletionStage 和 CompletableFuture 的一些函數的功能,下面的例子是一個實踐場景:
-
首先異步調用
cars
方法獲得 Car 的列表,它返回 CompletionStage 場景。cars
消費一個遠程的 REST API。 -
然後我們複合一個 CompletionStage 填寫每個汽車的評分,通過
rating(manufacturerId)
返回一個 CompletionStage, 它會異步地獲取汽車的評分 (可能又是一個 REST API 調用) -
當所有的汽車填好評分後,我們結束這個列表,所以我們調用
allOf
得到最終的階段, 它在前面階段所有階段完成後才完成。 -
在最終的階段調用
whenComplete()
, 我們打印出每個汽車和它的評分。
cars().thenCompose(cars -> {
List<CompletionStage> updatedCars = cars.stream()
.map(car -> rating(car.manufacturerId).thenApply(r -> {
car.setRating(r);
return car;
})).collect(Collectors.toList());
CompletableFuture done = CompletableFuture
.allOf(updatedCars.toArray(new CompletableFuture[updatedCars.size()]));
return done.thenApply(v -> updatedCars.stream().map(CompletionStage::toCompletableFuture)
.map(CompletableFuture::join).collect(Collectors.toList()));
}).whenComplete((cars, th) -> {
if (th == null) {
cars.forEach(System.out::println);
} else {
throw new RuntimeException(th);
}
}).toCompletableFuture().join();
因爲每個汽車的實例都是獨立的,得到每個汽車的評分都可以異步地執行,這會提高系統的性能 (延遲),而且,等待所有的汽車評分被處理使用的是allOf
方法,而不是手工的線程等待 (Thread#join()
或 a CountDownLatch
)。
原文:https://mahmoudanouti.wordpress.com/2018/01/26/20-examples-of-using-javas-completablefuture/
來源:https://colobu.com/2018/03/12/20-Examples-of-Using-Java%E2%80%99s-CompletableFuture/
參考文檔:
https://medium.com/@johnmcclean/reactive-programming-with-java-8-and-simple-react-the-tutorial-3634f512eeb1
http://javaday.org.ua/completablefuture-overview/
https://blog.takipi.com/back-to-the-completablefuture-java-8-feature-highlight/
https://github.com/spotify/completable-futures
本文由 Readfog 進行 AMP 轉碼,版權歸原作者所有。
來源:https://mp.weixin.qq.com/s/GHmc2JbfPfkBNnvs5XaemQ