Java CompletableFuture 实战:异步编排、组合与那些让你线上翻车的坑

名字还没想好☜

185人浏览 · 2026-07-08 09:04:44

名字还没想好☜ · 2026-07-08 09:04:44 发布

Java CompletableFuture 实战:异步编排、组合与那些让你线上翻车的坑

很多人对异步的印象还停留在「new Thread + Future.get()」。但 Future.get() 会阻塞,一旦要「查完用户再并行查订单和积分,最后合并返回」这种编排,代码立刻变成一堆嵌套的线程池和阻塞等待,又丑又容易出错。CompletableFuture(JDK 8 起)就是来解决这件事的:它让你像串珠子一样把异步任务链起来,不用手动阻塞。

这篇从「串行链式」讲到「并行合并」,再到三个真实会让你线上翻车的坑。

起步:异步跑一个任务,拿到结果继续处理

最基础的两个方法:supplyAsync(有返回值)和 runAsync(无返回值)。拿到结果后用 thenApply 做转换:

import java.util.concurrent.CompletableFuture;

CompletableFuture<String> future = CompletableFuture
    // supplyAsync 在线程池里异步执行,返回一个值
    .supplyAsync(() -> {
        // 模拟一次远程调用
        return queryUserName(123L);   // 返回 "Alice"
    })
    // thenApply:上一步的结果拿来做转换,仍在异步链里
    .thenApply(name -> "Hello, " + name);

// 这里主线程不阻塞,需要结果时才 join
System.out.println(future.join());  // Hello, Alice

thenApply 和 thenApplyAsync 的区别:前者可能复用上一步的线程执行回调,后者一定提交到线程池。回调里有耗时操作时用 Async 版本,避免占用不该占用的线程。

thenApply vs thenCompose:别把 Future 套进 Future

这是新手最常踩的第一个坑。假设你有一个方法本身就返回 CompletableFuture:

CompletableFuture<Long> getUserId(String name) { ... }
CompletableFuture<User>  getUserDetail(Long id) { ... }

想「先拿 id,再用 id 查详情」,如果用 thenApply:

// 错误:类型变成了 CompletableFuture<CompletableFuture<User>>,嵌套了两层!
CompletableFuture<CompletableFuture<User>> bad =
    getUserId("Alice").thenApply(id -> getUserDetail(id));

因为 getUserDetail 本身返回一个 future,thenApply 又包了一层,结果就套娃了。正确做法是 thenCompose(相当于其他语言的 flatMap),它会把里层的 future 拍平:

// 正确:thenCompose 拍平,结果是 CompletableFuture<User>
CompletableFuture<User> good =
    getUserId("Alice").thenCompose(id -> getUserDetail(id));

记忆点:回调返回普通值用 thenApply,回调返回 CompletableFuture 用 thenCompose。

并行合并:两个任务同时跑,都好了再汇总

真实场景:进详情页要同时查「商品信息」和「库存」,两个互不依赖,应该并行。用 thenCombine 合并两个 future 的结果:

CompletableFuture<Product> productF =
    CompletableFuture.supplyAsync(() -> queryProduct(1L));

CompletableFuture<Integer> stockF =
    CompletableFuture.supplyAsync(() -> queryStock(1L));

// 两个都完成后,把结果合并成一个视图对象
CompletableFuture<ProductView> viewF = productF.thenCombine(stockF,
    (product, stock) -> new ProductView(product, stock));

ProductView view = viewF.join();

两个 supplyAsync 一提交就同时在线程池里跑,总耗时约等于两者中较慢的那个,而不是相加。

如果是 N 个任务要等全部完成,用 allOf:

CompletableFuture<Void> all = CompletableFuture.allOf(productF, stockF /*, ... */);
// allOf 本身返回 Void,结果要从各自的 future 里取
all.join();
Product p = productF.join();   // 此时已完成,不会阻塞
Integer s = stockF.join();

异常处理:别让一个失败吞掉整条链

异步链里任何一步抛异常,后续的 thenApply 会被跳过。处理异常有三个方法,分清用途:

CompletableFuture<String> f = CompletableFuture
    .supplyAsync(() -> {
        if (Math.random() > 0.5) throw new RuntimeException("查询失败");
        return "ok";
    })
    // exceptionally:只在出异常时兜底,给个默认值
    .exceptionally(ex -> {
        log.warn("降级处理: {}", ex.getMessage());
        return "default";
    });

// handle:不管成功失败都会进来,两个参数一个是结果一个是异常
CompletableFuture<String> f2 = someFuture.handle((result, ex) -> {
    if (ex != null) return "error";
    return result.toUpperCase();
});

// whenComplete:只观察不改变结果,常用于打日志/清理,异常仍会往下传
someFuture.whenComplete((result, ex) -> {
    if (ex != null) log.error("任务失败", ex);
});

区别:exceptionally 只管异常分支;handle 成功失败都管且能改结果;whenComplete 只旁观不改结果。

三个线上翻车的坑

坑一:不指定线程池,默认全挤在 ForkJoinPool.commonPool。 supplyAsync 不传线程池时用的是全 JVM 共享的 commonPool,它的线程数默认是「CPU 核数 - 1」。如果你的任务是 IO 密集(查库、调接口),几个慢请求就能把这个池占满,拖垮整个应用所有异步任务。生产环境务必传自定义线程池:

ExecutorService ioPool = Executors.newFixedThreadPool(20);
CompletableFuture.supplyAsync(() -> queryUser(id), ioPool);  // 第二个参数指定池

坑二:get() 不设超时会永久阻塞。 future.get() 无参版本会一直等。下游卡死时你的线程也永远醒不过来。要么用 get(timeout, unit),要么直接用 JDK 9+ 的 orTimeout:

future.orTimeout(3, TimeUnit.SECONDS)   // 3 秒没完成就以 TimeoutException 结束
      .exceptionally(ex -> "timeout fallback");

坑三:异常在链中间被「静默吞掉」。 如果你只 supplyAsync 后就不管了,任务里抛的异常不会打印到任何地方,像人间蒸发。链末尾一定要接 exceptionally 或 whenComplete 记录异常,否则排查问题时你会一脸懵。

小结

CompletableFuture 用链式回调替代阻塞式 Future.get(),thenApply 做转换,thenCompose 拍平嵌套的 future(回调返回 future 就用它)。
并行合并两个任务用 thenCombine,等一批全完成用 allOf(结果要各自 join 取)。
异常处理三选一:exceptionally(只兜底异常)、handle(成功失败都处理并改结果)、whenComplete(只旁观不改)。
生产三条铁律:必须传自定义线程池(别用 commonPool)、get/超时兜底(orTimeout)、链尾必接异常处理(别让异常静默消失)。