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,而不是手动轮询。