資訊專欄INFORMATION COLUMN
摘要：這個例子想要說明兩個事情中以為結(jié)尾的方法將會異步執(zhí)行默認情況下即指沒有傳入的情況下，異步執(zhí)行會使用實現(xiàn)，該線程池使用一個后臺線程來執(zhí)行任務(wù)。這個例子展示了如何使用一個固定大小的線程池來實現(xiàn)大寫操作。
前言
這篇博客回顧JAVA8的CompletionStageAPI以及其在JAVA庫中的標(biāo)準實現(xiàn)CompletableFuture。將會通過幾個例子來展示API的各種行為。
因為CompletableFuture是CompletionInterface接口的實現(xiàn)，所以我們首先要了解該接口的契約。它代表某個同步或異步計算的一個階段。你可以把它理解為是一個為了產(chǎn)生有價值最終結(jié)果的計算的流水線上的一個單元。這意味著多個ComletionStage指令可以鏈接起來從而一個階段的完成可以觸發(fā)下一個階段的執(zhí)行。
除了實現(xiàn)了CompletionStage接口，Completion還繼承了Future，這個接口用于實現(xiàn)一個未開始的異步事件。因為能夠顯式的完成Future，所以取名為CompletableFuture。
1.新建一個完成的CompletableFuture這個簡單的示例中創(chuàng)建了一個已經(jīng)完成的預(yù)先設(shè)置好結(jié)果的CompletableFuture。通常作為計算的起點階段。
static void completedFutureExample() {
CompletableFuture cf = CompletableFuture.completedFuture("message");
assertTrue(cf.isDone());
assertEquals("message", cf.getNow(null));
}
getNow方法會返回完成后的結(jié)果（這里就是message），如果還未完成，則返回傳入的默認值null。
2.運行一個簡單的異步stage下面的例子解釋了如何創(chuàng)建一個異步運行Runnable的stage。
static void runAsyncExample() {
CompletableFuture cf = CompletableFuture.runAsync(() -> {
assertTrue(Thread.currentThread().isDaemon());
randomSleep();
});
assertFalse(cf.isDone());
sleepEnough();
assertTrue(cf.isDone());
}
這個例子想要說明兩個事情：
CompletableFuture中以Async為結(jié)尾的方法將會異步執(zhí)行
默認情況下（即指沒有傳入Executor的情況下），異步執(zhí)行會使用ForkJoinPool實現(xiàn)，該線程池使用一個后臺線程來執(zhí)行Runnable任務(wù)。注意這只是特定于CompletableFuture實現(xiàn)，其它的CompletableStage實現(xiàn)可以重寫該默認行為。
3.將方法作用于前一個Stage下面的例子引用了第一個例子中已經(jīng)完成的CompletableFuture，它將引用生成的字符串結(jié)果并將該字符串大寫。
static void thenApplyExample() {
CompletableFuture cf = CompletableFuture.completedFuture("message").thenApply(s -> {
assertFalse(Thread.currentThread().isDaemon());
return s.toUpperCase();
});
assertEquals("MESSAGE", cf.getNow(null));
}
這里的關(guān)鍵詞是thenApply：
then是指在當(dāng)前階段正常執(zhí)行完成后（正常執(zhí)行是指沒有拋出異常）進行的操作。在本例中，當(dāng)前階段已經(jīng)完成并得到值message。
Apply是指將一個Function作用于之前階段得出的結(jié)果
Function是阻塞的，這意味著只有當(dāng)大寫操作執(zhí)行完成之后才會執(zhí)行getNow()方法。
4.異步的的將方法作用于前一個Stage通過在方法后面添加Async后綴，該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());
}
使用一個自定義的Executor來異步執(zhí)行該方法
異步方法的一個好處是可以提供一個Executor來執(zhí)行CompletableStage。這個例子展示了如何使用一個固定大小的線程池來實現(xiàn)大寫操作。
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的結(jié)果
如果下一個Stage接收了當(dāng)前Stage的結(jié)果但是在計算中無需返回值（比如其返回值為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將會同步執(zhí)行，所以我們無需在返回的CompletableFuture上執(zhí)行join操作。
7.異步執(zhí)行Comsume同樣，使用Asyn后綴實現(xiàn)：
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.計算出現(xiàn)異常時
我們現(xiàn)在來模擬一個出現(xiàn)異常的場景。為了簡潔性，我們還是將一個字符串大寫，但是我們會模擬延時進行該操作。我們會使用thenApplyAsyn(Function, Executor)，第一個參數(shù)是大寫轉(zhuǎn)化方法，第二個參數(shù)是一個延時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());
}
首先，我們新建了一個已經(jīng)完成并帶有返回值message的CompletableFuture對象。然后我們調(diào)用thenApplyAsync方法，該方法會返回一個新的CompletableFuture。這個方法用異步的方式執(zhí)行大寫操作。這里還展示了如何使用delayedExecutor(timeout, timeUnit)方法來延時異步操作。
然后我們創(chuàng)建了一個handler stage，exceptionHandler，這個階段會處理一切異常并返回另一個消息message upon cancel。
最后，我們顯式的完成第二個階段并拋出異常，它會導(dǎo)致進行大寫操作的階段拋出CompletionException。它還會觸發(fā)handler階段。
API補充：9.取消計算
CompletableFuture handle(BiFunction fn)
返回一個新的CompletionStage，無論之前的Stage是否正常運行完畢。傳入的參數(shù)包括上一個階段的結(jié)果和拋出異常。
和計算時異常處理很相似，我們可以通過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());
}
