摘要：這個例子想要說明兩個事情中以為結尾的方法將會異步執行默認情況下即指沒有傳入的情況下，異步執行會使用實現，該線程池使用一個后臺線程來執行任務。這個例子展示了如何使用一個固定大小的線程池來實現大寫操作。
前言
這篇博客回顧JAVA8的CompletionStageAPI以及其在JAVA庫中的標準實現CompletableFuture。將會通過幾個例子來展示API的各種行為。
因為CompletableFuture是CompletionInterface接口的實現，所以我們首先要了解該接口的契約。它代表某個同步或異步計算的一個階段。你可以把它理解為是一個為了產生有價值最終結果的計算的流水線上的一個單元。這意味著多個ComletionStage指令可以鏈接起來從而一個階段的完成可以觸發下一個階段的執行。
除了實現了CompletionStage接口，Completion還繼承了Future，這個接口用于實現一個未開始的異步事件。因為能夠顯式的完成Future，所以取名為CompletableFuture。
1.新建一個完成的CompletableFuture
這個簡單的示例中創建了一個已經完成的預先設置好結果的CompletableFuture。通常作為計算的起點階段。
static void completedFutureExample() {
CompletableFuture cf = CompletableFuture.completedFuture("message");
assertTrue(cf.isDone());
assertEquals("message", cf.getNow(null));
}
getNow方法會返回完成后的結果（這里就是message），如果還未完成，則返回傳入的默認值null。
2.運行一個簡單的異步stage
下面的例子解釋了如何創建一個異步運行Runnable的stage。
static void runAsyncExample() {
CompletableFuture cf = CompletableFuture.runAsync(() -> {
assertTrue(Thread.currentThread().isDaemon());
randomSleep();
});
assertFalse(cf.isDone());
sleepEnough();
assertTrue(cf.isDone());
}
這個例子想要說明兩個事情：
CompletableFuture中以Async為結尾的方法將會異步執行
默認情況下（即指沒有傳入Executor的情況下），異步執行會使用ForkJoinPool實現，該線程池使用一個后臺線程來執行Runnable任務。注意這只是特定于CompletableFuture實現，其它的CompletableStage實現可以重寫該默認行為。
3.將方法作用于前一個Stage
下面的例子引用了第一個例子中已經完成的CompletableFuture，它將引用生成的字符串結果并將該字符串大寫。
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是指將一個Function作用于之前階段得出的結果
Function是阻塞的，這意味著只有當大寫操作執行完成之后才會執行getNow()方法。
4.異步的的將方法作用于前一個Stage
通過在方法后面添加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());
}
使用一個自定義的Executor來異步執行該方法
異步方法的一個好處是可以提供一個Executor來執行CompletableStage。這個例子展示了如何使用一個固定大小的線程池來實現大寫操作。
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.消費(Consume)前一個Stage的結果
如果下一個Stage接收了當前Stage的結果但是在計算中無需返回值（比如其返回值為void），那么它將使用方法thenAccept并傳入一個Consumer接口。
static void thenAcceptExample() {
StringBuilder result = new StringBuilder();
CompletableFuture.completedFuture("thenAccept message")
.thenAccept(s -> result.append(s));
assertTrue("Result was empty", result.length() > 0);
}
Consumer將會同步執行，所以我們無需在返回的CompletableFuture上執行join操作。
7.異步執行Comsume
同樣，使用Asyn后綴實現：
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.計算出現異常時
我們現在來模擬一個出現異常的場景。為了簡潔性，我們還是將一個字符串大寫，但是我們會模擬延時進行該操作。我們會使用thenApplyAsyn(Function, Executor)，第一個參數是大寫轉化方法，第二個參數是一個延時executor，它會延時一秒鐘再將操作提交給ForkJoinPool。
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());
}
首先，我們新建了一個已經完成并帶有返回值message的CompletableFuture對象。然后我們調用thenApplyAsync方法，該方法會返回一個新的CompletableFuture。這個方法用異步的方式執行大寫操作。這里還展示了如何使用delayedExecutor(timeout, timeUnit)方法來延時異步操作。
然后我們創建了一個handler stage，exceptionHandler，這個階段會處理一切異常并返回另一個消息message upon cancel。
最后，我們顯式的完成第二個階段并拋出異常，它會導致進行大寫操作的階段拋出CompletionException。它還會觸發handler階段。
API補充：
CompletableFuture handle(BiFunction fn)
返回一個新的CompletionStage，無論之前的Stage是否正常運行完畢。傳入的參數包括上一個階段的結果和拋出異常。
9.取消計算
和計算時異常處理很相似，我們可以通過Future接口中的cancel(boolean mayInterruptIfRunning)來取消計算。
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());
}
API補充
public CompletableFuture exceptionally(Function fn)
返回一個新的CompletableFuture，如果出現異常，則為該方法中執行的結果，否則就是正常執行的結果。
10.將Function作用于兩個已完成Stage的結果之一
下面的例子創建了一個CompletableFuture對象并將Function作用于已完成的兩個Stage中的任意一個（沒有保證哪一個將會傳遞給Function）。這兩個階段分別如下：一個將字符串大寫，另一個小寫。
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"));
}
public CompletableFuture applyToEitherAsync(CompletionStage other,Function fn)
返回一個全新的CompletableFuture，包含著this或是other操作完成之后，在二者中的任意一個執行fn
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
注意這里的兩個Stage都是同步運行的，第一個stage將字符串轉化為大寫之后，第二個stage將其轉化為小寫。
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接收兩個stage的結果
Biconsumer支持同時對兩個Stage的結果進行操作。
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后綴。
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.Compose CompletableFuture
我們可以使用thenCompose來完成前兩個例子中的操作。
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.當多個階段中有有何一個完成，即新建一個完成階段
static void anyOfExample() {
StringBuilder result = new StringBuilder();
List messages = Arrays.asList("a", "b", "c");
List 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 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.當所有階段完成以后，新建一個異步完成階段
static void allOfAsyncExample() {
StringBuilder result = new StringBuilder();
List messages = Arrays.asList("a", "b", "c");
List 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.真實場景
下面展示了一個實踐CompletableFuture的場景：
先通過調用cars()方法異步獲得Car列表。它將會返回一個CompletionStage>。cars()方法應當使用一個遠程的REST端點來實現。
我們將該Stage和另一個Stage組合，另一個Stage會通過調用rating(manufactureId)來異步獲取每輛車的評分。
當所有的Car對象都填入評分后，我們調用allOf()來進入最終Stage，它將在這兩個階段完成后執行
在最終Stage上使用whenComplete()，打印出車輛的評分。
cars().thenCompose(cars -> {
List 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();
參考資料
Java CompletableFuture 詳解
Guide To CompletableFuture
想要了解更多開發技術，面試教程以及互聯網公司內推，歡迎關注我的微信公眾號！將會不定期的發放福利哦~
