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