在現(xiàn)代軟件開發(fā)中，異步編程已成為提升系統(tǒng)性能、響應(yīng)能力和可擴(kuò)展性的關(guān)鍵手段。Java 8 引入了 CompletableFuture 類，為 Java 平臺(tái)帶來了強(qiáng)大的異步編程能力。

本篇文章將帶你認(rèn)識(shí)這個(gè)異步編程神器：CompletableFuture。

什么是 CompletableFuture

CompletableFuture 是 Java 8 引入的 java.util.concurrent 包下的一個(gè)類，它代表一個(gè)異步計(jì)算的結(jié)果，可以是已完成、正在進(jìn)行或尚未開始。CompletableFuture 提供了一種靈活、類型安全的方式來表達(dá)異步操作的生命周期，包括創(chuàng)建、組合、處理結(jié)果以及處理異常。其設(shè)計(jì)靈感來源于函數(shù)式編程中的 Promises/Futures 模式，旨在簡(jiǎn)化異步編程模型，提高代碼的可讀性和可維護(hù)性。

CompletableFuture 的核心功能

1. 創(chuàng)建 CompletableFuture

a. completedFuture(T value)

completedFuture(T value) 是一個(gè)靜態(tài)工廠方法，用于創(chuàng)建一個(gè)已經(jīng)處于完成狀態(tài)且包含給定結(jié)果值的 CompletableFuture。這適用于預(yù)先計(jì)算好的結(jié)果或常量值，使得其他組件可以以異步形式消費(fèi)。

b. supplyAsync(Supplier<U> supplier, Executor executor)

supplyAsync() 方法接受一個(gè) Supplier 函數(shù)和一個(gè)可選的 Executor，異步執(zhí)行 supplier.get()，并將結(jié)果封裝到一個(gè)新的 CompletableFuture 中。計(jì)算在 Executor 管理的線程中進(jìn)行，不阻塞當(dāng)前線程。

c. runAsync(Runnable runnable, Executor executor)

類似于 supplyAsync()，runAsync() 接受一個(gè) Runnable 任務(wù)和一個(gè) Executor，異步執(zhí)行任務(wù)。由于 Runnable 沒有返回值，runAsync() 返回的 CompletableFuture 完成時(shí)沒有結(jié)果。

2. 組合 CompletableFuture

a. thenApply(Function<? super T,? extends U> fn)

在當(dāng)前 CompletableFuture 完成后，應(yīng)用給定的 Function 處理結(jié)果，并返回一個(gè)新的 CompletableFuture，其結(jié)果為 Function 應(yīng)用后的值。

b. thenAccept(Consumer<? super T> action)

當(dāng)當(dāng)前 CompletableFuture 完成后，執(zhí)行給定的 Consumer 消費(fèi)結(jié)果。由于 Consumer 沒有返回值，返回的 CompletableFuture 完成時(shí)沒有結(jié)果。

c. thenCombine(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)

當(dāng)當(dāng)前 CompletableFuture 與另一個(gè) CompletionStage（如另一個(gè) CompletableFuture）都完成時(shí)，應(yīng)用給定的 BiFunction 合并兩個(gè)結(jié)果，并返回一個(gè)新的 CompletableFuture。

3. 異常處理

a. exceptionally(Function<Throwable,? extends T> fn)

當(dāng)當(dāng)前 CompletableFuture 因異常而未能正常完成時(shí)，應(yīng)用給定的 Function 處理異常，并返回一個(gè)新的 CompletableFuture，其結(jié)果為 Function 應(yīng)用后的值。

b. handle(BiFunction<? super T, Throwable, ? extends U> fn)

無論當(dāng)前 CompletableFuture 正常完成還是因異常未能完成，都會(huì)應(yīng)用給定的 BiFunction 處理結(jié)果或異常，并返回一個(gè)新的 CompletableFuture。

4. 其他重要方法

a. allOf(CompletableFuture<?>... cfs)

創(chuàng)建一個(gè)新的 CompletableFuture，當(dāng)所有給定的 CompletableFuture 都完成（不論成功與否）時(shí)，新 CompletableFuture 完成。

b. anyOf(CompletableFuture<?>... cfs)

創(chuàng)建一個(gè)新的 CompletableFuture，當(dāng)任意一個(gè)給定的 CompletableFuture 完成時(shí)，新 CompletableFuture 完成。

實(shí)戰(zhàn)應(yīng)用

CompletableFuture 的使用場(chǎng)景很廣泛，例如

1. 異步數(shù)據(jù)庫查詢與結(jié)果合并
2. 微服務(wù)間異步通信
3. 并行任務(wù)執(zhí)行與結(jié)果匯總
4. 異步事件處理與通知

這里以第一個(gè)場(chǎng)景舉例：場(chǎng)景：在一個(gè)訂單處理系統(tǒng)中，需要查詢訂單的詳細(xì)信息、關(guān)聯(lián)的商品信息以及用戶的個(gè)人信息。為減少查詢延遲，可以使用 CompletableFuture 對(duì)每個(gè)查詢進(jìn)行異步執(zhí)行，并在所有查詢完成后合并結(jié)果。示例：

如果我們不使用Java8提供的這個(gè)CompletableFuture 來實(shí)現(xiàn)

@Service
@RequiredArgsConstructor
public class OrderProcessingServiceLegacy {
    private final OrderRepository orderRepo;
    private final ProductRepository productRepo;
    private final UserRepository userRepo;

    public OrderDetails fetchOrderDetails(int orderId) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(3);
        CountDownLatch orderLatch = new CountDownLatch(1);
        CountDownLatch productsLatch = new CountDownLatch(1);
        CountDownLatch userLatch = new CountDownLatch(1);

        Order order = null;
        List<Product> products = null;
        User user = null;

        // 異步查詢訂單
        executor.execute(() -> {
            try {
                order = orderRepo.findOrderById(orderId);
                orderLatch.countDown();
            } finally {
                productsLatch.countDown();
            }
        });

        // 異步查詢商品
        executor.execute(() -> {
            try {
                products = productRepo.findProductsByOrderId(orderId);
            } finally {
                productsLatch.countDown();
            }
        });

        // 異步查詢用戶（等待訂單查詢完成后再執(zhí)行）
        executor.execute(() -> {
            try {
                orderLatch.await(); // 確保訂單查詢已完成
                user = userRepo.findUserById(order.getCustomerId());
            } finally {
                userLatch.countDown();
            }
        });

        // 等待所有查詢完成
        userLatch.await();

        return new OrderDetails(order, products, user);
    }

    // ... 其他方法 ...
 @Data
 @AllArgsConstructor 
    public static class OrderDetails {
        private final Order order;
        private final List<Product> products;
        private final User user;
    }
}

使用CompletableFuture實(shí)現(xiàn)

@Service
@RequiredArgsConstructor
public class OrderProcessingService {
    private final OrderRepository orderRepo;
    private final ProductRepository productRepo;
    private final UserRepository userRepo;
 private final ThreadPoolExecutor executor;


    public CompletableFuture<OrderDetails> fetchOrderDetails(int orderId) {

        CompletableFuture<Order> orderFuture = CompletableFuture.supplyAsync(() -> orderRepo.findOrderById(orderId), executor);
        CompletableFuture<List<Product>> productsFuture = CompletableFuture.supplyAsync(() -> productRepo.findProductsByOrderId(orderId), executor);
        CompletableFuture<User> userFuture = CompletableFuture.supplyAsync(() -> userRepo.findUserById(order.getCustomerId()), executor);

        return CompletableFuture.allOf(orderFuture, productsFuture, userFuture)
                .thenApplyAsync(unused -> {
                    Order order = orderFuture.join();
                    List<Product> products = productsFuture.join();
                    User user = userFuture.join();

                    return new OrderDetails(order, products, user);
                }, executor);
    }

    // ... 其他方法 ...
 @Data
    public static class OrderDetails {
        private final Order order;
        private final List<Product> products;
        private final User user;

    }
}

在這個(gè)示例中：

• 使用了CompletableFuture之后，代碼量減少了，整潔度和可讀性也得到提高。
• fetchOrderDetails 方法接受一個(gè)訂單 ID，使用 CompletableFuture.supplyAsync() 異步查詢訂單、商品和用戶信息。
• 使用 CompletableFuture.allOf() 監(jiān)控所有查詢的完成狀態(tài)。
• 當(dāng)所有查詢完成時(shí)，使用 thenApplyAsync() 合并結(jié)果，創(chuàng)建一個(gè)包含完整訂單詳情的 OrderDetails 對(duì)象。

小結(jié)

CompletableFuture 作為 Java 8 引入的重要異步編程工具，極大地提升了 Java 平臺(tái)在應(yīng)對(duì)高并發(fā)、高性能場(chǎng)景的能力。結(jié)合 Java 8 的并行流（Stream.parallel()）與 CompletableFuture，可以輕松實(shí)現(xiàn)數(shù)據(jù)集的并行處理和結(jié)果聚合。下次給大家聊聊Stream.parallel()。