背景与问题

在 Java 后端系统里，一个请求往往不是一次方法调用就结束。它可能要查用户信息、查订单、查库存、调风控、打优惠、拼装响应对象。
如果这些操作都串行执行，整体耗时通常就是各阶段耗时的累加：

• 用户服务 80ms
• 订单服务 120ms
• 库存服务 60ms
• 营销服务 90ms

串行下来接近 350ms，还没算网络抖动、线程切换和重试成本。对于高并发接口，这种写法非常容易把响应时间和线程池一起拖垮。

很多团队最初会这么做：

• 用 Future 提交任务，但拿结果时仍然阻塞
• 手写 CountDownLatch、BlockingQueue 组织流程
• 回调嵌套越来越深，最后谁先执行、谁失败、谁超时都不好管

这时，CompletableFuture 的价值就出来了：
它不是“另一个 Future”，而是一个可声明式编排异步任务的工具。

你可以把它理解成：

1. Future：表示“未来会有一个结果”
2. CompletionStage：表示“结果出来之后，还能继续接下一步”

所以它不仅能异步执行任务，还能把多个任务之间的并行、串行、聚合、异常处理、超时控制组织起来。

本文我从“架构实战”的角度讲，不只讲 API，而是讲在高并发场景下怎么设计、怎么写、怎么避坑。

典型业务场景：聚合查询接口

假设我们要实现一个商品详情页接口，需要并行获取：

• 商品基础信息
• 价格信息
• 库存信息
• 营销标签

然后汇总成一个结果返回。

如果写成串行调用，流程很直白，但延迟也很直白。
而使用 CompletableFuture，可以把互不依赖的部分并行化，把有依赖关系的部分串接起来。

flowchart LR
    A[接收商品详情请求] --> B[并行查询基础信息]
    A --> C[并行查询价格]
    A --> D[并行查询库存]
    A --> E[并行查询营销]
    B --> F[结果聚合]
    C --> F
    D --> F
    E --> F
    F --> G[返回响应]

这个模型很适合：

• 聚合查询
• 多服务并发调用
• 批量异步处理
• 事件流水线处理
• 对外部 I/O 密集型任务做并发编排

但也要注意：它不是银弹。
如果你的任务是 CPU 密集型计算，或者任务之间有复杂共享状态，设计重点就不是“多开几个异步任务”这么简单了。

核心原理

1. CompletableFuture 到底解决了什么

CompletableFuture 同时实现了两个角色：

• Future<T>：可以表示异步结果
• CompletionStage<T>：可以链式定义后续动作

这意味着它支持：

• 创建异步任务：supplyAsync / runAsync
• 串行依赖：thenApply / thenCompose
• 并行合并：thenCombine / allOf
• 竞争返回：applyToEither / anyOf
• 异常处理：exceptionally / handle / whenComplete

一句话概括：
把“线程执行”提升为“阶段编排”。

2. 常见编排关系

串行依赖：thenApply / thenCompose

• thenApply：上一步结果做转换
• thenCompose：上一步结果再触发一个新的异步任务，并“拍平”结果

比如：

• 先查用户
• 再根据用户等级异步查推荐列表

这类依赖最好用 thenCompose。

并行汇总：thenCombine / allOf

• thenCombine：两个任务都完成后合并结果
• allOf：等待多个任务全部完成

异常兜底：exceptionally / handle

高并发系统里，失败不是意外，而是常态。
一个下游接口偶发超时，不能直接把整个请求链路拖死。

3. 默认线程池不是万能的

很多人第一次用 CompletableFuture，会直接写：

CompletableFuture.supplyAsync(() -> query());

这样会默认使用 ForkJoinPool.commonPool()。
问题在于：

• 业务线程和公共线程池混用，互相影响
• I/O 阻塞型任务会拖慢整个池
• 不方便做容量隔离和监控

在生产环境里，我的建议是：几乎总是显式传入业务线程池。

4. 任务编排不是越多越好

异步化有收益，也有成本：

• 线程切换
• 对象创建
• 上下文切换
• 调试复杂度
• 异常传播复杂度

如果一个任务只有 1~2ms，而且纯本地计算，盲目异步化可能还会更慢。
所以更合理的原则是：

• I/O 密集且互不依赖：优先并行
• 强依赖链路：适度串联
• CPU 密集：关注池大小和限流
• 结果可降级：单点失败可兜底

方案对比与取舍分析

在 Java 里做异步编排，常见方案大概有这几类：

如果你的系统还是以同步 Servlet、普通 Spring MVC 为主，但又有明显的聚合查询/并行调用需求，
CompletableFuture 往往是性价比很高的一步升级。

核心原理图：任务生命周期与异常分支

stateDiagram-v2
    [*] --> Created
    Created --> Running: submit
    Running --> Success: completed
    Running --> Failed: exception
    Running --> Timeout: timeout
    Success --> Chained: thenApply/thenCompose
    Failed --> Recovered: exceptionally/handle
    Timeout --> Recovered: fallback
    Chained --> [*]
    Recovered --> [*]

这张图有一个重点：
异常、超时、成功，本质上都是“完成态”。
只是完成的结果不同，后续链路处理方式也不同。

实战代码（可运行）

下面我们做一个简化但可运行的示例：模拟商品详情聚合接口。

功能点包括：

• 自定义线程池
• 并行查询多个下游
• 聚合结果
• 超时控制
• 异常降级

代码基于 JDK 9+ 的 orTimeout / completeOnTimeout。如果你用 JDK 8，我后面会补充兼容思路。

import java.math.BigDecimal;
import java.util.concurrent.*;
import java.util.function.Supplier;

public class CompletableFutureOrchestrationDemo {

    private static final ThreadPoolExecutor IO_POOL = new ThreadPoolExecutor(
            8,
            16,
            60L,
            TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(200),
            new NamedThreadFactory("io-pool-"),
            new ThreadPoolExecutor.CallerRunsPolicy()
    );

    public static void main(String[] args) {
        try {
            ProductDetail detail = getProductDetail(1001L);
            System.out.println("最终结果：");
            System.out.println(detail);
        } finally {
            IO_POOL.shutdown();
        }
    }

    public static ProductDetail getProductDetail(Long productId) {
        long start = System.currentTimeMillis();

        CompletableFuture<ProductInfo> infoFuture = supplyAsyncWithLog(
                "queryProductInfo",
                () -> queryProductInfo(productId)
        ).completeOnTimeout(ProductInfo.defaultInfo(productId), 300, TimeUnit.MILLISECONDS)
         .exceptionally(ex -> {
             System.out.println("商品基础信息查询失败，降级: " + ex.getMessage());
             return ProductInfo.defaultInfo(productId);
         });

        CompletableFuture<PriceInfo> priceFuture = supplyAsyncWithLog(
                "queryPrice",
                () -> queryPrice(productId)
        ).orTimeout(250, TimeUnit.MILLISECONDS)
         .exceptionally(ex -> {
             System.out.println("价格查询失败，降级: " + ex.getMessage());
             return new PriceInfo(productId, BigDecimal.ZERO);
         });

        CompletableFuture<StockInfo> stockFuture = supplyAsyncWithLog(
                "queryStock",
                () -> queryStock(productId)
        ).completeOnTimeout(new StockInfo(productId, 0), 200, TimeUnit.MILLISECONDS)
         .exceptionally(ex -> {
             System.out.println("库存查询失败，降级: " + ex.getMessage());
             return new StockInfo(productId, 0);
         });

        CompletableFuture<MarketingInfo> marketingFuture = supplyAsyncWithLog(
                "queryMarketing",
                () -> queryMarketing(productId)
        ).exceptionally(ex -> {
            System.out.println("营销信息查询失败，降级: " + ex.getMessage());
            return new MarketingInfo(productId, "暂无活动");
        });

        CompletableFuture<ProductDetail> detailFuture =
                infoFuture.thenCombine(priceFuture, ProductDetail::withInfoAndPrice)
                          .thenCombine(stockFuture, ProductDetail::withStock)
                          .thenCombine(marketingFuture, ProductDetail::withMarketing)
                          .whenComplete((result, ex) -> {
                              long cost = System.currentTimeMillis() - start;
                              if (ex == null) {
                                  System.out.println("聚合完成，耗时: " + cost + " ms");
                              } else {
                                  System.out.println("聚合失败，耗时: " + cost + " ms, ex=" + ex.getMessage());
                              }
                          });

        return detailFuture.join();
    }

    private static <T> CompletableFuture<T> supplyAsyncWithLog(String taskName, Supplier<T> supplier) {
        return CompletableFuture.supplyAsync(() -> {
            String threadName = Thread.currentThread().getName();
            System.out.println("开始执行任务: " + taskName + ", thread=" + threadName);
            T result = supplier.get();
            System.out.println("完成任务: " + taskName + ", thread=" + threadName);
            return result;
        }, IO_POOL);
    }

    private static ProductInfo queryProductInfo(Long productId) {
        sleep(120);
        return new ProductInfo(productId, "机械键盘", "热插拔 RGB");
    }

    private static PriceInfo queryPrice(Long productId) {
        sleep(180);
        return new PriceInfo(productId, new BigDecimal("399.00"));
    }

    private static StockInfo queryStock(Long productId) {
        sleep(90);
        return new StockInfo(productId, 58);
    }

    private static MarketingInfo queryMarketing(Long productId) {
        sleep(140);
        return new MarketingInfo(productId, "满300减30");
    }

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

    static class ProductInfo {
        Long productId;
        String name;
        String desc;

        public ProductInfo(Long productId, String name, String desc) {
            this.productId = productId;
            this.name = name;
            this.desc = desc;
        }

        static ProductInfo defaultInfo(Long productId) {
            return new ProductInfo(productId, "默认商品", "系统降级返回");
        }
    }

    static class PriceInfo {
        Long productId;
        BigDecimal price;

        public PriceInfo(Long productId, BigDecimal price) {
            this.productId = productId;
            this.price = price;
        }
    }

    static class StockInfo {
        Long productId;
        int stock;

        public StockInfo(Long productId, int stock) {
            this.productId = productId;
            this.stock = stock;
        }
    }

    static class MarketingInfo {
        Long productId;
        String campaign;

        public MarketingInfo(Long productId, String campaign) {
            this.productId = productId;
            this.campaign = campaign;
        }
    }

    static class ProductDetail {
        Long productId;
        String name;
        String desc;
        BigDecimal price;
        int stock;
        String campaign;

        static ProductDetail withInfoAndPrice(ProductInfo info, PriceInfo price) {
            ProductDetail detail = new ProductDetail();
            detail.productId = info.productId;
            detail.name = info.name;
            detail.desc = info.desc;
            detail.price = price.price;
            return detail;
        }

        ProductDetail withStock(StockInfo stockInfo) {
            this.stock = stockInfo.stock;
            return this;
        }

        ProductDetail withMarketing(MarketingInfo marketingInfo) {
            this.campaign = marketingInfo.campaign;
            return this;
        }

        @Override
        public String toString() {
            return "ProductDetail{" +
                    "productId=" + productId +
                    ", name='" + name + '\'' +
                    ", desc='" + desc + '\'' +
                    ", price=" + price +
                    ", stock=" + stock +
                    ", campaign='" + campaign + '\'' +
                    '}';
        }
    }

    static class NamedThreadFactory implements ThreadFactory {
        private final String prefix;
        private int counter = 1;

        NamedThreadFactory(String prefix) {
            this.prefix = prefix;
        }

        @Override
        public synchronized Thread newThread(Runnable r) {
            Thread t = new Thread(r, prefix + counter++);
            t.setDaemon(false);
            return t;
        }
    }
}

这段代码里最值得注意的几点

1. 并行查询，而不是串行等待

四个下游调用一开始就都发出去了，而不是：

1. 查商品
2. 等返回
3. 查价格
4. 再等返回

这就是异步编排最直接的收益来源。

2. 用 thenCombine 表达“汇合点”

这比手工 get() + 拼装更自然。
它的意思很明确：等两个任务都完成，再合并。

如果任务特别多，也可以用 allOf：

CompletableFuture<Void> all = CompletableFuture.allOf(infoFuture, priceFuture, stockFuture, marketingFuture);

CompletableFuture<ProductDetail> future = all.thenApply(v -> {
    ProductInfo info = infoFuture.join();
    PriceInfo price = priceFuture.join();
    StockInfo stock = stockFuture.join();
    MarketingInfo marketing = marketingFuture.join();

    ProductDetail detail = ProductDetail.withInfoAndPrice(info, price);
    detail.withStock(stock);
    detail.withMarketing(marketing);
    return detail;
});

适合“先全部跑，再统一收集”。

3. 降级逻辑要就地绑定

我很不建议把异常处理拖到最后一起做。
原因很简单：到了链路末端，你已经很难分辨到底是哪一个分支失败了。

更好的方式是：

• 价格失败，价格分支自己兜底
• 库存失败，库存分支自己兜底
• 营销失败，营销分支自己兜底

这样聚合逻辑会干净很多。

依赖与并行混合编排示意

现实业务通常不是纯并行，也不是纯串行，而是混合型。

比如：

• 先查用户
• 再根据用户等级查优惠券
• 同时并行查订单和地址
• 最后统一组装

sequenceDiagram
    participant Req as Request
    participant U as UserService
    participant C as CouponService
    participant O as OrderService
    participant A as AddressService
    participant Agg as Aggregator

    Req->>U: 异步查询用户
    U-->>Req: 用户信息
    Req->>C: 基于用户等级查询优惠券
    Req->>O: 并行查询订单
    Req->>A: 并行查询地址
    C-->>Agg: 优惠券结果
    O-->>Agg: 订单结果
    A-->>Agg: 地址结果
    Agg-->>Req: 聚合响应

这类场景下，通常是：

• thenCompose 处理依赖链
• thenCombine 或 allOf 处理并发汇总

示例代码：

CompletableFuture<User> userFuture =
        CompletableFuture.supplyAsync(() -> queryUser(userId), IO_POOL);

CompletableFuture<Coupon> couponFuture =
        userFuture.thenCompose(user ->
                CompletableFuture.supplyAsync(() -> queryCoupon(user.level()), IO_POOL)
        );

CompletableFuture<Order> orderFuture =
        CompletableFuture.supplyAsync(() -> queryOrder(userId), IO_POOL);

CompletableFuture<Address> addressFuture =
        CompletableFuture.supplyAsync(() -> queryAddress(userId), IO_POOL);

CompletableFuture<UserProfileView> resultFuture =
        couponFuture.thenCombine(orderFuture, Pair::new)
                    .thenCombine(addressFuture, (pair, address) -> {
                        Coupon coupon = pair.coupon();
                        Order order = pair.order();
                        return new UserProfileView(coupon, order, address);
                    });

容量估算：线程池不是拍脑袋配的

高并发下，最容易出问题的不是 CompletableFuture 本身，而是线程池配置。

一个粗略经验：

• I/O 密集型任务：线程数可高于 CPU 核数
• CPU 密集型任务：线程数接近 CPU 核数

如果你的接口平均会并发触发 4 个远程调用，每个调用平均阻塞 100ms，单机 QPS 目标是 200，那么池容量至少要考虑：

• 每秒触发异步子任务数：200 * 4 = 800
• 平均占用时长：100ms
• 理论并发占用线程数约：800 * 0.1 = 80

再考虑抖动、超时、重试、偶发慢请求，线程池和队列都要留出余量。

当然，这只是粗估。
真正上线前，最好用压测结果校准：

• 平均响应时间
• P95/P99
• 活跃线程数
• 队列堆积
• 拒绝次数

常见坑与排查

1. join/get 用早了，异步变串行

这是最常见的坑之一。比如：

CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> serviceA(), IO_POOL);
String a = f1.join();

CompletableFuture<String> f2 = CompletableFuture.supplyAsync(() -> serviceB(a), IO_POOL);
String b = f2.join();

这样写本质上还是一步一步等。
如果任务之间无依赖，应该先全部发起，再统一等待。

2. 把阻塞 I/O 丢进 commonPool

默认公共线程池很方便，但线上问题也很方便。
症状通常是：

• 吞吐量上不去
• 某些任务响应越来越慢
• CPU 不高，但请求积压
• 线程堆栈大量卡在网络 I/O 或 Thread.sleep

排查方式：

1. 看线程名是不是 ForkJoinPool.commonPool-worker-*
2. 看线程栈是否长期阻塞
3. 看是否不同业务共用一个池

3. thenApply 和 thenCompose 混淆

错误示例：

CompletableFuture<CompletableFuture<String>> future =
        CompletableFuture.supplyAsync(() -> "user1", IO_POOL)
                .thenApply(user -> CompletableFuture.supplyAsync(() -> queryProfile(user), IO_POOL));

这里会得到“双层 Future”。
如果你想要真正的链式异步结果，应该用：

CompletableFuture<String> future =
        CompletableFuture.supplyAsync(() -> "user1", IO_POOL)
                .thenCompose(user -> CompletableFuture.supplyAsync(() -> queryProfile(user), IO_POOL));

记忆方法很简单：

• thenApply：同步映射
• thenCompose：异步展开

4. 异常被包装，看不清根因

join() 抛的是 CompletionException，get() 抛的是 ExecutionException。
真实异常通常包在 getCause() 里。

排查时要特别注意：

try {
    future.join();
} catch (CompletionException e) {
    Throwable cause = e.getCause();
    cause.printStackTrace();
}

很多人日志里只打印 e.getMessage()，结果看到一层包装信息，定位很痛苦。

5. allOf 不会直接给你聚合结果

CompletableFuture.allOf(...) 返回的是 CompletableFuture<Void>。
它只表示“都完成了”，不帮你自动收集每个结果。

所以你仍然要手动从各个 future 里取值。

6. 线程池队列过大，问题被“隐藏”

有些系统为了不拒绝任务，把队列设到几万。
短期看像“稳定”，长期看其实是在积压：

• 请求越堆越多
• 平均延迟越来越大
• 超时越来越多
• GC 压力也会上来

我踩过这个坑：监控上线程池拒绝是 0，但接口超时飙升。最后一看，队列里堆了几千个任务。
所以不要只看“有没有拒绝”，还要看“有没有排队过长”。

排查思路：从现象到根因

如果线上出现“异步编排接口突然变慢”，我通常会按下面路径查：

flowchart TD
    A[接口超时/变慢] --> B{线程池是否饱和}
    B -->|是| C[看活跃线程数 队列长度 拒绝次数]
    B -->|否| D{某个下游是否慢}
    C --> E[检查池大小 队列策略 隔离是否合理]
    D -->|是| F[查看分支耗时 超时配置 降级逻辑]
    D -->|否| G{是否提前 join 导致串行}
    G -->|是| H[调整编排结构]
    G -->|否| I[检查异常吞掉/日志不足]

配套监控建议至少包含：

• 每个异步分支耗时
• 每个线程池活跃线程数
• 队列长度
• 拒绝策略触发次数
• 超时次数
• 降级次数

这些指标一旦缺失，CompletableFuture 出问题时会显得特别“玄学”。

安全/性能最佳实践

1. 按业务维度隔离线程池

不要把所有异步任务都塞到一个池里。
建议至少按以下维度拆分：

• 核心链路 vs 非核心链路
• I/O 密集型 vs CPU 密集型
• 外部依赖调用 vs 本地处理任务

这样即使某个下游抖动，也不会把全部异步能力拖死。

2. 明确超时策略，不无限等待

任何远程调用都应该有超时：

• 下游 RPC 超时
• Future 等待超时
• 整体接口总超时

CompletableFuture 只是编排工具，不会自动帮你控制远程调用时间。
如果底层 HTTP/RPC 客户端没配置好超时，就算 orTimeout 了，底层线程可能仍在阻塞。

这是个很重要的边界条件：
Future 超时不等于底层任务一定停止。

3. 对可降级数据做兜底，对核心数据快速失败

不是所有数据都值得“死等”。

例如商品详情页里：

• 商品基础信息：核心，不建议随意缺失
• 营销标签：可降级
• 推荐位：通常也可降级
• 埋点上报：尽量异步解耦，不影响主流程

所以异步编排设计前，先做一层业务分级，比 API 选型更重要。

4. 保留链路日志与上下文

异步执行后，经常会遇到：

• 日志 traceId 丢了
• MDC 上下文丢了
• 无法把多个分支日志串起来

如果系统依赖 ThreadLocal/MDC 传递上下文，要考虑封装任务提交逻辑，把上下文复制到异步线程。
否则排查问题会非常难。

5. 不要吞异常

一些写法表面上“做了异常处理”，本质上却把错误信息弄没了：

future.exceptionally(ex -> null);

如果这样写，后面业务逻辑可能在 null 上继续跑，最终变成一个完全无关的 NullPointerException。
更好的做法是：

• 记录原始异常
• 返回明确的降级对象
• 给调用方可识别的状态

6. 控制任务粒度

别把一个请求切成几十上百个异步任务。
过细的任务粒度会导致：

• 调度开销增加
• 对象创建增多
• 线程竞争变重
• 代码可读性下降

经验上，优先异步化“真正耗时”的 I/O 分支，而不是为异步而异步。

7. 选择合适的拒绝策略

像本文示例中用的是 CallerRunsPolicy，优点是能做一定程度的反压。
但它也有副作用：调用线程会直接执行任务，可能拖慢上游请求线程。

所以拒绝策略没有绝对标准，要看场景：

• 核心链路：更关注系统稳定性
• 非核心任务：可丢弃、可降级
• 批量异步任务：可做限流和分批处理

JDK 8 怎么办

如果你还在 JDK 8，没有 orTimeout 和 completeOnTimeout，可以用：

• 底层 RPC/HTTP 客户端自身超时
• 配合 ScheduledExecutorService 手动构造超时 future
• 或引入工具库做补充

不过从工程实践上说，如果你的系统已经重度依赖异步编排，升级到更高版本 JDK 会省很多事。

什么时候不建议用 CompletableFuture

虽然它很好用，但这些场景需要谨慎：

1. 复杂流式处理和反压控制
   更适合 Reactor、RxJava 这类响应式方案。

2. 超高吞吐、低延迟、极致性能场景
   需要更精细的线程模型和事件驱动架构。

3. 任务之间共享可变状态非常多
   这时异步化会放大并发问题，先理顺状态边界更重要。

4. 只是简单的一次异步提交
   如果没有编排需求，普通线程池提交就够了，不必过度设计。

总结

CompletableFuture 真正强的地方，不是“把任务丢到线程池里跑”，而是把异步流程变成一条可组合、可控制、可降级、可观测的任务编排链。

如果你要在 Java 里做高并发异步任务编排，我建议按下面顺序落地：

1. 先识别哪些任务可以并行，哪些必须串行
2. 显式定义业务线程池，做隔离
3. 给每个分支配超时、异常处理、降级策略
4. 用 thenCompose、thenCombine、allOf 表达依赖关系
5. 补齐监控：耗时、队列、拒绝、超时、降级次数
6. 压测验证，而不是只看功能跑通

最后给一个很实用的判断标准：

• 如果你只是“想异步一下”，先别急着上
• 如果你已经遇到“串行等待慢、聚合逻辑乱、异常难管理”的问题，CompletableFuture 就很值得用

它不是最复杂的异步框架，但对于大量 Java 中后台系统来说，已经足够强，而且足够实用。