CompletableFuture 异步任务编排详解
1. 异步编程的痛点与 CompletableFuture 的定位
在 Java 中，传统的 Future 虽然能够把任务放到后台执行，但调用 get() 时会阻塞当前线程；而纯回调方式又容易陷入层层嵌套的"回调地狱"。CompletableFuture 自 Java 8 引入，提供了一套声明式 API，让异步任务可以像流水线一样组合、转换和异常恢复，真正实现非阻塞编程。
2. 创建并执行一个异步任务
最常用的创建方式是 CompletableFuture.supplyAsync(Supplier)，用于返回结果；若只需要执行副作用，可使用 runAsync(Runnable)。下面模拟异步查询商品价格：
import java.util.concurrent.CompletableFuture;
public class AsyncPriceDemo {
public static void main(String[] args) throws Exception {
CompletableFuture<String> priceFuture = CompletableFuture.supplyAsync(() -> queryPrice());
// 非阻塞回调
priceFuture.thenAccept(value -> System.out.println("商品价格：" + value));
System.out.println("主线程继续处理其他逻辑");
// 等待异步任务完成，防止 JVM 过早退出
priceFuture.join();
}
private static String queryPrice() {
sleep(600);
return "¥199";
}
private static void sleep(int millis) {
try {
Thread.sleep(millis);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
运行后可以看到，主线程的打印先于异步结果输出，说明当前线程没有被阻塞。
3. 任务的链式处理与组合
3.1 转换结果：thenApply
thenApply 接收上一个阶段的结果并返回新的类型，适合数据转换、简单计算等场景。
CompletableFuture<Integer> scoreFuture = CompletableFuture
.supplyAsync(() -> 85)
.thenApply(base -> base + 10) // 加分
.thenApply(total -> total * 2); // 翻倍
scoreFuture.thenAccept(result -> System.out.println("最终得分：" + result));
3.2 链式调用：thenCompose
当后续操作本身也是异步时，应使用 thenCompose 将两个异步阶段串起来，避免返回 CompletableFuture<CompletableFuture<T>> 的嵌套结构。
CompletableFuture<String> userProfile = CompletableFuture
.supplyAsync(() -> 1001)
.thenCompose(AsyncPriceDemo::fetchUserById)
.thenCompose(AsyncPriceDemo::fetchDepartmentName);
userProfile.thenAccept(System.out::println);
private static CompletableFuture<String> fetchUserById(int id) {
return CompletableFuture.supplyAsync(() -> "User:" + id);
}
private static CompletableFuture<String> fetchDepartmentName(String user) {
return CompletableFuture.supplyAsync(() -> user + "->Engineering");
}
3.3 合并两个独立任务：thenCombine
thenCombine 用于等待两个并行任务都完成后再合并它们的结果。
CompletableFuture<Integer> discountFuture = CompletableFuture.supplyAsync(() -> 30);
CompletableFuture<Integer> amountFuture = CompletableFuture.supplyAsync(() -> 200);
discountFuture
.thenCombine(amountFuture, (discount, amount) -> amount - discount)
.thenAccept(finalPrice -> System.out.println("实付金额：" + finalPrice));
4. 异常处理
4.1 exceptionally：异常时返回兜底值
CompletableFuture<String> safeFuture = CompletableFuture
.supplyAsync(() -> {
throw new RuntimeException("下游服务超时");
})
.exceptionally(ex -> {
System.out.println("捕获异常：" + ex.getMessage());
return "默认响应";
});
safeFuture.thenAccept(System.out::println);
4.2 handle：统一处理成功与失败
handle 不论任务成功还是失败都会执行，通过判断异常参数是否存在来做出相应处理。
CompletableFuture<String> handledFuture = CompletableFuture
.supplyAsync(() -> {
if (true) {
throw new IllegalStateException("数据库连接失败");
}
return "业务数据";
})
.handle((data, ex) -> ex != null ? "兜底数据" : data);
handledFuture.thenAccept(System.out::println);
5. 多任务聚合：allOf 与 anyOf
5.1 allOf：等待全部完成
CompletableFuture<String> inventoryTask = CompletableFuture.supplyAsync(() -> "库存查询完成");
CompletableFuture<String> logisticsTask = CompletableFuture.supplyAsync(() -> "物流查询完成");
CompletableFuture<String> reviewTask = CompletableFuture.supplyAsync(() -> "评价查询完成");
CompletableFuture<Void> allTasks = CompletableFuture.allOf(inventoryTask, logisticsTask, reviewTask);
allTasks.thenRun(() -> System.out.println("所有查询已结束"));
5.2 anyOf：任意一个完成即返回
CompletableFuture<Object> anyTask = CompletableFuture.anyOf(inventoryTask, logisticsTask);
anyTask.thenAccept(result -> System.out.println("最先返回的结果：" + result));
6. 实践建议
- 为 IO 密集型任务指定自定义线程池，避免长期占用
ForkJoinPool.commonPool()。 - 在回调链中尽量避免调用
get()或join()，以免破坏异步非阻塞的优势。 - 每个异步阶段都应通过
exceptionally或handle处理异常，防止异常被静默吞掉。 - 需要多个任务结果时，优先使用
thenCombine、allOf等组合 API，而不是手动轮询。
