從Future聊起

Future是java 1.5引入的異步編程api,它表示一個異步計算結果，提供了獲取異步結果的能力，解決了多線程場景下Runnable線程任務無法獲取結果的問題。

但是其獲取異步結果的方式并不夠優(yōu)雅，我們必須使用Future.get的方式阻塞調用線程，或者使用輪詢方式判斷 Future.isDone 任務是否結束，再獲取結果。

public interface Future<V> {

    //任務是否完成
    boolean isDone();
    
    //阻塞調用線程獲取異步結果
    V get() throws InterruptedException, ExecutionException;

   //在指定時間內阻塞線程獲取異步結果
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

假如存在多個異步任務相互依賴，一個或多個異步線程任務需要依賴上一個異步線程任務結果，并且多個異步任務能夠組合結果，顯然這種阻塞線程的方式并不能優(yōu)雅解決。

我們更希望能夠提供一種異步回調的方式，組合各種異步任務，而無需開發(fā)者對多個異步任務結果的監(jiān)聽編排。

為了解決優(yōu)化上述問題，java8 新增了CompletableFutureAPI ,其大大擴展了Future能力，并提供了異步任務編排能力。

CompletableFuture

CompletableFuture實現(xiàn)了新的接口CompletionStage，并擴展了Future接口。查看類圖

創(chuàng)建異步任務

CompletableFuture 提供了四種方法去創(chuàng)建一個異步任務。

• static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier):創(chuàng)建一個有返回值的異步任務實例
• static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,Executor executor):創(chuàng)建一個有返回值的異步任務實例，可以指定線程池
• static CompletableFuture<Void> runAsync(Runnable runnable):創(chuàng)建一個無返回值的任務實例
• static CompletableFuture<Void> runAsync(Runnable runnable,Executor executor):創(chuàng)建一個無返回值的任務實例，允許指定線程池

著幾個方法本質上是有返回值和無返回值兩種類型方法，supply方法可以獲取異步結果，而run方法則無返回值，根據(jù)需要使用。

同時兩種類型的方法均提供了指定線程池的重載，如果不指定線程池會默認使用ForkJoinPool.commonPool(),默認線程數(shù)為cpu核心數(shù)，建議使用自定義線程池的方式，避免線程資源競爭

一個簡單樣例

        CompletableFuture<Void> runAsync = CompletableFuture.runAsync(() -> { System.out.println("無返回值任務"); });
        runAsync.get();
        CompletableFuture<String> supplyAsync = CompletableFuture.supplyAsync(() -> "hello completableFuture");
        String result = supplyAsync.get();
        System.out.println(result);

我們依然可以通過get()方法阻塞獲取異步結果任務，但是CompletableFuture主要還是用于異步回調及異步任務編排使用。

異步回調

在任務執(zhí)行結束后我們希望能夠自動觸發(fā)回調方法，CompletableFuture提供了兩種方法實現(xiàn)。

• CompletableFuture<T> whenComplete( BiConsumer<? super T, ? super Throwable> action):當上一階段任務執(zhí)行結束后，回調方法接受上一階段結果或者異常，返回上一階段任務結果
• <U> CompletableFuture<U> handle( BiFunction<? super T, Throwable, ? extends U> fn):當上一階段任務執(zhí)行結束后，回調方法接受上一階段結果或者異常,并最終返回回調方法處理結果
• CompletableFuture<T> exceptionally( Function<Throwable, ? extends T> fn):上一階段任務出現(xiàn)異常后的回調，返回結果是回調函數(shù)的返回結果。

whenComplete 與 handle 區(qū)別：兩者均接受上一階段任務結果或異常，但是whenComplete 回調中沒有返回值，所以其結果是上一階段任務，而handle 最終返回的是其回調方法方法，其主要是BiConsumer與BiFunction的區(qū)別。

異步編排

CompletionStage表示異步計算的一個階段，當一個計算處理完成后會觸發(fā)其他依賴的階段。當然一個階段的觸發(fā)也可以是由多個階段的完成觸發(fā)或者多個中的任意一個完成觸發(fā)。該接口定義了異步任務編排的各種場景，CompletableFuture則實現(xiàn)了這些場景。

可以把這些場景大致分為三類：串行、AND和OR。下面會逐個分析各個場景，接口中定義的以Async結尾的方法，指下一階段任務會被單獨提交到線程池中執(zhí)行，后面不在贅述。

串行

當上一階段任務執(zhí)行完畢后，繼續(xù)提交執(zhí)行其他任務

• <U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn):接收上一階段任務結果，并可獲取返回值。
• CompletableFuture<Void> thenAccept(Consumer<? super T> action):接收上一階段任務結果，無返回值。
• CompletableFuture<Void> thenRun(Runnable action):不接收上一階段任務結果，并且無返回值。

T：上一個任務返回結果的類型 U：當前任務的返回值類型

AND

組合多個異步任務，當多個任務執(zhí)行完畢繼續(xù)執(zhí)行其他任務

• <U,V> CompletableFuture<V> thenCombine( CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn):上一階段任務與other任務均執(zhí)行結束，接收兩個任務的結果，并可獲取返回值
• <U> CompletableFuture<U> thenCompose( Function<? super T, ? extends CompletionStage<U>> fn): 使用上一階段任務的結果，返回一個新的CompletableFuture實例
• <U> CompletableFuture<Void> thenAcceptBoth( CompletionStage<? extends U> other, BiConsumer<? super T, ? super U> action):上一階段任務與other任務均執(zhí)行結束，接收兩個任務的結果,無返回值
• CompletableFuture<Void> runAfterBoth(CompletionStage<?> other, Runnable action):上一階段任務與other任務均執(zhí)行結束，不接收兩個任務的結果,無返回值
• static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs):等待所有異步任務執(zhí)行結束

T：上一個任務返回結果的類型 U：上一個other任務的返回值類型 V:當前任務返回值

OR

當多個任務中任意任務執(zhí)行完成則繼續(xù)執(zhí)行其他任務。

• <U> CompletableFuture<U> applyToEither( CompletionStage<? extends T> other, Function<? super T, U> fn): 接收上一階段任務與other任務最快執(zhí)行完成的結果，并可獲取返回值
• CompletableFuture<Void> acceptEither( CompletionStage<? extends T> other, Consumer<? super T> action):接收上一階段任務與other任務最快執(zhí)行完成的結果，無返回值
• CompletableFuture<Void> runAfterEither(CompletionStage<?> other, Runnable action):上一階段任務與other任務任意任務完成執(zhí)行，不接收結果，無返回值
• static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs):組合多個任務，返回最快執(zhí)行結束的任務結果

Future 機制擴展

CompletableFuture不僅實現(xiàn)了Future接口，同時對其進行了擴展，提供了更加優(yōu)雅的實現(xiàn)。

• T join() :與get()方法用法一致，阻塞調用線程獲取結果，但是不會拋出具體異常，簡化了使用上下文
• T getNow(T valueIfAbsent):當任務結束返回任務結果，否則返回給定的結果valueIfAbsent。
• boolean complete(T value):當任務未結束時設置給定的結果value并結束任務,已結束的任務不會生效。
• boolean completeExceptionally(Throwable ex):當任務未結束時設置異常結果并結束任務，已結束的任務不會生效

CompletableFuture 實踐

我們通過CompletableFuture實現(xiàn)一個經(jīng)典的燒水程序。

我們可以把這個流程分為三個異步任務。

任務1：洗水壺->燒水

任務2：洗水壺->洗茶杯->拿茶葉

任務3：泡茶，需要等待任務1與任務2結束。

通過代碼模擬實現(xiàn)

        CompletableFuture<String> task1 = CompletableFuture.supplyAsync(() -> {
            System.out.println("洗水壺");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException ex) {
                ex.printStackTrace();
            }
            return "水壺";
        }).thenApply(e->{
            System.out.println("燒水");
            try {
                Thread.sleep(5000);
            } catch (InterruptedException ex) {
                ex.printStackTrace();
            }
            return "熱水";
        });
        //洗水壺->洗水杯->拿茶葉
        CompletableFuture<String> task2 = CompletableFuture.supplyAsync(() -> {
            System.out.println("洗茶壺");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException ex) {
                ex.printStackTrace();
            }
            return "茶壺";
        }).thenApply(e->{
            try {
                Thread.sleep(2000);
            } catch (InterruptedException ex) {
                ex.printStackTrace();
            }
            System.out.println("洗水杯");
            return "水杯";
        }).thenApply(e->{
            System.out.println("拿茶葉");
            return "茶葉";
        });
        //泡茶
        CompletableFuture<String> task3 = task1.thenCombine(task2, (a, b) -> {
            System.out.println("泡茶");
            return "茶";
        });
        String tea = task3.join();
        System.out.println(tea);

以上就是CompletableFuture 異步編排示例詳解的詳細內容，更多關于CompletableFuture 異步編排的資料請關注腳本之家其它相關文章！

最新評論