Java 8 中的 CompletableFuture 太好用了！20 個示例全分享…

時間  2021-03-26
標籤 java react git github 面試 spring 編程 多線程 併發 intellij-idea 欄目 Java 简体版
原文   原文鏈接
在Java中異步編程，不必定非要使用rxJava, Java自己的庫中的CompletableFuture能夠很好的應對大部分的場景。

這篇文章介紹 Java 8 的 CompletionStage API 和它的標準庫的實現 CompletableFuture。API經過例子的方式演示了它的行爲，每一個例子演示一到兩個行爲。java

既然CompletableFuture類實現了CompletionStage接口，首先咱們須要理解這個接口的契約。它表明了一個特定的計算的階段，能夠同步或者異步的被完成。你能夠把它當作一個計算流水線上的一個單元，最終會產生一個最終結果，這意味着幾個CompletionStage能夠串聯起來，一個完成的階段能夠觸發下一階段的執行，接着觸發下一次，接着……react

除了實現CompletionStage接口， CompletableFuture也實現了future接口, 表明一個未完成的異步事件。CompletableFuture提供了方法，可以顯式地完成這個future,因此它叫CompletableFuture。git

一、 建立一個完成的CompletableFuturegithub

最簡單的例子就是使用一個預約義的結果建立一個完成的CompletableFuture,一般咱們會在計算的開始階段使用它。面試

static void completedFutureExample() {
    CompletableFuture cf = CompletableFuture.completedFuture("message");
    assertTrue(cf.isDone());
    assertEquals("message", cf.getNow(null));
}

getNow(null)方法在future完成的狀況下會返回結果，就好比上面這個例子，不然返回null (傳入的參數)。spring

二、運行一個簡單的異步階段編程

這個例子建立一個一個異步執行的階段：多線程

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這個默認的行爲。intellij-idea

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

下面這個例子使用前面 #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()只有打斜操做被完成後才返回。

四、在前一個階段上異步應用函數

經過調用異步方法(方法後邊加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來異步地執行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());
}

六、消費前一階段的結果

若是下一階段接收了當前階段的結果，可是在計算的時候不須要返回值(它的返回類型是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方法。

七、異步地消費遷移階段的結果

一樣，可使用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);
}

八、完成計算異常

如今咱們來看一下異步操做如何顯式地返回異常，用來指示計算失敗。咱們簡化這個例子，操做處理一個字符串，把它轉換成答謝，咱們模擬延遲一秒。

咱們使用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階段。

九、取消計算

和完成異常相似，咱們能夠調用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());
}

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

下面的例子建立了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"));
}

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

和前一個例子很相似了，只不過咱們調用的是消費者函數 (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

這個例子演示了依賴的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三、 使用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四、使用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五、異步使用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六、組合 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七、當幾個階段中的一個完成，建立一個完成的階段

下面的例子演示了當任意一個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八、當全部的階段都完成後建立一個階段

上一個例子是當任意一個階段完成後接着處理，接下來的兩個例子演示當全部的階段完成後才繼續處理, 同步地方式和異步地方式兩種。

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九、當全部的階段都完成後異步地建立一個階段

使用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 的一些函數的功能，下面的例子是一個實踐場景：

  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)。

最後，關注公衆號Java技術棧，在後臺回覆：面試，能夠獲取我整理的 Java 併發多線程系列面試題和答案，很是齊全。

原文：https://mahmoudanouti.wordpre...\
來源：https://colobu.com/2018/03/12...

參考文檔：

https://medium.com/@johnmccle...\
http://javaday.org.ua/complet...\
https://blog.takipi.com/back-...\
https://github.com/spotify/co...

近期熱文推薦：

1.600+ 道 Java面試題及答案整理(2021最新版)

2.終於靠開源項目弄到 IntelliJ IDEA 激活碼了，真香！

3.阿里 Mock 工具正式開源，幹掉市面上全部 Mock 工具！

4.Spring Cloud 2020.0.0 正式發佈，全新顛覆性版本！

5.《Java開發手冊（嵩山版）》最新發布，速速下載！

以爲不錯，別忘了隨手點贊+轉發哦！

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程