Java 8 中的 CompletableFuture 太好用了!20 個(gè)示例全分享…
在Java中異步編程,不一定非要使用rxJava, Java本身的庫(kù)中的CompletableFuture可以很好的應(yīng)對(duì)大部分的場(chǎng)景。
除了實(shí)現(xiàn)CompletionStage接口, CompletableFuture也實(shí)現(xiàn)了future接口, 代表一個(gè)未完成的異步事件。CompletableFuture提供了方法,能夠顯式地完成這個(gè)future,所以它叫CompletableFuture。
最簡(jiǎn)單的例子就是使用一個(gè)預(yù)定義的結(jié)果創(chuàng)建一個(gè)完成的CompletableFuture,通常我們會(huì)在計(jì)算的開始階段使用它。
static void completedFutureExample() {
CompletableFuture cf = CompletableFuture.completedFuture("message");
assertTrue(cf.isDone());
assertEquals("message", cf.getNow(null));
}
getNow(null)方法在future完成的情況下會(huì)返回結(jié)果,就比如上面這個(gè)例子,否則返回null (傳入的參數(shù))。
這個(gè)例子創(chuàng)建一個(gè)一個(gè)異步執(zhí)行的階段:
static void runAsyncExample() {
CompletableFuture cf = CompletableFuture.runAsync(() -> {
assertTrue(Thread.currentThread().isDaemon());
randomSleep();
});
assertFalse(cf.isDone());
sleepEnough();
assertTrue(cf.isDone());
}
Async結(jié)尾,它會(huì)異步的執(zhí)行(沒(méi)有指定executor的情況下), 異步執(zhí)行通過(guò)ForkJoinPool實(shí)現(xiàn), 它使用守護(hù)線程去執(zhí)行任務(wù)。注意這是CompletableFuture的特性, 其它CompletionStage可以override這個(gè)默認(rèn)的行為。3、在前一個(gè)階段上應(yīng)用函數(shù)
下面這個(gè)例子使用前面 #1 的完成的CompletableFuture, #1返回結(jié)果為字符串message,然后應(yīng)用一個(gè)函數(shù)把它變成大寫字母。
static void thenApplyExample() {
CompletableFuture cf = CompletableFuture.completedFuture("message").thenApply(s -> {
assertFalse(Thread.currentThread().isDaemon());
return s.toUpperCase();
});
assertEquals("MESSAGE", cf.getNow(null));
}
注意thenApply方法名稱代表的行為。
then意味著這個(gè)階段的動(dòng)作發(fā)生當(dāng)前的階段正常完成之后。本例中,當(dāng)前節(jié)點(diǎn)完成,返回字符串message。
Apply意味著返回的階段將會(huì)對(duì)結(jié)果前一階段的結(jié)果應(yīng)用一個(gè)函數(shù)。
函數(shù)的執(zhí)行會(huì)被阻塞,這意味著getNow()只有打斜操作被完成后才返回。
另外,關(guān)注公眾號(hào)互聯(lián)網(wǎng)架構(gòu)師,在后臺(tái)回復(fù):2T,可以獲取我整理的 Java 并發(fā)多線程系列面試題和答案,非常齊全。
通過(guò)調(diào)用異步方法(方法后邊加Async后綴),串聯(lián)起來(lái)的CompletableFuture可以異步地執(zhí)行(使用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());
}
異步方法一個(gè)非常有用的特性就是能夠提供一個(gè)Executor來(lái)異步地執(zhí)行CompletableFuture。
這個(gè)例子演示了如何使用一個(gè)固定大小的線程池來(lái)應(yīng)用大寫函數(shù)。
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());
}
如果下一階段接收了當(dāng)前階段的結(jié)果,但是在計(jì)算的時(shí)候不需要返回值(它的返回類型是void), 那么它可以不應(yīng)用一個(gè)函數(shù),而是一個(gè)消費(fèi)者, 調(diào)用方法也變成了thenAccept:
static void thenAcceptExample() {
StringBuilder result = new StringBuilder();
CompletableFuture.completedFuture("thenAccept message")
.thenAccept(s -> result.append(s));
assertTrue("Result was empty", result.length() > 0);
}
本例中消費(fèi)者同步地執(zhí)行,所以我們不需要在CompletableFuture調(diào)用join方法。
同樣,可以使用thenAcceptAsync方法, 串聯(lián)的CompletableFuture可以異步地執(zhí)行。
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);
}
現(xiàn)在我們來(lái)看一下異步操作如何顯式地返回異常,用來(lái)指示計(jì)算失敗。我們簡(jiǎn)化這個(gè)例子,操作處理一個(gè)字符串,把它轉(zhuǎn)換成答謝,我們模擬延遲一秒。
我們使用thenApplyAsync(Function, Executor)方法,第一個(gè)參數(shù)傳入大寫函數(shù), executor是一個(gè)delayed executor,在執(zhí)行前會(huì)延遲一秒。
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());
}
讓我們看一下細(xì)節(jié)。
首先我們創(chuàng)建了一個(gè)CompletableFuture, 完成后返回一個(gè)字符串message,接著我們調(diào)用thenApplyAsync方法,它返回一個(gè)CompletableFuture。這個(gè)方法在第一個(gè)函數(shù)完成后,異步地應(yīng)用轉(zhuǎn)大寫字母函數(shù)。
這個(gè)例子還演示了如何通過(guò)delayedExecutor(timeout, timeUnit)延遲執(zhí)行一個(gè)異步任務(wù)。
我們創(chuàng)建了一個(gè)分離的handler階段:exceptionHandler, 它處理異常異常,在異常情況下返回message upon cancel。
下一步我們顯式地用異常完成第二個(gè)階段。在階段上調(diào)用join方法,它會(huì)執(zhí)行大寫轉(zhuǎn)換,然后拋出CompletionException(正常的join會(huì)等待1秒,然后得到大寫的字符串。不過(guò)我們的例子還沒(méi)等它執(zhí)行就完成了異常), 然后它觸發(fā)了handler階段。
和完成異常類似,我們可以調(diào)用cancel(boolean mayInterruptIfRunning)取消計(jì)算。對(duì)于CompletableFuture類,布爾參數(shù)并沒(méi)有被使用,這是因?yàn)樗](méi)有使用中斷去取消操作,相反,cancel等價(jià)于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());
}
下面的例子創(chuàng)建了CompletableFuture, applyToEither處理兩個(gè)階段, 在其中之一上應(yīng)用函數(shù)(包保證哪一個(gè)被執(zhí)行)。本例中的兩個(gè)階段一個(gè)是應(yīng)用大寫轉(zhuǎn)換在原始的字符串上, 另一個(gè)階段是應(yīng)用小些轉(zhuǎn)換。
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"));
}
和前一個(gè)例子很類似了,只不過(guò)我們調(diào)用的是消費(fèi)者函數(shù) (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"));
}
Runnable這個(gè)例子演示了依賴的CompletableFuture如果等待兩個(gè)階段完成后執(zhí)行了一個(gè)Runnable。
注意下面所有的階段都是同步執(zhí)行的,第一個(gè)階段執(zhí)行大寫轉(zhuǎn)換,第二個(gè)階段執(zhí)行小寫轉(zhuǎn)換。
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);
}
上面的例子還可以通過(guò)BiConsumer來(lái)實(shí)現(xiàn):
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());
}
如果CompletableFuture依賴兩個(gè)前面階段的結(jié)果, 它復(fù)合兩個(gè)階段的結(jié)果再返回一個(gè)結(jié)果,我們就可以使用thenCombine()函數(shù)。整個(gè)流水線是同步的,所以getNow()會(huì)得到最終的結(jié)果,它把大寫和小寫字符串連接起來(lái)。
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));
}
類似上面的例子,但是有一點(diǎn)不同:依賴的前兩個(gè)階段異步地執(zhí)行,所以thenCombine()也異步地執(zhí)行,即時(shí)它沒(méi)有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方法等待結(jié)果的完成。
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());
}
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());
}
下面的例子演示了當(dāng)任意一個(gè)CompletableFuture完成后, 創(chuàng)建一個(gè)完成的CompletableFuture.
待處理的階段首先創(chuàng)建, 每個(gè)階段都是轉(zhuǎn)換一個(gè)字符串為大寫。因?yàn)楸纠羞@些階段都是同步地執(zhí)行(thenApply), 從anyOf中創(chuàng)建的CompletableFuture會(huì)立即完成,這樣所有的階段都已完成,我們使用whenComplete(BiConsumer<? super Object, ? super Throwable> action)處理完成的結(jié)果。
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);
}
上一個(gè)例子是當(dāng)任意一個(gè)階段完成后接著處理,接下來(lái)的兩個(gè)例子演示當(dāng)所有的階段完成后才繼續(xù)處理, 同步地方式和異步地方式兩種。
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);
}
使用thenApplyAsync()替換那些單個(gè)的CompletableFutures的方法,allOf()會(huì)在通用池中的線程中異步地執(zhí)行。所以我們需要調(diào)用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);
}
現(xiàn)在你已經(jīng)了解了CompletionStage 和 CompletableFuture 的一些函數(shù)的功能,下面的例子是一個(gè)實(shí)踐場(chǎng)景:
首先異步調(diào)用 cars方法獲得Car的列表,它返回CompletionStage場(chǎng)景。cars消費(fèi)一個(gè)遠(yuǎn)程的REST API。然后我們復(fù)合一個(gè)CompletionStage填寫每個(gè)汽車的評(píng)分,通過(guò) rating(manufacturerId)返回一個(gè)CompletionStage, 它會(huì)異步地獲取汽車的評(píng)分(可能又是一個(gè)REST API調(diào)用)當(dāng)所有的汽車填好評(píng)分后,我們結(jié)束這個(gè)列表,所以我們調(diào)用 allOf得到最終的階段, 它在前面階段所有階段完成后才完成。在最終的階段調(diào)用
whenComplete(),我們打印出每個(gè)汽車和它的評(píng)分。
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();
最后,關(guān)注公眾號(hào)互聯(lián)網(wǎng)架構(gòu)師,在后臺(tái)回復(fù):2T,可以獲取我整理的 Java 并發(fā)多線程系列面試題和答案,非常齊全。
原文:https://mahmoudanouti.wordpress.com/2018/01/26/20-examples-of-using-javas-completablefuture/
來(lái)源: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
END
順便給大家推薦一個(gè)GitHub項(xiàng)目,這個(gè) GitHub 整理了上千本常用技術(shù)PDF,絕大部分核心的技術(shù)書籍都可以在這里找到,
GitHub地址:https://github.com/javadevbooks/books
Gitee地址:https://gitee.com/javadevbooks/books
電子書已經(jīng)更新好了,你們需要的可以自行下載了,記得點(diǎn)一個(gè)star,持續(xù)更新中..

