Java 中基于 CompletableFuture 的异步编排实战：从并行聚合到超时降级设计

在做服务端开发时，我们经常会碰到一种很典型的场景：一个接口要同时依赖多个下游服务。
比如商品详情页，可能要同时查：

商品基础信息
价格
库存
营销活动
推荐列表

如果这些调用串行执行，接口总耗时通常会被最慢链路拖垮；如果一味并行，又很容易把线程池、超时控制、异常传播搞得一团糟。

CompletableFuture 的价值，就在于它把 异步执行、结果组合、异常处理、超时控制 放到了一个统一模型里。
这篇文章我不打算只讲 API，而是从“一个聚合接口如何设计”出发，带你把一套 可运行、可排查、可扩展 的异步编排方案走一遍。

背景与问题

先看一个常见的聚合接口需求：

/api/product/detail?id=1001

它内部可能依赖 4 个服务：

商品服务：拿基础信息
库存服务：拿库存
价格服务：拿实时价格
营销服务：拿活动标签

串行调用的问题

最朴素的写法是这样：

Product product = productService.getProduct(id);
Stock stock = stockService.getStock(id);
Price price = priceService.getPrice(id);
Promotion promotion = promotionService.getPromotion(id);

如果每个服务平均 80ms，总耗时可能接近：

80 + 80 + 80 + 80 = 320ms

而真实线上环境里，下游服务抖动、偶发慢调用、个别接口超时非常常见。最终你会遇到几个老问题：

尾延迟高：一个慢依赖拖慢整个接口
局部失败影响整体：一个服务报错导致整个页面失败
线程资源浪费：阻塞等待过多
超时策略分散：每段调用都在自己处理超时，代码很碎
排查困难：日志里看不到到底是哪个 future 卡住了

我们想要的目标

一个更稳妥的聚合架构，通常要满足：

可并行的请求尽量并行
核心依赖失败时快速失败
非核心依赖失败时自动降级
整体接口有统一超时边界
线程池隔离，避免相互拖死
错误能定位，指标能观测

核心原理

CompletableFuture 可以把异步编排分成几类操作：

1. 创建异步任务

常见方式：

runAsync：无返回值
supplyAsync：有返回值

CompletableFuture<Product> productFuture =
    CompletableFuture.supplyAsync(() -> productService.getProduct(id), executor);

2. 结果转换与依赖编排

thenApply：同步转换结果
thenCompose：把“future 套 future”拉平
thenCombine：组合两个独立结果
allOf：等待全部完成
anyOf：等待任意一个完成

3. 异常处理

exceptionally：出错时给默认值
handle：无论成功失败都处理
whenComplete：做收尾日志，但不改结果

4. 超时控制

Java 9 之后常用：

orTimeout：超时则抛异常
completeOnTimeout：超时则返回默认值

这两个方法在聚合场景里非常关键。
我的经验是：

核心链路：倾向 orTimeout
可降级链路：倾向 completeOnTimeout

一张图先看懂整体编排

flowchart LR
    A[请求进入 /product/detail] --> B[并行发起商品/库存/价格/营销查询]
    B --> C1[商品服务]
    B --> C2[库存服务]
    B --> C3[价格服务]
    B --> C4[营销服务]
    C1 --> D[结果聚合]
    C2 --> D
    C3 --> D
    C4 --> D
    D --> E{是否有核心依赖失败}
    E -- 是 --> F[快速失败或返回兜底页]
    E -- 否 --> G[返回聚合结果]

方案设计：并行聚合 + 核心/非核心分级 + 超时降级

先明确一个设计原则：

核心依赖与非核心依赖要分开

以商品详情页为例：

核心依赖
- 商品基础信息
- 价格
非核心依赖
- 库存提示
- 营销标签

为什么要这么分？

因为用户看不到商品基本信息和价格，页面基本不可用；
但营销标签没返回，页面通常还能继续展示。

一个推荐的决策表

依赖项	是否核心	超时策略	异常策略
商品信息	是	`orTimeout`	抛出，整体失败
价格	是	`orTimeout`	抛出，整体失败
库存	否	`completeOnTimeout`	返回默认库存状态
营销	否	`completeOnTimeout`	返回空活动

这一步非常重要。很多团队写异步编排时，最大的问题不是 API 不会用，而是没有先定义业务降级边界。

实战代码（可运行）

下面给一份完整示例，基于 JDK 11，可直接运行。
为了方便演示，我用 sleep 模拟远程调用延迟。

示例目标

实现一个商品详情聚合服务：

并行查询四个下游
核心依赖超时直接失败
非核心依赖超时自动降级
使用独立线程池
打印总耗时

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

public class CompletableFutureOrchestrationDemo {

    public static void main(String[] args) {
        ProductDetailService service = new ProductDetailService();

        try {
            ProductDetailDTO detail = service.getProductDetail(1001L);
            System.out.println("调用成功: " + detail);
        } catch (Exception e) {
            System.err.println("调用失败: " + e.getMessage());
            e.printStackTrace();
        } finally {
            service.shutdown();
        }
    }

    static class ProductDetailService {
        private final ExecutorService executor = new ThreadPoolExecutor(
                8,
                16,
                60L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(200),
                r -> {
                    Thread t = new Thread(r);
                    t.setName("detail-async-" + t.getId());
                    return t;
                },
                new ThreadPoolExecutor.CallerRunsPolicy()
        );

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

            CompletableFuture<Product> productFuture =
                    CompletableFuture.supplyAsync(wrap("product", () -> getProduct(productId)), executor)
                            .orTimeout(300, TimeUnit.MILLISECONDS);

            CompletableFuture<Price> priceFuture =
                    CompletableFuture.supplyAsync(wrap("price", () -> getPrice(productId)), executor)
                            .orTimeout(250, TimeUnit.MILLISECONDS);

            CompletableFuture<Stock> stockFuture =
                    CompletableFuture.supplyAsync(wrap("stock", () -> getStock(productId)), executor)
                            .completeOnTimeout(Stock.degraded(), 150, TimeUnit.MILLISECONDS)
                            .exceptionally(ex -> {
                                System.out.println("[降级] stock 查询失败: " + ex.getMessage());
                                return Stock.degraded();
                            });

            CompletableFuture<Promotion> promotionFuture =
                    CompletableFuture.supplyAsync(wrap("promotion", () -> getPromotion(productId)), executor)
                            .completeOnTimeout(Promotion.empty(), 120, TimeUnit.MILLISECONDS)
                            .exceptionally(ex -> {
                                System.out.println("[降级] promotion 查询失败: " + ex.getMessage());
                                return Promotion.empty();
                            });

            try {
                CompletableFuture.allOf(productFuture, priceFuture, stockFuture, promotionFuture).join();

                Product product = productFuture.join();
                Price price = priceFuture.join();
                Stock stock = stockFuture.join();
                Promotion promotion = promotionFuture.join();

                ProductDetailDTO dto = new ProductDetailDTO(
                        product.getId(),
                        product.getName(),
                        price.getAmount(),
                        stock.getStatus(),
                        promotion.getTags(),
                        LocalDateTime.now()
                );

                long cost = System.currentTimeMillis() - start;
                System.out.println("总耗时: " + cost + "ms");
                return dto;
            } catch (CompletionException e) {
                long cost = System.currentTimeMillis() - start;
                System.err.println("聚合失败，总耗时: " + cost + "ms");
                throw new RuntimeException("商品详情查询失败", e.getCause());
            }
        }

        private <T> Supplier<T> wrap(String taskName, Supplier<T> supplier) {
            return () -> {
                long start = System.currentTimeMillis();
                try {
                    T result = supplier.get();
                    long cost = System.currentTimeMillis() - start;
                    System.out.println(taskName + " success, cost=" + cost + "ms, thread=" + Thread.currentThread().getName());
                    return result;
                } catch (Exception e) {
                    long cost = System.currentTimeMillis() - start;
                    System.err.println(taskName + " fail, cost=" + cost + "ms, thread=" + Thread.currentThread().getName()
                            + ", ex=" + e.getMessage());
                    throw e;
                }
            };
        }

        private Product getProduct(Long productId) {
            sleep(100);
            return new Product(productId, "机械键盘");
        }

        private Price getPrice(Long productId) {
            sleep(120);
            return new Price(new BigDecimal("399.00"));
        }

        private Stock getStock(Long productId) {
            sleep(200); // 故意慢一点，触发降级
            return new Stock("有货");
        }

        private Promotion getPromotion(Long productId) {
            sleep(80);
            return new Promotion(List.of("满减", "会员价"));
        }

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

        public void shutdown() {
            executor.shutdown();
        }
    }

    static class Product {
        private final Long id;
        private final String name;

        public Product(Long id, String name) {
            this.id = id;
            this.name = name;
        }

        public Long getId() {
            return id;
        }

        public String getName() {
            return name;
        }
    }

    static class Price {
        private final BigDecimal amount;

        public Price(BigDecimal amount) {
            this.amount = amount;
        }

        public BigDecimal getAmount() {
            return amount;
        }
    }

    static class Stock {
        private final String status;

        public Stock(String status) {
            this.status = status;
        }

        public static Stock degraded() {
            return new Stock("库存未知");
        }

        public String getStatus() {
            return status;
        }
    }

    static class Promotion {
        private final List<String> tags;

        public Promotion(List<String> tags) {
            this.tags = tags;
        }

        public static Promotion empty() {
            return new Promotion(Collections.emptyList());
        }

        public List<String> getTags() {
            return tags;
        }
    }

    static class ProductDetailDTO {
        private final Long productId;
        private final String productName;
        private final BigDecimal price;
        private final String stockStatus;
        private final List<String> promotionTags;
        private final LocalDateTime queryTime;

        public ProductDetailDTO(Long productId, String productName, BigDecimal price,
                                String stockStatus, List<String> promotionTags,
                                LocalDateTime queryTime) {
            this.productId = productId;
            this.productName = productName;
            this.price = price;
            this.stockStatus = stockStatus;
            this.promotionTags = promotionTags;
            this.queryTime = queryTime;
        }

        @Override
        public String toString() {
            return "ProductDetailDTO{" +
                    "productId=" + productId +
                    ", productName='" + productName + '\'' +
                    ", price=" + price +
                    ", stockStatus='" + stockStatus + '\'' +
                    ", promotionTags=" + promotionTags +
                    ", queryTime=" + queryTime +
                    '}';
        }
    }
}

运行结果预期

因为 getStock() 人为 sleep 了 200ms，而库存超时设置为 150ms，所以它会降级为：

库存未知

而其他核心依赖正常返回，最终整个接口仍然成功。

关键编排过程时序图

sequenceDiagram
    participant Client as 调用方
    participant Service as 聚合服务
    participant P as 商品服务
    participant R as 价格服务
    participant S as 库存服务
    participant M as 营销服务

    Client->>Service: 请求商品详情
    Service->>P: 异步查询商品
    Service->>R: 异步查询价格
    Service->>S: 异步查询库存
    Service->>M: 异步查询营销

    P-->>Service: 100ms 返回成功
    R-->>Service: 120ms 返回成功
    M-->>Service: 80ms 返回成功
    S--xService: 超过 150ms，触发默认值降级

    Service->>Service: allOf + join 聚合
    Service-->>Client: 返回商品详情（库存=库存未知）

核心原理再往前一步：为什么 `allOf` 后还要 `join`

不少人第一次写会疑惑：

allOf(...).join() 都已经等完了，为什么还要对每个 future 再 join() 一次？

因为：

CompletableFuture.allOf() 的返回值本身不携带每个任务结果
它只是表示“这些 future 都完成了”
真正的结果，还得分别从各自 future 中取

典型写法就是：

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

T1 r1 = f1.join();
T2 r2 = f2.join();
T3 r3 = f3.join();

这点看起来有点啰嗦，但它能保持组合过程清晰。

方案对比与取舍分析

在架构设计里，异步编排不是唯一方案。常见有三种：

方案一：串行调用

优点

代码最简单
排查容易

缺点

总耗时高
下游多时性能差

适合依赖很少、响应时间要求不高的内部接口。

方案二：手工线程池 + Future

优点

比串行快
能做基础并发

缺点

结果组合不方便
异常处理繁琐
API 可读性差

这是很多旧系统的历史包袱，我自己也维护过，后面重构时痛苦非常明显。

方案三：CompletableFuture 编排

优点

组合能力强
能表达依赖关系
统一处理异常、超时、降级
更适合聚合型接口

缺点

链式调用容易写乱
对线程池和异常传播理解要求更高

取舍建议

如果你的接口满足下面特征，我会优先推荐 CompletableFuture：

下游依赖 3 个以上
下游之间大多独立
需要部分失败可降级
需要统一超时边界
需要做聚合返回

常见坑与排查

这部分很重要。我当时踩过不少坑，很多不是业务问题，而是“以为异步了，其实没有”。

坑 1：默认线程池被误用

如果你这样写：

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

没有传入自定义 executor，它默认走 ForkJoinPool.commonPool()。

这会带来两个风险：

线程池与你的业务线程混用
阻塞型 IO 调用容易把 commonPool 拖死

建议：
只要是服务端聚合接口，几乎都应该显式传入自定义线程池。

坑 2：把 `join()` 写早了，导致“伪并行”

错误示例：

Product product = CompletableFuture
        .supplyAsync(() -> getProduct(id), executor)
        .join();

Price price = CompletableFuture
        .supplyAsync(() -> getPrice(id), executor)
        .join();

看起来用了异步，实际上还是串行等待。

正确方式：先全部发起，再统一等待。

CompletableFuture<Product> pf = CompletableFuture.supplyAsync(() -> getProduct(id), executor);
CompletableFuture<Price> rf = CompletableFuture.supplyAsync(() -> getPrice(id), executor);

CompletableFuture.allOf(pf, rf).join();

坑 3：异常被吞掉了

有些代码只写了：

future.exceptionally(ex -> null);

这样虽然接口没报错，但也可能把真实故障掩盖掉。
线上最怕的不是报错，而是“默默返回错误数据”。

建议：

降级前先记日志
区分核心与非核心依赖
给默认值时要能从结果中识别“这是降级数据”

例如：

.exceptionally(ex -> {
    logError("promotion", ex);
    return Promotion.empty();
})

坑 4：超时只超 future，不一定真的取消下游执行

orTimeout / completeOnTimeout 更多是对 Future 结果层面 生效。
如果底层是阻塞 IO，请求本身未必真的被中断。

这意味着：

调用方已经返回了
下游任务可能还在跑
线程和连接资源还在占用

排查方法：

看线程池活跃线程数
看 HTTP 客户端连接池
看超时后下游是否仍持续收到请求

建议：
真正要止损，底层 HTTP/RPC 客户端也必须设置连接超时、读超时、请求超时。

坑 5：线程池配置拍脑袋

线程池太小：

并发一高，队列堆积
超时率上升

线程池太大：

上下文切换增加
内存占用上涨
下游被放大打爆

一个粗略思路：

线程数 ≈ CPU核数 * (1 + 平均等待时间 / 平均计算时间)

但在聚合接口里，大多数任务偏 IO 型，这个公式只能做起点。
最终还是要结合压测数据调优。

一张线程池与超时边界关系图

flowchart TB
    A[请求进入] --> B[聚合线程发起多个 CompletableFuture]
    B --> C[业务专用线程池]
    C --> D1[商品调用]
    C --> D2[价格调用]
    C --> D3[库存调用]
    C --> D4[营销调用]

    D1 --> E[核心依赖超时: 失败]
    D2 --> E
    D3 --> F[非核心依赖超时: 默认值]
    D4 --> F

    E --> G[整体失败或兜底页]
    F --> H[聚合后成功返回]

安全/性能最佳实践

这里我把“安全”和“性能”放一起讲，因为在服务端异步编排里，它们经常是一体两面。

1. 不要把用户输入直接透传到并行任务里无限放大

比如一个接口允许传：

商品 ID 列表
每次最多 500 个

然后你对每个 ID 都起 4 个 future，那一次请求就可能变成 2000 个异步任务。
这会直接引发：

线程池拥塞
下游被打穿
内存暴涨

建议：

限制单次请求 fan-out 数量
对批量场景优先走下游批量接口
必要时分批并行

2. 专用线程池隔离不同业务类型

至少区分：

IO 密集型聚合线程池
CPU 密集型计算线程池

不要让：

报表导出
图片处理
商品详情聚合

共用一个线程池。

否则某个慢业务高峰时，会把其他接口一起拖慢。

3. 降级默认值必须“业务可接受”

默认值不是随便填一个就行。
例如库存降级成“有货”，可能导致超卖风险；
更稳妥的做法通常是：

库存未知
暂不展示库存
请稍后重试

同理，价格失败也不能随便返回 0，这可能造成严重业务事故。

4. 给异步任务打日志与指标

至少建议记录：

taskName
开始时间 / 耗时
是否超时
是否降级
异常类型
线程池队列长度
活跃线程数

如果没有这些指标，线上问题往往只能靠猜。

5. 聚合接口要有总超时，不要只配子任务超时

很多系统只给每个下游设置了 100ms、200ms 超时，但整体接口没有总超时。
结果是某些边界情况里，聚合逻辑自己还在等待。

可以在更高层再包一层：

CompletableFuture<ProductDetailDTO> resultFuture =
    CompletableFuture.supplyAsync(() -> service.getProductDetail(id), executor)
        .orTimeout(500, TimeUnit.MILLISECONDS);

当然，是否这样封装，要结合你的框架和调用模型来定。

6. 注意上下文传递问题

如果你在线程上下文里放了这些信息：

TraceId
用户身份
租户信息
MDC 日志上下文

切到异步线程后，它们默认可能丢失。

这会导致：

链路追踪断裂
日志无法串起来
安全上下文失效

建议：

使用支持上下文传递的封装 executor
或在任务提交前手动复制上下文

这是很多团队一开始最容易忽略的“隐性故障点”。

一个更稳的编排模板

如果你不想每次都手写一堆重复逻辑，可以沉淀一个小工具方法，比如：

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

public class AsyncHelper {

    public static <T> CompletableFuture<T> supplyAsyncWithFallback(
            Supplier<T> supplier,
            T fallback,
            long timeout,
            TimeUnit unit,
            Executor executor) {

        return CompletableFuture.supplyAsync(supplier, executor)
                .completeOnTimeout(fallback, timeout, unit)
                .exceptionally(ex -> fallback);
    }

    public static <T> CompletableFuture<T> supplyAsyncRequired(
            Supplier<T> supplier,
            long timeout,
            TimeUnit unit,
            Executor executor) {

        return CompletableFuture.supplyAsync(supplier, executor)
                .orTimeout(timeout, unit);
    }
}

业务层就会清爽很多：

CompletableFuture<Product> productFuture =
        AsyncHelper.supplyAsyncRequired(() -> getProduct(id), 300, TimeUnit.MILLISECONDS, executor);

CompletableFuture<Stock> stockFuture =
        AsyncHelper.supplyAsyncWithFallback(() -> getStock(id), Stock.degraded(), 150, TimeUnit.MILLISECONDS, executor);

注意，这种工具类只是统一模式，不要把异常信息彻底吃掉。
真实项目里最好把日志、指标、trace 也一起封进去。

容量估算的简单思路

对于聚合接口，做容量估算时可以先看三个数：

单请求 fan-out 数量
下游平均 RT / TP99
峰值 QPS

举个简化例子：

接口 QPS：200
每个请求并行 4 个下游
平均单任务占线程 100ms

那么每秒异步任务量大致是：

200 * 4 = 800 个任务/秒

平均并发中的在途任务数约为：

800 * 0.1 = 80

线程池核心线程数就不可能只配 8、16 这种非常小的值，否则高峰会明显排队。
当然，这只是粗估，真实配置还要考虑：

队列长度
突发流量
超时比例
下游限流阈值
降级比例

常见排查路径

线上如果出现“接口偶发超时”，我一般会按这个顺序查：

先看总体耗时分布
- 平均值、P95、P99
再看各个子任务耗时
- 哪个依赖最慢
看线程池状态
- active count、queue size、reject count
看超时和降级数量
- 是个别依赖抖动，还是整体资源不够
看下游客户端配置
- connect timeout、read timeout 是否合理
检查是否有伪并行
- 中途提前 join
检查异常处理
- 是否被 exceptionally 吃掉

这个流程比“上来就调大线程池”靠谱得多。

总结

CompletableFuture 不是为了“把代码写得更炫”，它真正解决的是聚合场景下几个现实问题：

如何把独立依赖并行化
如何清晰表达依赖关系
如何给不同依赖配置不同超时策略
如何在部分失败时优雅降级
如何把线程池、异常、日志、指标统一起来

如果你准备在项目里落地，我建议按下面顺序推进：

先划分核心/非核心依赖
给每类依赖定义超时与降级策略
统一使用业务专用线程池
先全部发起，再统一聚合，避免伪并行
为每个异步任务补齐日志、指标、trace
压测验证线程池容量与降级边界

最后说一个边界条件：
如果你的链路里已经引入了响应式框架，或者需要更复杂的流式处理、背压控制，那 CompletableFuture 未必是终点。但对于绝大多数 Java 服务端聚合接口 来说，它依然是一种成本适中、效果很好的异步编排方案。

如果你现在的接口还在串行查 4 个下游，不妨先挑一个最典型的聚合接口，按本文的方式做一版。通常第一版改完，RT 和可用性就会有比较明显的改善。