Java 中基于 CompletableFuture 的异步编排实战：并行调用、超时控制与异常兜底

在 Java 后端开发里，CompletableFuture 几乎是“异步编排”绕不过去的一站。

很多同学第一次接触它，通常是从 supplyAsync() 开始；但真正到了业务里，问题马上就变复杂了：

一个接口要并行查 3 个下游服务，怎么写得清晰？
某个依赖接口特别慢，怎么超时降级？
其中一个调用失败了，整体是不是就全挂？
线程池到底要不要自己配？
join()、get()、allOf()、handle() 看起来都能用，怎么选？

这篇文章我不打算只讲 API 列表，而是用一个更贴近线上场景的例子，带你从并行调用、超时控制、异常兜底一路搭起来，最后再说常见坑和性能建议。你看完后，应该能直接把思路搬到项目里。

背景与问题

假设我们要做一个“用户首页聚合接口”，前端只调一次 /home，后端内部要去拿：

用户基本信息
账户余额
优惠券列表

如果你按串行写法：

UserInfo userInfo = userService.getUserInfo(userId);
Balance balance = accountService.getBalance(userId);
List<Coupon> coupons = couponService.listCoupons(userId);
return new HomeDTO(userInfo, balance, coupons);

这段代码的问题很直接：

总耗时 = 各下游耗时之和
某个下游慢，整体就慢
某个下游抛异常，如果没处理好，整个接口直接失败
没有统一超时和降级策略

而聚合型接口最常见的目标其实是：

能并行就并行
关键字段优先保证
非关键字段超时后快速降级
部分失败不拖垮整体

这就是 CompletableFuture 的主战场。

前置知识与环境准备

本文示例基于以下前提：

JDK 9+ 更方便，因为有 orTimeout() / completeOnTimeout()
如果你是 JDK 8，也可以通过 ScheduledExecutorService 自己补超时逻辑
建议具备这些基础：
- Java 线程池基本概念
- Lambda 表达式
- Future 和阻塞等待的区别

为了避免示例过于抽象，我会写一个可直接运行的 Demo，用 sleep 模拟下游调用耗时和异常。

核心原理

1. CompletableFuture 不是“只是异步执行”，而是“异步结果的编排器”

很多人把它当成“线程池任务提交器”，这是不够的。

CompletableFuture 更重要的价值在于：
你不仅能启动异步任务，还能描述任务之间的依赖、合并、异常处理和超时策略。

比如：

supplyAsync()：异步产生结果
thenApply()：对结果做转换
thenCombine()：把两个结果合并
allOf()：等待多个任务完成
exceptionally()：失败兜底
handle()：统一处理成功/失败
orTimeout()：超时直接失败
completeOnTimeout()：超时给默认值

2. 并行调用的核心：拆分独立任务，再聚合结果

如果 3 个下游之间没有依赖关系，就应该并行。

flowchart LR
    A[收到首页请求] --> B[异步查用户信息]
    A --> C[异步查余额]
    A --> D[异步查优惠券]
    B --> E[聚合结果]
    C --> E
    D --> E
    E --> F[返回响应]

这样做的收益通常很明显：

串行：120ms + 180ms + 250ms = 550ms
并行：约等于 max(120ms, 180ms, 250ms) = 250ms 左右

当然，这是理想情况，真实环境还要考虑线程池、上下文切换、超时、网络抖动等因素。

3. 超时控制不是“可选项”，而是异步编排的边界

异步不是魔法。
如果某个下游一直不返回，主线程最终还是可能卡在 join() 或 get() 上。

所以我一般建议：

核心数据：超时后失败，或者返回明确错误
非核心数据：超时后降级为默认值

这两种策略分别对应：

orTimeout(timeout, unit)：超时抛异常
completeOnTimeout(defaultValue, timeout, unit)：超时返回默认值

4. 异常兜底要区分“吞掉异常”和“保留可观测性”

新手常见写法：

future.exceptionally(ex -> defaultValue);

它确实能兜底，但如果你不打日志、不带上下文，线上排查会非常痛苦。

我踩过这个坑：接口表面“都成功了”，但某个分支其实经常失败，只是被悄悄吃掉，最后只能靠业务方反馈“怎么总是没券”。

正确思路是：

兜底前先记录必要日志
区分超时、业务异常、系统异常
默认值要让调用方可识别，而不是伪装成正常数据

一张图看完整执行链路

sequenceDiagram
    participant Client as 调用方
    participant Home as 聚合服务
    participant U as 用户服务
    participant A as 账户服务
    participant C as 优惠券服务

    Client->>Home: 请求首页数据
    Home->>U: 异步获取用户信息
    Home->>A: 异步获取余额
    Home->>C: 异步获取优惠券
    U-->>Home: 返回用户信息
    A-->>Home: 返回余额
    alt 优惠券超时/异常
        Home-->>Home: 降级为空列表
    else 优惠券成功
        C-->>Home: 返回优惠券列表
    end
    Home-->>Client: 聚合结果

实战代码（可运行）

下面这段代码演示：

自定义线程池
3 个异步任务并行执行
对优惠券接口做超时降级
对余额接口做异常兜底
最终聚合结果返回

你可以直接复制运行。

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.*;
import java.util.function.Supplier;

public class CompletableFutureOrchestrationDemo {

    private static final ExecutorService BIZ_POOL = new ThreadPoolExecutor(
            4,
            8,
            60L,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(100),
            new ThreadFactory() {
                private int index = 1;
                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, "biz-pool-" + index++);
                }
            },
            new ThreadPoolExecutor.CallerRunsPolicy()
    );

    public static void main(String[] args) {
        try {
            HomeDTO homeDTO = getHomePage("user-1001");
            System.out.println("最终结果: " + homeDTO);
        } finally {
            BIZ_POOL.shutdown();
        }
    }

    public static HomeDTO getHomePage(String userId) {
        long start = System.currentTimeMillis();

        CompletableFuture<UserInfo> userFuture = CompletableFuture
                .supplyAsync(wrap("queryUserInfo", () -> queryUserInfo(userId)), BIZ_POOL)
                .orTimeout(500, TimeUnit.MILLISECONDS)
                .whenComplete((r, ex) -> logCompletion("userFuture", r, ex));

        CompletableFuture<Balance> balanceFuture = CompletableFuture
                .supplyAsync(wrap("queryBalance", () -> queryBalance(userId)), BIZ_POOL)
                .orTimeout(400, TimeUnit.MILLISECONDS)
                .exceptionally(ex -> {
                    System.out.println("[WARN] balanceFuture 异常，使用默认余额兜底: " + ex.getMessage());
                    return new Balance(BigDecimal.ZERO, true);
                });

        CompletableFuture<List<Coupon>> couponFuture = CompletableFuture
                .supplyAsync(wrap("queryCoupons", () -> queryCoupons(userId)), BIZ_POOL)
                .completeOnTimeout(Collections.emptyList(), 300, TimeUnit.MILLISECONDS)
                .exceptionally(ex -> {
                    System.out.println("[WARN] couponFuture 异常，降级为空列表: " + ex.getMessage());
                    return Collections.emptyList();
                });

        CompletableFuture<Void> all = CompletableFuture.allOf(userFuture, balanceFuture, couponFuture);

        try {
            all.join();

            UserInfo userInfo = userFuture.join();
            Balance balance = balanceFuture.join();
            List<Coupon> coupons = couponFuture.join();

            HomeDTO dto = new HomeDTO(userInfo, balance, coupons, LocalDateTime.now());
            System.out.println("总耗时: " + (System.currentTimeMillis() - start) + " ms");
            return dto;
        } catch (CompletionException e) {
            System.out.println("[ERROR] 首页聚合失败: " + e.getMessage());
            throw e;
        }
    }

    private static <T> Supplier<T> wrap(String taskName, Supplier<T> supplier) {
        return () -> {
            long start = System.currentTimeMillis();
            try {
                System.out.println("[INFO] 开始执行 " + taskName + ", thread=" + Thread.currentThread().getName());
                T result = supplier.get();
                System.out.println("[INFO] 执行完成 " + taskName + ", cost=" + (System.currentTimeMillis() - start) + " ms");
                return result;
            } catch (Exception e) {
                System.out.println("[ERROR] 执行异常 " + taskName + ", cost=" + (System.currentTimeMillis() - start) + " ms, ex=" + e.getMessage());
                throw e;
            }
        };
    }

    private static void logCompletion(String futureName, Object result, Throwable ex) {
        if (ex != null) {
            System.out.println("[ERROR] " + futureName + " 完成异常: " + ex.getMessage());
        } else {
            System.out.println("[INFO] " + futureName + " 正常完成, result=" + result);
        }
    }

    private static UserInfo queryUserInfo(String userId) {
        sleep(120);
        return new UserInfo(userId, "张三", 28);
    }

    private static Balance queryBalance(String userId) {
        sleep(200);
        // 你可以打开下面这行，模拟异常
        // throw new RuntimeException("账户服务不可用");
        return new Balance(new BigDecimal("1024.88"), false);
    }

    private static List<Coupon> queryCoupons(String userId) {
        sleep(350);
        return Arrays.asList(
                new Coupon("C100", "满100减10"),
                new Coupon("C200", "满200减30")
        );
    }

    private static void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("线程被中断", e);
        }
    }

    static class HomeDTO {
        private final UserInfo userInfo;
        private final Balance balance;
        private final List<Coupon> coupons;
        private final LocalDateTime generatedAt;

        public HomeDTO(UserInfo userInfo, Balance balance, List<Coupon> coupons, LocalDateTime generatedAt) {
            this.userInfo = userInfo;
            this.balance = balance;
            this.coupons = coupons;
            this.generatedAt = generatedAt;
        }

        @Override
        public String toString() {
            return "HomeDTO{" +
                    "userInfo=" + userInfo +
                    ", balance=" + balance +
                    ", coupons=" + coupons +
                    ", generatedAt=" + generatedAt +
                    '}';
        }
    }

    static class UserInfo {
        private final String userId;
        private final String name;
        private final int age;

        public UserInfo(String userId, String name, int age) {
            this.userId = userId;
            this.name = name;
            this.age = age;
        }

        @Override
        public String toString() {
            return "UserInfo{" +
                    "userId='" + userId + '\'' +
                    ", name='" + name + '\'' +
                    ", age=" + age +
                    '}';
        }
    }

    static class Balance {
        private final BigDecimal amount;
        private final boolean degraded;

        public Balance(BigDecimal amount, boolean degraded) {
            this.amount = amount;
            this.degraded = degraded;
        }

        @Override
        public String toString() {
            return "Balance{" +
                    "amount=" + amount +
                    ", degraded=" + degraded +
                    '}';
        }
    }

    static class Coupon {
        private final String code;
        private final String title;

        public Coupon(String code, String title) {
            this.code = code;
            this.title = title;
        }

        @Override
        public String toString() {
            return "Coupon{" +
                    "code='" + code + '\'' +
                    ", title='" + title + '\'' +
                    '}';
        }
    }
}

逐步拆解这段代码

1. 为什么要自定义线程池，而不是直接用默认线程池？

很多示例都这么写：

CompletableFuture.supplyAsync(() -> queryUserInfo(userId));

这会使用 ForkJoinPool.commonPool()。

它不是不能用，但在业务服务里，我通常不建议直接依赖默认线程池，原因有几个：

不同业务共用，容易互相影响
不方便做容量隔离
不方便命名线程，排查日志麻烦
遇到阻塞型任务时，commonPool 未必合适

所以更稳妥的方式是：

CompletableFuture.supplyAsync(task, bizExecutor)

一句话总结：
业务异步任务，尽量放到自己可控的线程池里。

2. `allOf()` 只是“等都完成”，不会直接帮你拿结果

很多人第一次看到 allOf() 会以为它能把多个结果自动组合起来。其实不会。

CompletableFuture<Void> all = CompletableFuture.allOf(f1, f2, f3);
all.join();

它只表示：

f1、f2、f3 都完成了
但结果还得你自己再从各自的 future 里取

所以通常会写成：

all.join();
A a = f1.join();
B b = f2.join();
C c = f3.join();

如果你就是 2 个任务合并，也可以考虑 thenCombine()，代码会更语义化。

3. 超时控制怎么选：`orTimeout()` vs `completeOnTimeout()`

这两个方法很像，但语义完全不同。

`orTimeout`

超时后，future 以异常结束。

适用于：

核心数据必须拿到
拿不到就应该让上层感知失败

future.orTimeout(300, TimeUnit.MILLISECONDS);

`completeOnTimeout`

超时后，future 正常返回一个默认值。

适用于：

非核心信息
可以接受降级展示

future.completeOnTimeout(Collections.emptyList(), 300, TimeUnit.MILLISECONDS);

我个人经验是：

用户身份、订单金额、库存结果：更偏向 orTimeout
推荐列表、优惠券、画像标签：更偏向 completeOnTimeout

4. 异常处理怎么选：`exceptionally()`、`handle()`、`whenComplete()`

这是最容易混的地方。

`exceptionally()`

只在异常时执行，适合做简单兜底。

future.exceptionally(ex -> defaultValue);

`handle()`

不管成功还是失败都会执行，并且可以转换结果。

future.handle((result, ex) -> {
    if (ex != null) {
        return defaultValue;
    }
    return transform(result);
});

`whenComplete()`

更像“结果回调”，常用于日志、监控，不改变结果。

future.whenComplete((result, ex) -> log(...));

一个简单记忆法：

要兜底：exceptionally
要统一收口并改结果：handle
只想打点/记录：whenComplete

进阶：任务之间有依赖时怎么编排

并不是所有场景都能平铺并行。
有些任务需要依赖上一步结果，比如：

先查用户
再根据用户等级查权益

这时就该用 thenCompose()，而不是在 thenApply() 里再套一个 future。

flowchart TD
    A[查用户信息] --> B{是否VIP}
    B -->|是| C[查VIP权益]
    B -->|否| D[返回普通权益]
    C --> E[聚合输出]
    D --> E

示例：

CompletableFuture<UserInfo> userFuture =
        CompletableFuture.supplyAsync(() -> queryUserInfo("u1"), BIZ_POOL);

CompletableFuture<String> benefitFuture = userFuture.thenCompose(user -> {
    if ("张三".equals(user.name)) {
        return CompletableFuture.supplyAsync(() -> "VIP权益包", BIZ_POOL);
    }
    return CompletableFuture.completedFuture("普通权益");
});

为什么不用 thenApply()？

因为 thenApply() 会得到 CompletableFuture<CompletableFuture<T>> 这种嵌套结构，不好用。
而 thenCompose() 会自动“拍平”。

常见坑与排查

这一节我会把项目里最容易踩的坑挑出来。

坑 1：用了异步，但最后很早就 `join()`，等于白并行

错误示例：

UserInfo userInfo = CompletableFuture
        .supplyAsync(() -> queryUserInfo(userId), BIZ_POOL)
        .join();

Balance balance = CompletableFuture
        .supplyAsync(() -> queryBalance(userId), BIZ_POOL)
        .join();

这其实还是串行。因为第一个任务一提交就立刻等结果了。

正确做法是先全部发出去，再统一等待：

CompletableFuture<UserInfo> f1 = CompletableFuture.supplyAsync(() -> queryUserInfo(userId), BIZ_POOL);
CompletableFuture<Balance> f2 = CompletableFuture.supplyAsync(() -> queryBalance(userId), BIZ_POOL);

CompletableFuture.allOf(f1, f2).join();

坑 2：线程池太小，任务互相阻塞

如果线程池只有 2 个线程，你同时提交了很多阻塞 IO 任务，队列一长，接口时延会非常难看。

排查思路：

看线程池活跃线程数
看队列积压
看任务平均执行时间
看是否有下游调用无超时导致线程长期占用

建议至少把这些指标打出来：

poolSize
activeCount
queueSize
completedTaskCount

坑 3：`join()` 抛的是 `CompletionException`，根因被包了一层

很多人日志里只打印：

e.getMessage()

结果看到的是一层包装异常，不知道真正下游出了什么问题。

更稳妥的做法：

try {
    future.join();
} catch (CompletionException e) {
    Throwable cause = e.getCause();
    System.out.println("root cause = " + cause);
}

坑 4：超时了不等于底层任务一定停止了

这是很容易误解的一点。

orTimeout() / completeOnTimeout() 主要是让 CompletableFuture 在超时点完成，
但底层那个实际执行中的任务，未必就被真正中断了。

这意味着什么？

你表面上已经返回降级结果了
但后台线程可能还在继续跑
如果慢请求很多，线程池还是会被拖住

所以超时控制不能只靠 CompletableFuture，还要配合：

HTTP 客户端超时
数据库查询超时
下游 SDK 超时
合理的线程池隔离

坑 5：默认值兜底过头，导致“假成功”

比如余额查询失败，你直接返回 0 元；
前端看起来接口成功了，但业务含义其实错了。

更好的方式是让默认值带降级标记：

new Balance(BigDecimal.ZERO, true);

这样前端或调用方可以知道：

这是降级值
不是用户真实余额

排查清单：线上接口慢时怎么查

这里给一个我自己常用的简版清单。

第一步：看是否真的并行了

检查代码有没有这种问题：

future 创建后马上 join()
中间某一步不必要地阻塞等待
依赖关系写错，本来可并行却串起来了

第二步：看线程池是否健康

重点看：

核心线程数/最大线程数是否过小
队列是否堆积
拒绝策略是否频繁触发
是否混用了 CPU 密集和 IO 密集任务

第三步：看超时是否只做了“表面超时”

如果 future 已经超时，但底层 HTTP 还在卡 5 秒，那线程池仍然会慢慢被占满。

第四步：看异常是否被吞

尤其检查：

exceptionally() 里是否只返回默认值不记录日志
是否缺 traceId、userId、接口名等上下文
是否区分了超时、限流、业务异常

安全/性能最佳实践

这一部分不只讲“能跑”，而是讲“上线后更稳”。

1. 线程池隔离要按业务域做

不要把所有异步任务都丢到同一个大池子里。

更推荐按业务隔离：

用户中心相关线程池
营销相关线程池
搜索/推荐相关线程池

这样某个依赖抖动时，不容易把整个系统的异步任务都拖死。

2. IO 密集型任务不要照搬 CPU 核数配置

如果任务主要是远程调用、数据库访问、缓存访问，它们大部分时间在等待 IO。

这类线程池参数不能简单套“CPU 核数 + 1”的经验值，要结合：

平均响应时间
峰值并发
可接受排队时间
下游限流能力

我建议先从压测数据反推，而不是拍脑袋。

3. 为每个下游设置独立超时和降级策略

不要全局统一写成“500ms 超时”。

因为不同下游的重要性、稳定性、平均耗时都不一样：

用户身份：200ms 超时，失败即报错
余额：300ms 超时，降级并标记
推荐内容：150ms 超时，直接返回空列表

统一超时看起来简单，实际上往往不合理。

4. 日志要可观测，但不要打爆

异步编排场景建议记录这些字段：

traceId/requestId
userId/orderId
taskName
startTime/cost
timeout 标记
degraded 标记
exception 类型

但注意别在高频成功路径上打印太多 INFO 日志，否则又会带来额外性能开销。

5. 不要把 CompletableFuture 当成分布式容错框架

CompletableFuture 很适合做进程内异步编排，但它不是全能方案。

如果你面对的是更复杂的场景，比如：

分布式重试
熔断限流
批量隔离舱
服务级 fallback
链路级治理

那还要结合：

Resilience4j
Sentinel
Hystrix（老项目里还会见到）
网关超时/限流策略
RPC 框架自身治理能力

也就是说，CompletableFuture 解决的是**“怎么编排”**，不是解决所有稳定性问题。

一个推荐的编码模板

如果你准备在业务里落地，我建议参考这种结构：

CompletableFuture<ResultA> aFuture = CompletableFuture
        .supplyAsync(() -> callA(), executorA)
        .orTimeout(200, TimeUnit.MILLISECONDS)
        .exceptionally(ex -> fallbackA(ex));

CompletableFuture<ResultB> bFuture = CompletableFuture
        .supplyAsync(() -> callB(), executorB)
        .completeOnTimeout(defaultB(), 150, TimeUnit.MILLISECONDS)
        .whenComplete((r, ex) -> metrics("callB", ex));

CompletableFuture<ResultC> cFuture = CompletableFuture
        .supplyAsync(() -> callC(), executorC);

CompletableFuture<Void> all = CompletableFuture.allOf(aFuture, bFuture, cFuture);
all.join();

return assemble(aFuture.join(), bFuture.join(), cFuture.join());

这个模板的优点是：

启动、超时、兜底、打点分层清楚
读代码时容易看出每个分支的责任
后续扩展监控和降级比较方便

边界条件：什么时候不适合用 CompletableFuture？

也别神化它。下面这些情况，我会更谨慎：

1. 任务之间关系特别复杂

如果有大量条件分支、动态依赖、批量 fan-out/fan-in，纯 CompletableFuture 链条会越来越难读。

2. 下游基本都是阻塞调用，线程数会膨胀

这种场景异步编排虽然能提升聚合效率，但本质还是“多线程包裹阻塞 IO”。
如果并发极高，可能还要考虑响应式方案或更彻底的架构优化。

3. 团队对异常链和线程池不熟

如果大家不熟悉 CompletionException、超时语义、线程池容量，写出来的代码表面上很“高级”，其实更难维护。

总结

用一句话概括这篇文章：

CompletableFuture 最有价值的地方，不是“开异步”，而是“把多个异步任务编排得可控、可降级、可排查”。

你可以优先记住这几个实战原则：

先并行发起，再统一等待，别刚提交就 join()
业务线程池自己配，别过度依赖默认线程池
核心依赖用 orTimeout()，非核心依赖用 completeOnTimeout()
异常兜底要带日志和降级标记，别做“假成功”
Future 超时不代表底层调用停止，HTTP/RPC/DB 超时也要单独配置
把可观测性补齐：耗时、异常、超时、线程池状态都要看得见

如果你现在正在写聚合接口，我建议你先挑一个最典型的场景，把串行调用改成“3 路并行 + 1 个超时降级 + 1 个异常兜底”，跑一遍压测和日志观察。
只要你真正做过一次，CompletableFuture 这套东西就不再只是 API 记忆题，而会变成非常实用的工程工具。