前言

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

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

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

1、為什么叫CompletableFuture?

CompletableFuture字面翻譯過(guò)來(lái)，就是“可完成的Future”。

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

而Future由于不能主動(dòng)設(shè)置計(jì)算結(jié)果值，一旦調(diào)用get()進(jìn)行阻塞等待，要么當(dāng)計(jì)算結(jié)果產(chǎn)生，要么超時(shí)，才會(huì)返回。

下面的示例，比較簡(jiǎn)單的說(shuō)明了，CompletableFuture是如何被主動(dòng)完成的。

在下面這段代碼中，由于調(diào)用了complete方法，所以最終的打印結(jié)果是“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 構(gòu)造函數(shù)創(chuàng)建

最簡(jiǎn)單的方式就是通過(guò)構(gòu)造函數(shù)創(chuàng)建一個(gè)CompletableFuture實(shí)例。如下代碼所示。由于新創(chuàng)建的CompletableFuture還沒(méi)有任何計(jì)算結(jié)果，這時(shí)調(diào)用join，當(dāng)前線程會(huì)一直阻塞在這里。

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

此時(shí)，如果在另外一個(gè)線程中，主動(dòng)設(shè)置該CompletableFuture的值，則上面線程中的結(jié)果就能返回。

future.complete("test");

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

2.2 supplyAsync創(chuàng)建

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

supplyAsync有兩種簽名：

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

第一種只需傳入一個(gè)Supplier實(shí)例（一般使用lamda表達(dá)式），此時(shí)框架會(huì)默認(rèn)使用ForkJoin的線程池來(lái)執(zhí)行被提交的任務(wù)。

第二種可以指定自定義的線程池，然后將任務(wù)提交給該線程池執(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);

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

2.3 runAsync創(chuàng)建

CompletableFuture.runAsync()也可以用來(lái)創(chuàng)建CompletableFuture實(shí)例。與supplyAsync()不同的是，runAsync()傳入的任務(wù)要求是Runnable類型的，所以沒(méi)有返回值。因此，runAsync適合創(chuàng)建不需要返回值的計(jì)算任務(wù)。同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());

在示例中，由于任務(wù)沒(méi)有返回值， 所以最后的打印結(jié)果是"get result: null"。

3、常見(jiàn)的使用方式

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

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

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

其中，帶Async后綴的函數(shù)表示需要連接的后置任務(wù)會(huì)被單獨(dú)提交到線程池中，從而相對(duì)前置任務(wù)來(lái)說(shuō)是異步運(yùn)行的。

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

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

3.1 thenApply / thenAccept / thenRun

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

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

因此，簡(jiǎn)單起見(jià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通過(guò)調(diào)用thenApply將后置任務(wù)連接起來(lái)，并形成future2。

該示例的最終打印結(jié)果為11，可見(jiàn)程序在運(yùn)行中，future1的結(jié)果計(jì)算出來(lái)后，會(huì)傳遞給通過(guò)thenApply連接的任務(wù)，從而產(chǎn)生future2的最終結(jié)果為1+10=11。

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

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

3.2 thenCombine

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

        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為獨(dú)立的CompletableFuture任務(wù)，他們分別會(huì)在各自的線程中并行執(zhí)行，然后future1通過(guò)thenCombine與future2連接，并且以lamda表達(dá)式傳入處理結(jié)果的表達(dá)式，該表達(dá)式代表的任務(wù)會(huì)將future1與future2的結(jié)果作為入?yún)⒉⒂?jì)算他們的和。

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

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

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

3.3 thenCompose

前面講了thenCombine主要用于沒(méi)有前后依賴關(guān)系之間的任務(wù)進(jìn)行連接。那么，如果兩個(gè)任務(wù)之間有前后依賴關(guān)系，但是連接任務(wù)又是獨(dú)立的CompletableFuture，該怎么實(shí)現(xiàn)呢？

先來(lái)看一下直接使用thenApply來(lái)實(shí)現(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嵌套，而且在獲取結(jié)果的時(shí)候，也需要嵌套調(diào)用join或者get。

這樣，當(dāng)連接的任務(wù)越多時(shí)，代碼會(huì)變得越來(lái)越復(fù)雜，嵌套獲取層級(jí)也越來(lái)越深。

因此，需要一種方式，能將這種嵌套模式展開(kāi)，使其沒(méi)有那么多層級(jí)。

thenCompose的主要目的就是解決這個(gè)問(wèn)題（這里也可以將thenCompose的作用類比于stream接口中的flatMap，因?yàn)樗麄兌伎梢詫㈩愋颓短渍归_(kāi)）。

看一下通過(guò)thenCompose如何實(shí)現(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());

通過(guò)示例代碼可以看出來(lái)，很明顯，在使用了thenCompose后，future2不再存在CompletableFuture類型嵌套了，從而比較簡(jiǎn)潔的達(dá)到了我們的目的。

3.4 whenComplete

whenComplete主要用于注入任務(wù)完成時(shí)的回調(diào)通知邏輯。這個(gè)解決了傳統(tǒng)future在任務(wù)完成時(shí)，無(wú)法主動(dòng)發(fā)起通知的問(wèn)題。前置任務(wù)會(huì)將計(jì)算結(jié)果或者拋出的異常作為入?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獲得的結(jié)果是前置任務(wù)的結(jié)果，whenComplete中的邏輯不會(huì)影響計(jì)算結(jié)果。

3.5 handle

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

以下為示例：

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

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

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

最新評(píng)論