Java中的CompletableFuture原理與用法

更新時間：2024年01月15日 15:05:21 作者：曠野歷程

CompletableFuture 是由Java8引入的,這讓我們編寫清晰可讀的異步代碼變得更加容易,該類功能比Future 更加強大,在Java中CompletableFuture用于異步編程,異步通常意味著非阻塞,運行任務(wù)單獨的線程,與主線程隔離,這篇文章介紹CompletableFuture原理與用法,一起看看吧

CompletableFuture 是由Java8引入的，這讓我們編寫清晰可讀的異步代碼變得更加容易，該類功能比Future 更加強大。

什么是 CompletableFuture

在Java中CompletableFuture用于異步編程，異步通常意味著非阻塞，運行任務(wù)單獨的線程，與主線程隔離。并且通過回調(diào)可以在主線程中得到異步任務(wù)的執(zhí)行狀態(tài)，是否完成和異常等信息。

通過這種方式，主線程不會被阻塞，不需要一直等到子線程完成。主線程可以并行的執(zhí)行其他任務(wù)。使用這種并行方式，可以極大的提高程序的性能。

為什么要引入 CompletableFuture

一些業(yè)務(wù)場景我們需要使用多線程異步執(zhí)行任務(wù)，加快任務(wù)執(zhí)行速度，所以 JDK5 新增了 Future 接口，用于描述一個異步計算的結(jié)果。

雖然 Future 以及相關(guān)使用方法提供了異步執(zhí)行任務(wù)的能力，但是對于結(jié)果的獲取卻是很不方便，我們必須使用 Future.get 的方式阻塞調(diào)用線程，或者使用輪詢方式判斷 Future.isDone 任務(wù)是否結(jié)束，再獲取結(jié)果。

Future<String> future = executor.submit(()->{
       Thread.sleep(2000);
       return "hello world";
});
// 獲取結(jié)果
System.out.println(future.get());

從上面的形式看來不能及時地得到計算結(jié)果，所以要實現(xiàn)真正的異步，上述這樣是完全不夠的。

若需要更強大的異步處理能力，單純使用 Future 接口或者 FutureTask 類并不能很好地完成以下業(yè)務(wù)場景：

將兩個異步計算合并為一個，這兩個異步計算之間相互獨立，同時第二個又依賴于第一個的結(jié)果；
等待Future集合種的所有任務(wù)都完成；
僅等待Future集合種最快結(jié)束的任務(wù)完成（有可能因為他們試圖通過不同的方式計算同一個值），并返回它的結(jié)果；
通過編程方式完成一個Future任務(wù)的執(zhí)行（即以手工設(shè)定異步操作結(jié)果的方式）；
應(yīng)對Future的完成時間（即當(dāng)Future的完成時間完成時會收到通知，并能使用Future的計算結(jié)果進(jìn)行下一步的的操作，不只是簡單地阻塞等待操作的結(jié)果）。

所以JDK 8.0新增了CompletableFuture 來解決上述這些痛點。

Future vs CompletableFuture

CompletableFuture 是 Future API的擴(kuò)展。

Future 被用于作為一個異步計算結(jié)果的引用。提供一個 isDone() 方法來檢查計算任務(wù)是否完成。當(dāng)任務(wù)完成時，get() 方法用來接收計算任務(wù)的結(jié)果。

Future 的局限性

不能手動完成

當(dāng)你寫了一個函數(shù)，用于通過一個遠(yuǎn)程API獲取一個電子商務(wù)產(chǎn)品最新價格。因為這個 API 太耗時，你把它允許在一個獨立的線程中，并且從你的函數(shù)中返回一個 Future?，F(xiàn)在假設(shè)這個API服務(wù)宕機了，這時你想通過該產(chǎn)品的最新緩存價格手工完成這個Future 。你會發(fā)現(xiàn)無法這樣做。

Future 的結(jié)果在非阻塞的情況下，不能執(zhí)行更進(jìn)一步的操作

Future 不會通知你它已經(jīng)完成了，它提供了一個阻塞的 get() 方法通知你結(jié)果。你無法給 Future 植入一個回調(diào)函數(shù)，當(dāng) Future 結(jié)果可用的時候，用該回調(diào)函數(shù)自動的調(diào)用 Future 的結(jié)果。

多個 Future 不能串聯(lián)在一起組成鏈?zhǔn)秸{(diào)用

有時候你需要執(zhí)行一個長時間運行的計算任務(wù)，并且當(dāng)計算任務(wù)完成的時候，你需要把它的計算結(jié)果發(fā)送給另外一個長時間運行的計算任務(wù)等等。你會發(fā)現(xiàn)你無法使用 Future 創(chuàng)建這樣的一個工作流。

不能組合多個 Future 的結(jié)果

假設(shè)你有10個不同的Future，你想并行的運行，然后在它們運行未完成后運行一些函數(shù)。你會發(fā)現(xiàn)你也無法使用 Future 這樣做。

沒有異常處理

Future API 沒有任務(wù)的異常處理結(jié)構(gòu)居然有如此多的限制，幸好我們有CompletableFuture，你可以使用 CompletableFuture 達(dá)到以上所有目的。

CompletableFuture 實現(xiàn)了 Future 和 CompletionStage 接口，并且提供了許多關(guān)于創(chuàng)建，鏈?zhǔn)秸{(diào)用和組合多個 Future 的便利方法集，而且有廣泛的異常處理支持。

CompletableFuture的應(yīng)用場景

執(zhí)行比較耗時的操作時，尤其是那些依賴一個或多個遠(yuǎn)程服務(wù)的操作，使用異步任務(wù)可以改善程序的性能，加快程序的響應(yīng)速度。
使用CompletableFuture類，它提供了異常管理的機制，讓你有機會拋出、管理異步任務(wù)執(zhí)行種發(fā)生的異常。
如果這些異步任務(wù)之間相互獨立，或者他們之間的的某一些的結(jié)果是另一些的輸入，你可以講這些異步任務(wù)構(gòu)造或合并成一個。

CompletableFuture設(shè)計思想

CompletableFuture 按照類似“觀察者模式”的設(shè)計思想，原理分析可以從“觀察者”和“被觀察者”兩個方面著手。

由于回調(diào)種類多，但結(jié)構(gòu)差異不大，所以這里單以一元依賴中的thenApply為例，不再枚舉全部回調(diào)類型，如下圖所示：

被觀察者

每個CompletableFuture都可以被看作一個被觀察者，其內(nèi)部有一個Completion類型的鏈表成員變量stack，用來存儲注冊到其中的所有觀察者。當(dāng)被觀察者執(zhí)行完成后會彈棧stack屬性，依次通知注冊到其中的觀察者。上面例子中步驟fn2就是作為觀察者被封裝在UniApply中。
被觀察者CF中的result屬性，用來存儲返回結(jié)果數(shù)據(jù)。這里可能是一次RPC調(diào)用的返回值，也可能是任意對象，在上面的例子中對應(yīng)步驟fn1的執(zhí)行結(jié)果。

觀察者

CompletableFuture支持很多回調(diào)方法，例如thenAccept、thenApply、exceptionally等，這些方法接收一個函數(shù)類型的參數(shù)f，生成一個Completion類型的對象（即觀察者），并將入?yún)⒑瘮?shù)f賦值給Completion的成員變量fn，然后檢查當(dāng)前CF是否已處于完成狀態(tài)（即result!=null），如果已完成直接觸發(fā)fn，否則將觀察者Completion加入到CF的觀察者鏈stack中，再次嘗試觸發(fā)，如果被觀察者未執(zhí)行完則其執(zhí)行完畢之后通知觸發(fā)。

觀察者中的dep屬性：指向其對應(yīng)的CompletableFuture，在上面的例子中dep指向CF2。
觀察者中的src屬性：指向其依賴的CompletableFuture，在上面的例子中src指向CF1。
觀察者Completion中的fn屬性：用來存儲具體的等待被回調(diào)的函數(shù)。這里需要注意的是不同的回調(diào)方法（thenAccept、thenApply、exceptionally等）接收的函數(shù)類型也不同，即fn的類型有很多種，在上面的例子中fn指向fn2。

CompletableFuture核心功能

CompletableFuture的功能主要體現(xiàn)在它的CompletionStage,如下圖所示：

CompletionStage接口定義了任務(wù)編排的方法，執(zhí)行某一階段，可以向下執(zhí)行后續(xù)階段，可以實現(xiàn)如下功能：

轉(zhuǎn)換（thenCompose）
組合（thenCombine）
消費（thenAccept）
運行（thenRun）
帶返回的消費（thenApply）

具體其他功能大家可以根據(jù)需求自行查看。

CompletableFuture創(chuàng)建使用

創(chuàng)建CompletableFuture對象，提供了四個靜態(tài)方法用來創(chuàng)建CompletableFuture對象：

public static CompletableFuture<Void>   runAsync(Runnable runnable)
public static CompletableFuture<Void>   runAsync(Runnable runnable, Executor executor)
public static <U> CompletableFuture<U>  supplyAsync(Supplier<U> supplier)
public static <U> CompletableFuture<U>  supplyAsync(Supplier<U> supplier, Executor executor)

Asynsc 表示異步，而 supplyAsync 與 runAsync 不同在與前者異步返回一個結(jié)果，后者是 void。第二個函數(shù)第二個參數(shù)表示是用我們自己創(chuàng)建的線程池，否則采用默認(rèn)的 ForkJoinPool.commonPool() 作為它的線程池.其中Supplier是一個函數(shù)式接口，代表是一個生成者的意思，傳入0個參數(shù)，返回一個結(jié)果。

CompletableFuture<String> future = CompletableFuture.supplyAsync(()->{
    return "hello world";
});
//阻塞的獲取結(jié)果  ''helllo world"
System.out.println(future.get());

CompletableFuture用法詳解

沒有返回值的異步任務(wù)

如果你想異步的運行一個后臺任務(wù)并且不需要任務(wù)返回結(jié)果，就可以使用 runAsync()。

runAsync() 返回一個新的 CompletableFuture，它在運行給定操作后由在 ForkJoinPool.commonPool() 運行的任務(wù)異步完成。

/**
 * 沒有返回值的異步任務(wù)
 */
@Test
public void runAsync() throws Exception {
    CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
        log.info("Current thread name: {}", Thread.currentThread().getName());
        // 模擬長時間運行的作業(yè)
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            throw new IllegalStateException(e);
        };
    });
    // 主線程阻塞
    future.get();
    System.out.println("主線程結(jié)束");
}

有返回值的異步任務(wù)

當(dāng)運行一個異步任務(wù)并且需要有返回結(jié)果時，就可以使用 supplyAsync()。

CompletableFuture.supplyAsync() 它持有 supplier<T> 并且返回 CompletableFuture<T>，T 是通過調(diào)用傳入的 supplier 取得的值的類型。

CompletableFuture<String> future = CompletableFuture.supplyAsync(new Supplier<String>() {
    @Override
    public String get() {
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            throw new IllegalStateException(e);
        }
        return "Success";
    }
});
System.out.println(future.get());

Supplier<T> 是一個簡單的函數(shù)式接口，表示 supplier 的結(jié)果。它有一個 get()，該方法可以寫入你的后臺任務(wù)中，并且返回結(jié)果。

還可以使用 lambda 表達(dá)式使得上面的示例更加簡明：

/**
 * 有返回值的異步任務(wù)
 */
@Test
public void supplyAsync() throws Exception {
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        log.info("Current thread name: {}", Thread.currentThread().getName());
        // 模擬長時間運行的作業(yè)
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            throw new IllegalStateException(e);
        };
        return "Success";
    });
    // 主線程阻塞
    System.out.println(future.get());
}

上述 runAsync() 和 supplyAsync() 都是在單獨的線程中執(zhí)行他們的任務(wù)，但在實際業(yè)務(wù)中我們不會只創(chuàng)建一個線程。

自定義線程池執(zhí)行方法

CompletableFuture 可以從全局的 ForkJoinPool.commonPool() 獲得一個線程中執(zhí)行這些任務(wù)。但也可以創(chuàng)建一個線程池并傳給 runAsync() 和 supplyAsync() 來讓他們從線程池中獲取一個線程執(zhí)行它們的任務(wù)。

CompletableFuture API 的所有方法都有兩個變體-一個接受Executor作為參數(shù)，另一個不這樣：

static CompletableFuture<Void>  runAsync(Runnable runnable)
static CompletableFuture<Void>  runAsync(Runnable runnable, Executor executor)
static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

創(chuàng)建一個線程池，并傳遞給其中一個方法：

Executor executor = Executors.newFixedThreadPool(10);
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    try {
        TimeUnit.SECONDS.sleep(1);
    } catch (InterruptedException e) {
        throw new IllegalStateException(e);
    }
    return "Success";
}, executor);

線程串行化

由于 CompletableFuture.get() 方法是阻塞的，它會一直等到 Future 完成，并且在完成后返回結(jié)果。但是，這是我們想要的嗎？對于構(gòu)建異步系統(tǒng)，我們應(yīng)該附上一個回調(diào)給 CompletableFuture，當(dāng) Future 完成的時候，自動的獲取結(jié)果。

如果不想等待結(jié)果返回，就可以把需要等待 Future 完成執(zhí)行的邏輯寫入到回調(diào)函數(shù)中。

可以使用 thenApply()、thenAccept() 、thenRun() 回調(diào)給 CompletableFuture。

thenApply()

當(dāng)一個線程依賴另一個線程時，可以使用 thenApply() 來把這兩個線程串行化。

thenApply：可以使用 thenApply() 處理和改變 CompletableFuture 的結(jié)果。

/**
 * 使用 thenApply() 處理和改變CompletableFuture的結(jié)果
 *
 * @throws Exception
 */
@Test
public void thenApply1() throws Exception {
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        return "Java";
    }).thenApply(o -> {
        return "Hello " + o;
    });
    System.out.println(future.get());
}

thenAccept()

如果你不想從你的回調(diào)函數(shù)中返回任何東西，僅僅想在 Future 完成后運行一些代碼片段，你可以使用 thenAccept() 和 thenRun()，這些方法經(jīng)常在調(diào)用鏈的最末端的最后一個回調(diào)函數(shù)中使用。

thenAccept：消費處理結(jié)果，接收任務(wù)的處理結(jié)果，并消費處理，無返回結(jié)果。

@Test
public void thenAccept() throws Exception {
    CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
        // 模擬長時間運行的作業(yè)
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            throw new IllegalStateException(e);
        }
        return "Success";
    }).thenAccept(o -> {
        if ("Success".equals(o)) {
            // 模擬長時間運行的作業(yè)
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                throw new IllegalStateException(e);
            }
        }
    });
    future.get();
    log.info("======================");
    System.out.println("結(jié)束.");
}

thenRun()

thenRun() 不能訪 Future 的結(jié)果，它持有一個 Runnable 返回 CompletableFuture：

@Test
public void thenRun() throws Exception {
    CompletableFuture<Void> future = CompletableFuture.supplyAsync(() -> {
        // 模擬長時間運行的作業(yè)
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            throw new IllegalStateException(e);
        }
        return "Success";
    }).thenRun(() -> {
        // 作業(yè)完成后執(zhí)行一些代碼片段
        System.out.println("thenRun 執(zhí)行一些代碼片段");
    });
    future.get();
}

結(jié)果合并

thenCompose（）

使用 thenCompose() 合并兩個有依賴關(guān)系的 CompletableFutures 的執(zhí)行結(jié)果。

private static Integer num = 10;
@Test
public void thenCompose() throws Exception {
    //第一步加 10
    CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
        System.out.println("讓num+10;任務(wù)開始");
        num += 10;
        return num;
    });
    //合并
    CompletableFuture<Integer> future1 = future.thenCompose(i ->
            //再來一個 CompletableFuture
            CompletableFuture.supplyAsync(() -> i + 1)
    );
    System.out.println(future.get());
    System.out.println(future1.get());
}

thenCombine()

使用 thenCombine() 組合兩個獨立的 future。當(dāng)兩個獨立的 Future 都完成的時候使用 thenCombine() 用來做一些事情。

@Test
public void thenCompose() throws Exception {
    // 長方形:S=長*寬
    //第一步加 10
    CompletableFuture<Integer> lengthFuture = CompletableFuture.supplyAsync(() -> {
        return 50;
    });
    CompletableFuture<Integer> widthFuture = CompletableFuture.supplyAsync(() -> {
        return 30;
    });
    CompletableFuture<Integer> combinedFuture = lengthFuture.thenCombine(widthFuture, (t1, t2) -> {
        System.out.println(t1);
        System.out.println(t2);
        return t1 * t2;
    });
    System.out.println(combinedFuture.get());
}

合并多個任務(wù)的結(jié)果

使用 thenCompose() 和 thenCombine() 可以把兩個 CompletableFuture 組合在一起。如果要是想組合任意數(shù)量的 CompletableFuture，應(yīng)該怎么做？

可以使用 allOf 和 anyOf 組合任意多個 CompletableFuture。這兩個函數(shù)都是靜態(tài)函數(shù)，參數(shù)是變長的 CompletableFuture 的集合。

allOf 和 anyOf 的區(qū)別，前者是「與」，后者是「或」。

allOf()

allOf 的返回值是 CompletableFuture<Void> 類型，這是因為每個傳入的 CompletableFuture 的返回值都可能不同，所以組合的結(jié)果是無法用某種類型來表示的，索性返回 Void 類型。那么，如何獲取每個 CompletableFuture 的執(zhí)行結(jié)果呢？

例子：并行地下載 N 個資源，待下載完成之后，并資源額外處理。

@Test
public void allOf() throws Exception {
    // URL 列表集合
    List<String> webLinks = Arrays.asList("https://www.baidu.com/", "https://www.bing.com/", "https://www.so.com/");
    // 并行下載多個網(wǎng)頁
    List<CompletableFuture<String>> contentFutureList = webLinks.stream().map(webLink -> downloadWebLink(webLink)).collect(Collectors.toList());
    // 通過allof,等待所有網(wǎng)頁下載完畢，收集返回結(jié)果
    CompletableFuture<Void> allFutures = CompletableFuture.allOf(contentFutureList.toArray(new CompletableFuture[contentFutureList.size()]));
    // 附上回調(diào)函數(shù)，獲取結(jié)果集
    // 方式一
    CompletableFuture<List<String>> result = allFutures.thenApply(v -> contentFutureList.stream().map(o -> {
        try {
            return o.get();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        }
    }).collect(Collectors.toList()));
    System.out.println(result.get());
    // 方式二
    result = allFutures.thenApply(v -> contentFutureList.stream().map(CompletableFuture<String>::join).collect(Collectors.toList()));
    System.out.println(result.get());
}
private CompletableFuture<String> downloadWebLink(String webLink) {
    return CompletableFuture.supplyAsync(() -> {
        // 模擬下載過程
        System.out.println("開始下載網(wǎng)頁：" + webLink);
        return "這是一個網(wǎng)頁內(nèi)容";
    });
}

這里有個關(guān)鍵問題，因為 allof 沒有返回值，所以通過 theApply，給 allFutures 附上一個回調(diào)函數(shù)。在回調(diào)函數(shù)里面，以此調(diào)用么一個 Future 的 Get() 函數(shù)，獲取結(jié)果后存入 List<String> 中。

anyOf()

anyOf 的含義是只要有任意一個 CompletableFuture 結(jié)束，就可以做接下來的事情，而無須像 allOf 那樣，等待所有的 CompletableFuture 結(jié)束。

但由于每個 CompletableFuture 的返回值類型可能不同，意味著無法判斷是什么類型，所以 anyOf 的返回值是 CompletableFuture<Object> 類型。

@Test
public void anyOf() throws Exception {
    CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
        try {
            TimeUnit.SECONDS.sleep(3);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return "future1 結(jié)果";
    });
    CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
        try {
            TimeUnit.SECONDS.sleep(4);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return "future2 結(jié)果";
    });
    CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return "future3 結(jié)果";
    });
    CompletableFuture<Object> future = CompletableFuture.anyOf(future1, future2, future3);
    System.out.println(future.get());
}

在該例子中，因為 future1、future2、future3 的返回值都是 CompletableFuture<String>，所以anyOf 的返回的 Object 一定也是 String 類型。

并且在 3 個 future 中，future3 睡眠時間最短，會最先執(zhí)行完成， anyOfFuture.get() 獲取的也就是 future3 的內(nèi)容。future1、future2 的返回結(jié)果會被丟棄。

異常處理

在調(diào)用 supplyAsync() 任務(wù)中發(fā)生一個錯誤，這時候沒有任何 thenApply 會被調(diào)用并且 future 將以一個異常結(jié)束。如果在第一個 thenApply() 發(fā)生錯誤，這時候第二個和第三個將不會被調(diào)用，同樣的，future 將以異常結(jié)束。

exceptionally()

回調(diào)處理異常，從原始Future中生成的錯誤恢復(fù)的機會。

@Test
public void exceptionally() throws Exception {
    Integer age = -1;
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        if (age < 0) {
            throw new IllegalArgumentException("年齡不能為負(fù)數(shù)");
        }
        return "張三";
    }).exceptionally(ex -> {
        System.out.println(ex.getMessage());
        return "Unknown!";
    });
    System.out.println(future.get());
}

handle()

從異?；謴?fù)，無論一個異常是否發(fā)生它都會被調(diào)用。

@Test
public void handle() throws Exception {
    Integer age = -1;
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        if (age < 0) {
            throw new IllegalArgumentException("年齡不能為負(fù)數(shù)");
        }
        return "張三";
    }).handle((res, ex) -> {
        if (ex != null) {
            System.out.println(ex.getMessage());
            return "Unknown!";
        }
        return res;
    });
    System.out.println(future.get());
}

如果異常發(fā)生 res 參數(shù)將是 null，否則 ex 將是 null。

參考：CompletableFuture原理與用法詳解

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

您可能感興趣的文章: