前言

CompletableFuture是java.util.concurrent庫在java 8中新增的主要工具，同傳統(tǒng)的Future相比，其支持流式計算、函數(shù)式編程、完成通知、自定義異常處理等很多新的特性。

由于函數(shù)式編程在java中越來越多的被使用到，熟練掌握CompletableFuture，對于更好的使用java 8后的主要新特性很重要。

簡單起見，本文使用的CompletableFuture版本為java 8（java 11的CompletableFuture新增了一些方法）。

1、為什么叫CompletableFuture?

CompletableFuture字面翻譯過來，就是“可完成的Future”。

同傳統(tǒng)的Future相比較，CompletableFuture能夠主動設置計算的結果值（主動終結計算過程，即completable），從而在某些場景下主動結束阻塞等待。

而Future由于不能主動設置計算結果值，一旦調(diào)用get()進行阻塞等待，要么當計算結果產(chǎn)生，要么超時，才會返回。

下面的示例，比較簡單的說明了，CompletableFuture是如何被主動完成的。

在下面這段代碼中，由于調(diào)用了complete方法，所以最終的打印結果是“manual test”，而不是"test"。

CompletableFuture<String> future = CompletableFuture.supplyAsync(()->{
    try{
        Thread.sleep(1000L);
        return "test";
    } catch (Exception e){
        return "failed test";
    }
});
future.complete("manual test");
System.out.println(future.join());

2、創(chuàng)建CompletableFuture

2.1 構造函數(shù)創(chuàng)建

最簡單的方式就是通過構造函數(shù)創(chuàng)建一個CompletableFuture實例。如下代碼所示。由于新創(chuàng)建的CompletableFuture還沒有任何計算結果，這時調(diào)用join，當前線程會一直阻塞在這里。

CompletableFuture<String> future = new CompletableFuture();
String result = future.join();
System.out.println(result);

此時，如果在另外一個線程中，主動設置該CompletableFuture的值，則上面線程中的結果就能返回。

future.complete("test");

這展示了CompletableFuture最簡單的創(chuàng)建及使用方法。

2.2 supplyAsync創(chuàng)建

CompletableFuture.supplyAsync()也可以用來創(chuàng)建CompletableFuture實例。通過該函數(shù)創(chuàng)建的CompletableFuture實例會異步執(zhí)行當前傳入的計算任務。在調(diào)用端，則可以通過get或join獲取最終計算結果。

supplyAsync有兩種簽名：

public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

第一種只需傳入一個Supplier實例（一般使用lamda表達式），此時框架會默認使用ForkJoin的線程池來執(zhí)行被提交的任務。

第二種可以指定自定義的線程池，然后將任務提交給該線程池執(zhí)行。

下面為使用supplyAsync創(chuàng)建CompletableFuture的示例：

CompletableFuture<String> future = CompletableFuture.supplyAsync(()->{
    System.out.println("compute test");
    return "test";
});
String result = future.join();
System.out.println("get result: " + result);

在示例中，異步任務中會打印出“compute test”，并返回"test"作為最終計算結果。所以，最終的打印信息為“get result: test”。

2.3 runAsync創(chuàng)建

CompletableFuture.runAsync()也可以用來創(chuàng)建CompletableFuture實例。與supplyAsync()不同的是，runAsync()傳入的任務要求是Runnable類型的，所以沒有返回值。因此，runAsync適合創(chuàng)建不需要返回值的計算任務。同supplyAsync()類似，runAsync()也有兩種簽名：

public static CompletableFuture<Void> runAsync(Runnable runnable)
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)

 下面為使用runAsync()的例子：

CompletableFuture<Void> future = CompletableFuture.runAsync(()->{
    System.out.println("compute test");
});
System.out.println("get result: " + future.join());

在示例中，由于任務沒有返回值， 所以最后的打印結果是"get result: null"。

3、常見的使用方式

同F(xiàn)uture相比，CompletableFuture最大的不同是支持流式（Stream）的計算處理，多個任務之間，可以前后相連，從而形成一個計算流。比如：任務1產(chǎn)生的結果，可以直接作為任務2的入?yún)?，參與任務2的計算，以此類推。

CompletableFuture中常用的流式連接函數(shù)包括：

• thenApply
• thenApplyAsync
• thenAccept
• thenAcceptAsync
• thenRun
• thenRunAsync
• thenCombine
• thenCombineAsync
• thenCompose
• thenComposeAsync
• whenComplete
• whenCompleteAsync
• handle
• handleAsync

其中，帶Async后綴的函數(shù)表示需要連接的后置任務會被單獨提交到線程池中，從而相對前置任務來說是異步運行的。

除此之外，兩者沒有其他區(qū)別。

因此，為了快速理解，在接下來的介紹中，我們主要介紹不帶Async的版本。

3.1 thenApply / thenAccept / thenRun

這里將thenApply / thenAccept / thenRun放在一起講，因為這幾個連接函數(shù)之間的唯一區(qū)別是提交的任務類型不一樣。

• thenApply提交的任務類型需遵從Function簽名，也就是有入?yún)⒑头祷刂?，其中入?yún)榍爸萌蝿盏慕Y果。
• thenAccept提交的任務類型需遵從Consumer簽名，也就是有入?yún)⒌菦]有返回值，其中入?yún)榍爸萌蝿盏慕Y果。
• thenRun提交的任務類型需遵從Runnable簽名，即沒有入?yún)⒁矝]有返回值。

因此，簡單起見，我們這里主要講thenApply。

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(()->{
    System.out.println("compute 1");
    return 1;
});
CompletableFuture<Integer> future2 = future1.thenApply((p)->{
    System.out.println("compute 2");
    return p+10;
});
System.out.println("result: " + future2.join());

在上面的示例中，future1通過調(diào)用thenApply將后置任務連接起來，并形成future2。

該示例的最終打印結果為11，可見程序在運行中，future1的結果計算出來后，會傳遞給通過thenApply連接的任務，從而產(chǎn)生future2的最終結果為1+10=11。

當然，在實際使用中，我們理論上可以無限連接后續(xù)計算任務，從而實現(xiàn)鏈條更長的流式計算。

需要注意的是，通過thenApply / thenAccept / thenRun連接的任務，當且僅當前置任務計算完成時，才會開始后置任務的計算。因此，這組函數(shù)主要用于連接前后有依賴的任務鏈。

3.2 thenCombine

同前面一組連接函數(shù)相比，thenCombine最大的不同是連接任務可以是一個獨立的CompletableFuture（或者是任意實現(xiàn)了CompletionStage的類型），從而允許前后連接的兩個任務可以并行執(zhí)行（后置任務不需要等待前置任務執(zhí)行完成），最后當兩個任務均完成時，再將其結果同時傳遞給下游處理任務，從而得到最終結果。

        CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(()->{
            System.out.println("compute 1");
            return 1;
        });
        CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(()->{
            System.out.println("compute 2");
            return 10;
        });
        CompletableFuture<Integer> future3 = future1.thenCombine(future2, (r1, r2)->r1 + r2);
        System.out.println("result: " + future3.join());

上面示例代碼中，future1和future2為獨立的CompletableFuture任務，他們分別會在各自的線程中并行執(zhí)行，然后future1通過thenCombine與future2連接，并且以lamda表達式傳入處理結果的表達式，該表達式代表的任務會將future1與future2的結果作為入?yún)⒉⒂嬎闼麄兊暮汀?/p>

因此，上面示例代碼中，最終的打印結果是11。

一般，在連接任務之間互相不依賴的情況下，可以使用thenCombine來連接任務，從而提升任務之間的并發(fā)度。

注意，thenAcceptBoth、thenAcceptBothAsync、runAfterBoth、runAfterBothAsync的作用與thenConbime類似，唯一不同的地方是任務類型不同，分別是BiConumser、Runnable。

3.3 thenCompose

前面講了thenCombine主要用于沒有前后依賴關系之間的任務進行連接。那么，如果兩個任務之間有前后依賴關系，但是連接任務又是獨立的CompletableFuture，該怎么實現(xiàn)呢？

先來看一下直接使用thenApply來實現(xiàn)：

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(()->{
    System.out.println("compute 1");
    return 1;
});
CompletableFuture<CompletableFuture<Integer>> future2 =
        future1.thenApply((r)->CompletableFuture.supplyAsync(()->r+10));
System.out.println(future2.join().join());

可以發(fā)現(xiàn)，上面示例代碼中，future2的類型變成了CompletableFuture嵌套，而且在獲取結果的時候，也需要嵌套調(diào)用join或者get。

這樣，當連接的任務越多時，代碼會變得越來越復雜，嵌套獲取層級也越來越深。

因此，需要一種方式，能將這種嵌套模式展開，使其沒有那么多層級。

thenCompose的主要目的就是解決這個問題（這里也可以將thenCompose的作用類比于stream接口中的flatMap，因為他們都可以將類型嵌套展開）。

看一下通過thenCompose如何實現(xiàn)上面的代碼：

        CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(()->{
            System.out.println("compute 1");
            return 1;
        });
        CompletableFuture<Integer> future2 = future1.thenCompose((r)->CompletableFuture.supplyAsync(()->r+10));
        System.out.println(future2.join());

通過示例代碼可以看出來，很明顯，在使用了thenCompose后，future2不再存在CompletableFuture類型嵌套了，從而比較簡潔的達到了我們的目的。

3.4 whenComplete

whenComplete主要用于注入任務完成時的回調(diào)通知邏輯。這個解決了傳統(tǒng)future在任務完成時，無法主動發(fā)起通知的問題。前置任務會將計算結果或者拋出的異常作為入?yún)鬟f給回調(diào)通知函數(shù)。

以下為示例：

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(()->{
    System.out.println("compute 1");
    return 1;
});
CompletableFuture future2 = future1.whenComplete((r, e)->{
    if(e != null){
        System.out.println("compute failed!");
    } else {
        System.out.println("received result is " + r);
    }
});
System.out.println("result: " + future2.join());

需要注意的是，future2獲得的結果是前置任務的結果，whenComplete中的邏輯不會影響計算結果。

3.5 handle

handle與whenComplete的作用有些類似，但是handle接收的處理函數(shù)有返回值，而且返回值會影響最終獲取的計算結果。

以下為示例：

CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(()->{
    System.out.println("compute 1");
    return 1;
});
CompletableFuture<Integer> future2 = future1.handle((r, e)->{
    if(e != null){
        System.out.println("compute failed!");
        return r;
    } else {
        System.out.println("received result is " + r);
        return r + 10;
    }
});
System.out.println("result: " + future2.join());

在以上示例中，打印出的最終結果為11。說明經(jīng)過handle計算后產(chǎn)生了新的結果。

到此這篇關于Java中的CompletableFuture基本用法的文章就介紹到這了,更多相關CompletableFuture用法內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

最新評論