CompletableFuture——20個使用的例子

*黑桃~A*#丨 2022-01-07 15:56:03 阅读数:460

completablefuture 使用 例子

目錄

 ​1、 創建一個完成的CompletableFuture​

 ​2、運行一個簡單的异步階段​

 ​3、在前一個階段上應用函數​

 ​4、在前一個階段上异步應用函數​

 ​5、使用定制的Executor在前一個階段上异步應用函數​

 ​6、消費前一階段的結果​

 ​7、异步地消費遷移階段的結果​

 ​8、完成計算异常​

 ​9、取消計算​

 ​10、在兩個完成的階段其中之一上應用函數​

 ​11、在兩個完成的階段其中之一上調用消費函數​

 ​12、在兩個階段都執行完後運行一個Runnable​

 ​13、 使用BiConsumer處理兩個階段的結果​

 ​14、使用BiFunction處理兩個階段的結果​

 ​15、异步使用BiFunction處理兩個階段的結果​

 ​16、組合 CompletableFuture​

 ​17、當幾個階段中的一個完成,創建一個完成的階段​

 ​18、當所有的階段都完成後創建一個階段​

 ​19、當所有的階段都完成後异步地創建一個階段​

 ​20、真實的例子​

這篇文章介紹 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));
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.

​getNow(null)​​方法在future完成的情况下會返回結果,就比如上面這個例子,否則返回null (傳入的參數)。

2、運行一個簡單的异步階段

這個例子創建一個一個异步執行的階段:

static void runAsyncExample() {
CompletableFuture cf = CompletableFuture.runAsync(() -> {
assertTrue(Thread.currentThread().isDaemon());
randomSleep();
});
assertFalse(cf.isDone());
sleepEnough();
assertTrue(cf.isDone());
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.

通過這個例子可以學到兩件事情:

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));
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.

注意​​thenApply​​方法名稱代錶的行為。

​then​​​意味著這個階段的動作發生當前的階段正常完成之後。本例中,當前節點完成,返回字符串​​message​​。

​Apply​​意味著返回的階段將會對結果前一階段的結果應用一個函數。

函數的執行會被阻塞,這意味著​​getNow()​​只有打斜操作被完成後才返回。

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());
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.

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());
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.
  • 19.
  • 20.

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);
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.

本例中消費者同步地執行,所以我們不需要在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);
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.

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());
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.

首先我們創建了一個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());
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.

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"));
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.

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"));
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.

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);
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.

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());
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.

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));
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.

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());
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.

16、組合 CompletableFuture

我們可以使用​​thenCompose()​​完成上面兩個例子。這個方法等待第一個階段的完成(大寫轉換), 它的結果傳給一個指定的返回CompletableFuture函數,它的結果就是返回的CompletableFuture的結果。

有點拗口,但是我們看例子來理解。函數需要一個大寫字符串做參數,然後返回一個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());
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.

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);
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.

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);
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.

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);
}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.

20、真實的例子

現在你已經了解了CompletionStage 和 CompletableFuture 的一些函數的功能,下面的例子是一個實踐場景:


  1. 首先异步調用​​cars​​方法獲得Car的列錶,它返回CompletionStage場景。​​cars​​消費一個遠程的REST API。
  2. 然後我們複合一個CompletionStage填寫每個汽車的評分,通過​​rating(manufacturerId)​​返回一個CompletionStage, 它會异步地獲取汽車的評分(可能又是一個REST API調用)
  3. 當所有的汽車填好評分後,我們結束這個列錶,所以我們調用​​allOf​​得到最終的階段, 它在前面階段所有階段完成後才完成。
  4. 在最終的階段調用​​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)。
    這些例子可以幫助你更好的理解相關的API。
版权声明:本文为[*黑桃~A*#丨]所创,转载请带上原文链接,感谢。 https://gsmany.com/2022/01/202201071556030952.html