Java中的CompletableFuture異步編程詳解

更新時間：2023年12月22日 09:16:18 作者：Java面試365

這篇文章主要介紹了Java中的CompletableFuture異步編程詳解,只要提到多線程來優(yōu)化性能,那么必定離不開異步化,異步化的出現(xiàn)才是多線程優(yōu)化性能這個核心方案的基礎,需要的朋友可以參考下

場景引入

只要提到多線程來優(yōu)化性能，那么必定離不開異步化，異步化的出現(xiàn)才是多線程優(yōu)化性能這個核心方案的基礎。異步化其實我們早已接觸，如下Thread類，主線程不需要等待線程T1,T2的執(zhí)行結(jié)果，就能實現(xiàn)異步邏輯。

public static void main(String[] args) {
    Thread T1 = new Thread(()->{
        // 執(zhí)行方法A邏輯
    });
    Thread T2 = new Thread(()->{
        // 執(zhí)行方法B邏輯
    });
    // 省略其它邏輯
}

這種方案雖然可行，但是在開發(fā)中明顯不是最優(yōu)，現(xiàn)實生產(chǎn)業(yè)務對應各種各樣的需求，簡單的Thread已經(jīng)不滿足需求，所以Java在1.8版本提出CompletableFuture工具類來解決生產(chǎn)中遇到的異步化問題。

CompletableFuture初體驗

先從之前提到的華羅庚提出的最優(yōu)泡茶問題入手，簡易體驗下CompletableFuture工具類的優(yōu)勢。最優(yōu)泡茶問題可以分為如下步驟

在用FutureTask實現(xiàn)時是將上述步驟拆分為兩個線程執(zhí)行，如下所示

通過代碼實現(xiàn)明顯能感覺到，線程需要手動維護，代碼邏輯復雜，不能專注于業(yè)務代碼。

為了方便采用CompletableFuture實現(xiàn)，將最優(yōu)泡茶方案進一步細分，如下所示，將流程拆分三個線程執(zhí)行，線程F3需要等待線程F1,F2都返回后才執(zhí)行。

代碼實現(xiàn)如下

public static void main(String[] args) throws Exception {
    // 無返回值的異步調(diào)用
    CompletableFuture f1 = CompletableFuture.runAsync(()->{
        System.out.println("F1 洗水壺");
        sleep(1);
        System.out.println("F1 燒水");
        sleep(15);
    });
    // 有返回值的實例化
    CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> {
        System.out.println("F2 洗茶壺");
        sleep(1);
        System.out.println("F2 洗茶杯");
        sleep(2);
        System.out.println("F2 拿茶葉");
        sleep(1);
        return "龍井";
    });
    // f3等待f1和f2到達后才能執(zhí)行
    // param1是f1的返回值，這里沒有就是空  param2是f2的返回值
    CompletableFuture f3 = f1.thenCombine(f2, (param1, param2) -> {
        System.out.println("F3 拿到茶葉：" + param2);
        System.out.println("F3 泡茶");
        return param2;
    });
    // 阻塞等待
    f3.get();
}
public static void sleep(int t){
    try {
        TimeUnit.SECONDS.sleep(t);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

返回結(jié)果

從實現(xiàn)代碼來看CompletableFuture有如下優(yōu)勢

不需要手動維護線程，不需要手動給任務分配工作線程。
代碼簡練可以專注業(yè)務邏輯。

創(chuàng)建CompletableFuture對象

創(chuàng)建CompletableFuture除了初體驗CompletableFuture代碼中的兩種還有兩種，總共四種方法簽名如下

// 有返回值   supplier供給型接口（不進有出）
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier);
// 有返回值，任務使用自定義的線程池
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,Executor executor);
// 無返回值  Runnable接口執(zhí)行run方法無返回值
public static CompletableFuture<Void> runAsync(Runnable runnable);
// 無返回值，任務使用自定義的線程池
public static CompletableFuture<Void> runAsync(Runnable runnable,Executor executor)

注意：在創(chuàng)建CompletableFuture對象時盡量使用自定義的線程池，因為默認是采用公共的ForkJoinPool線程池，如果某個CompletableFuture中有I/O操作非常耗時的就會阻塞該線程池中所有的線程，導致線程饑餓的風險，進而影響整個系統(tǒng)的性能，生產(chǎn)中最好按照業(yè)務創(chuàng)建不同類型的線程池，互不干擾。

CompletableFuture類定義

工具類CompletableFuture類定義如下

 public class CompletableFuture<T> implements Future<T>, CompletionStage<T>

由于CompletableFuture類實現(xiàn)了Future接口，所以異步線程的兩大問題，線程什么時候執(zhí)行完畢？線程的返回值是什么？都可以利用Future接口的特性解決，

另外CompletableFuture也實現(xiàn)了CompletionStage接口，那CompletionStage又是什么呢？

CompletionStage接口

CompletionStage接口是去描述任務之間的時序關(guān)系，包括前面提到的f1.thenCombine(f2, (param1, param2) -> {})就是一種典型的AND聚合關(guān)系，還能描述OR聚合關(guān)系，串行關(guān)系，以及異步編程中的異常處理關(guān)系。

AND匯聚關(guān)系

AND匯聚關(guān)系即表示依賴任務全部執(zhí)行完畢才能執(zhí)行當前任務，在CompletableFuture初體驗中有使用，可以參考。

CompletionStage接口描述匯聚關(guān)系的方法簽名如下：

public CompletionStage<V> thenCombine(CompletionStage other,BiFunction fn);
public CompletionStage<V> thenCombineAsync(CompletionStage other,BiFunction fn);
public CompletionStage<V> thenCombineAsync(CompletionStage other,BiFunction fn,Executor executor);
public CompletionStage<Void> thenAcceptBoth(CompletionStage other,BiConsumer consumer);
public CompletionStage<Void> thenAcceptBothAsync(CompletionStage other,BiConsumer consumer);
public CompletionStage<Void> thenAcceptBothAsync(CompletionStage other,BiConsumer consumer,Executor executor);
public CompletionStage<Void> runAfterBoth(CompletionStage other,Runnable action);
public CompletionStage<Void> runAfterBothAsync(CompletionStage other,Runnable action);
public CompletionStage<Void> runAfterBothAsync(CompletionStage other,Runnable action,Executor executor);

thenCombine、thenAcceptBoth、runAfterBoth這三個系列方法的區(qū)別源自核心參數(shù)區(qū)別

BiFunction fn參數(shù)是函數(shù)式接口，這個參數(shù)既能支持入?yún)⒁仓С址祷刂怠?/li>
BiConsumer consumer參數(shù)是消費型接口，只有入?yún)]有返回值。
Runnable action 參數(shù)不支持入?yún)⒁膊恢С殖鰠ⅰ?/li>
Executor executor 表示指定線程池，不使用公共的ForkJoin線程池。
其中Async表示異步執(zhí)行fn、consumer、action邏輯。

OR聚合關(guān)系

OR聚合關(guān)系表示當其中一個依賴任務執(zhí)行完畢就可以執(zhí)行當前任務。

CompletionStage接口描述聚合關(guān)系的方法簽名如下：

public CompletionStage<U> applyToEither(CompletionStage other,Function fn);
public CompletionStage<U> applyToEitherAsync(CompletionStage other,Function fn);
public CompletionStage<U> applyToEitherAsync(CompletionStage other,Function fn,Executor executor);
public CompletionStage<Void> acceptEither(CompletionStage other,Consumer action);
public CompletionStage<Void> acceptEitherAsync(CompletionStage other,Consumer action);
public CompletionStage<Void> acceptEitherAsync(CompletionStage other,Consumer action,Executor executor);
public CompletionStage<Void> runAfterEither(CompletionStage other,Runnable action);
public CompletionStage<Void> runAfterEitherAsync(CompletionStage other,Runnable action);
public CompletionStage<Void> runAfterEitherAsync(CompletionStage other,Runnable action,Executor executor)

這三個系列方法的區(qū)別也是源自核心參數(shù)區(qū)別。

public static void main(String[] args) throws ExecutionException, InterruptedException {
    CompletableFuture<Double> f1 = CompletableFuture.supplyAsync(()->{
        double num = Math.random();
        System.out.println("f1 返回值："+num);
        // TimeUnit.SECONDS.sleep(1);
        return num;
    });
    CompletableFuture<Double> f2 = CompletableFuture.supplyAsync(()->{
        double num = Math.random();
        System.out.println("f2 返回值："+num);
        return num;
    });
    CompletableFuture<Void> f3 = f1.acceptEither(f2, (num) -> {
        System.out.println("最后返回值：" + num);
    });
    // 不是需要返回值，而是阻塞等待f3執(zhí)行結(jié)束
    // 返回結(jié)果就是任務f1,f2中的任意一個返回值，需要保證f1,f2的返回值類型相同
    f3.get();
}

串行關(guān)系

串行關(guān)系表示依賴任務按照編寫順序先后執(zhí)行。

CompletionStage接口描述串行關(guān)系的方法簽名如下：

public <U> CompletionStage<U> thenApply(Function fn);
public <U> CompletionStage<U> thenApplyAsync(Function fn);
public <U> CompletionStage<U> thenApplyAsync(Function fn,Executor executor);
public CompletionStage<Void> thenAccept(Consume action);
public CompletionStage<Void> thenAcceptAsync(Consumer action);
public CompletionStage<Void> thenAcceptAsync(Consumer action,Executor executor);
public CompletionStage<Void> thenRun(Runnable action);
public CompletionStage<Void> thenRunAsync(Runnable action);
public CompletionStage<Void> thenRunAsync(Runnable action,Executor executor);
public <U> CompletionStage<U> thenCompose(Function<? super T, ? extends CompletionStage<U>> fn);
public <U> CompletionStage<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn);
public <U> CompletionStage<U> thenComposeAsync(Function<? super T, ? extends CompletionStage<U>> fn,Executor executor);

thenApply、thenAccept、thenRun這三個系列方法的區(qū)別也是源自核心參數(shù)區(qū)別。

需要注意的是thenCompose系列方法，這個方法會新創(chuàng)建出一個子流程，最終結(jié)果和thenApply系列方法相同。

如下所示，從上往下依次執(zhí)行。

public static void main(String[] args) throws ExecutionException, InterruptedException {
    CompletableFuture completableFuture = CompletableFuture.supplyAsync(()->{
        return "hello world";
    }).thenApply((param1)->{
        return param1 + " CompletableFuture";
    }).thenAccept((param2)->{
        param2 = param2.toUpperCase();
        System.out.println(param2);
    });
    completableFuture.get();
}

異常關(guān)系

在串行關(guān)系、OR聚合關(guān)系、AND匯聚關(guān)系中，由于其核心方法參數(shù)fn、consumer、action都不允許拋出異常，但是都無法限制它們拋出異常，如下所示

public static void main(String[] args) throws Exception {
    CompletableFuture<Integer> f0 = CompletableFuture
        .supplyAsync(()->(7/0))
        .thenApply(r->r*10);
    System.out.println(f0.get());
}

非異步編程可以采用try{}catch{}捕獲那么異步編程如何處理呢？

CompletionStage接口支持異常處理方法簽名如下

public CompletionStage<T> exceptionally(Function<Throwable, ? extends T> fn);
public CompletionStage<T> whenComplete(BiConsumer<? super T, ? super Throwable> action);
public CompletionStage<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action);
public CompletionStage<T> whenCompleteAsync(BiConsumer<? super T, ? super Throwable> action,Executor executor);
public <U> CompletionStage<U> handle(BiFunction<? super T, Throwable, ? extends U> fn);
public <U> CompletionStage<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn);
public <U> CompletionStage<U> handleAsync(BiFunction<? super T, Throwable, ? extends U> fn,Executor executor);

exceptionally使用類似try{}catch{}中的 catch{}。
whenComplete系列使用類似try{}finally{}中的 finally{}，無論是否發(fā)生異常都會執(zhí)行whenComplete中的邏輯代碼，但是無返回結(jié)果。
handle系列使用和whenComplete一樣，不過handle支持返回結(jié)果

測試代碼如下

public static void main(String[] args) throws Exception {
    CompletableFuture<Integer> f0 = CompletableFuture
            .supplyAsync(()->(7/0))
            .thenApply(r->r*10)
            .exceptionally((throwable)->{
                // 相當于catch完后，不拋出異常
                System.out.println("異常了："+throwable);
                return 1;
            }).whenComplete((integer,throwable)->{
                // integer 返回值   throwable返回異常（如果exceptionally捕獲返回null）
                System.out.println("返回值："+integer);
                System.out.println("異常："+throwable);
            }).handle((integer,throwable)->{
                // integer 返回值   throwable返回異常（如果exceptionally捕獲返回null）
                System.out.println("返回值："+integer);
                System.out.println("異常："+throwable);
                return 2;
            });
    System.out.println(f0.get());
}

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

您可能感興趣的文章: