歡迎光臨
每天分享高質量文章

20 個使用 Java CompletableFuture 的例子

(點選上方公眾號,可快速關註)


來源:鳥窩,

colobu.com/2018/03/12/20-Examples-of-Using-Java%E2%80%99s-CompletableFuture/

在Java中非同步程式設計,不一定非要使用rxJava, Java本身的庫中的CompletableFuture可以很好的應對大部分的場景。

這篇文章介紹 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));

}

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

2、執行一個簡單的非同步階段

這個例子建立一個一個非同步執行的階段:

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這個預設的行為。

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

}

註意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());

}

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());

}

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

}

本例中消費者同步地執行,所以我們不需要在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);

}

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());

}

讓我們看一下細節。

首先我們建立了一個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());

}

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”));

}

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

這個例子演示了依賴的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);

}

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());

}

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字尾。

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());

}

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());

}

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 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”);

    });

    ass

19、當所有的階段都完成後非同步地建立一個階段

使用thenApplyAsync()替換那些單個的CompletableFutures的方法,allOf()會在通用池中的執行緒中非同步地執行。所以我們需要呼叫join方法等待它完成。

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、真實的例子

Now that the functionality of CompletionStage and specifically CompletableFuture is explored, the below example applies them in a practical scenario:

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

  • 首先非同步呼叫cars方法獲得Car的串列,它傳回CompletionStage場景。cars消費一個遠端的REST API。

  • 然後我們複合一個CompletionStage填寫每個汽車的評分,透過rating(manufacturerId)傳回一個CompletionStage, 它會非同步地獲取汽車的評分(可能又是一個REST API呼叫)

  • 當所有的汽車填好評分後,我們結束這個串列,所以我們呼叫allOf得到最終的階段, 它在前面階段所有階段完成後才完成。

  • 在最終的階段呼叫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();

因為每個汽車的實體都是獨立的,得到每個汽車的評分都可以非同步地執行,這會提高系統的效能(延遲),而且,等待所有的汽車評分被處理使用的是allOf方法,而不是手工的執行緒等待(Thread#join() 或 a CountDownLatch)。

這些例子可以幫助你更好的理解相關的API,你可以在github上得到所有的例子的程式碼。

https://github.com/manouti/completablefuture-examples

其它參考檔案

  • Reactive programming with Java 8 and simple-react : The Tutorial

    https://medium.com/@johnmcclean/reactive-programming-with-java-8-and-simple-react-the-tutorial-3634f512eeb1

  • CompletableFuture Overview

    http://javaday.org.ua/completablefuture-overview/

  • CompletableFuture vs Future: going async with Java 8 new features

    https://blog.takipi.com/back-to-the-completablefuture-java-8-feature-highlight/

  • spotify/completable-futures

    https://github.com/spotify/completable-futures

看完本文有收穫?請轉發分享給更多人

關註「ImportNew」,提升Java技能

贊(0)

分享創造快樂