CompletableFuture：Java 8 中的異步編程利器
在現代軟件開發中，異步編程已成為提升系統性能、響應能力和可擴展性的關鍵手段。Java 8 引入了 CompletableFuture 類，為 Java 平臺帶來了強大的異步編程能力。
本篇文章將帶你認識這個異步編程神器：CompletableFuture。
什么是 CompletableFuture
CompletableFuture 是 Java 8 引入的 java.util.concurrent 包下的一個類，它代表一個異步計算的結果，可以是已完成、正在進行或尚未開始。CompletableFuture 提供了一種靈活、類型安全的方式來表達異步操作的生命周期，包括創建、組合、處理結果以及處理異常。其設計靈感來源于函數式編程中的 Promises/Futures 模式，旨在簡化異步編程模型，提高代碼的可讀性和可維護性。
CompletableFuture 的核心功能
1. 創建 CompletableFuture
a. completedFuture(T value)
completedFuture(T value) 是一個靜態工廠方法，用于創建一個已經處于完成狀態且包含給定結果值的 CompletableFuture。這適用于預先計算好的結果或常量值，使得其他組件可以以異步形式消費。
b. supplyAsync(Supplier<U> supplier, Executor executor)
supplyAsync() 方法接受一個 Supplier 函數和一個可選的 Executor，異步執行 supplier.get()，并將結果封裝到一個新的 CompletableFuture 中。計算在 Executor 管理的線程中進行，不阻塞當前線程。
c. runAsync(Runnable runnable, Executor executor)
類似于 supplyAsync()，runAsync() 接受一個 Runnable 任務和一個 Executor，異步執行任務。由于 Runnable 沒有返回值，runAsync() 返回的 CompletableFuture 完成時沒有結果。
2. 組合 CompletableFuture
a. thenApply(Function<? super T,? extends U> fn)
在當前 CompletableFuture 完成后，應用給定的 Function 處理結果，并返回一個新的 CompletableFuture，其結果為 Function 應用后的值。
b. thenAccept(Consumer<? super T> action)
當當前 CompletableFuture 完成后，執行給定的 Consumer 消費結果。由于 Consumer 沒有返回值，返回的 CompletableFuture 完成時沒有結果。
c. thenCombine(CompletionStage<? extends U> other, BiFunction<? super T,? super U,? extends V> fn)
當當前 CompletableFuture 與另一個 CompletionStage（如另一個 CompletableFuture）都完成時，應用給定的 BiFunction 合并兩個結果，并返回一個新的 CompletableFuture。
3. 異常處理
a. exceptionally(Function<Throwable,? extends T> fn)
當當前 CompletableFuture 因異常而未能正常完成時，應用給定的 Function 處理異常，并返回一個新的 CompletableFuture，其結果為 Function 應用后的值。
b. handle(BiFunction<? super T, Throwable, ? extends U> fn)
無論當前 CompletableFuture 正常完成還是因異常未能完成，都會應用給定的 BiFunction 處理結果或異常，并返回一個新的 CompletableFuture。
4. 其他重要方法
a. allOf(CompletableFuture<?>... cfs)
創建一個新的 CompletableFuture，當所有給定的 CompletableFuture 都完成（不論成功與否）時，新 CompletableFuture 完成。
b. anyOf(CompletableFuture<?>... cfs)
創建一個新的 CompletableFuture，當任意一個給定的 CompletableFuture 完成時，新 CompletableFuture 完成。
實戰應用
CompletableFuture 的使用場景很廣泛，例如
- 異步數據庫查詢與結果合并
- 微服務間異步通信
- 并行任務執行與結果匯總
- 異步事件處理與通知
這里以第一個場景舉例：場景：在一個訂單處理系統中，需要查詢訂單的詳細信息、關聯的商品信息以及用戶的個人信息。為減少查詢延遲，可以使用 CompletableFuture 對每個查詢進行異步執行，并在所有查詢完成后合并結果。示例：
如果我們不使用Java8提供的這個CompletableFuture 來實現
@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();
}
});
// 異步查詢用戶（等待訂單查詢完成后再執行）
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實現
@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;
}
}在這個示例中：
- 使用了CompletableFuture之后，代碼量減少了，整潔度和可讀性也得到提高。
- fetchOrderDetails 方法接受一個訂單 ID，使用 CompletableFuture.supplyAsync() 異步查詢訂單、商品和用戶信息。
- 使用 CompletableFuture.allOf() 監控所有查詢的完成狀態。
- 當所有查詢完成時，使用 thenApplyAsync() 合并結果，創建一個包含完整訂單詳情的 OrderDetails 對象。
小結
CompletableFuture 作為 Java 8 引入的重要異步編程工具，極大地提升了 Java 平臺在應對高并發、高性能場景的能力。結合 Java 8 的并行流（Stream.parallel()）與 CompletableFuture，可以輕松實現數據集的并行處理和結果聚合。下次給大家聊聊Stream.parallel()。
