CompletableFuture 在聚合接口中的性能实践

商品详情页 P99 1.8 秒，五个 RPC 串行调用

12 月中旬，前端同学甩过来一张截图：商品详情页首屏白屏 2 秒多，用户投诉"点商品没反应"。我去看 SkyWalking 的拓扑，这个接口平均 690 ms，P99 1.8 s，P999 3.2 s。

接口干的事很简单，串着调了 5 个下游：

public ItemDetailVO getItemDetail(Long itemId) {
    ItemBase base   = itemService.getBase(itemId);        // 42 ms
    Stock stock     = stockService.getStock(itemId);      // 183 ms
    Price price     = priceService.getPrice(itemId);      // 121 ms
    List<Promo> promos = promoService.listPromo(itemId);  // 264 ms
    int commentCnt  = commentService.countComment(itemId);// 91 ms

    return assemble(base, stock, price, promos, commentCnt);
}

注释里是各家的平均耗时，加起来 701 ms，跟 P50 对得上。这五个调用彼此之间没有依赖，纯粹是写代码时图省事串行下来的。

P99 高是因为营销服务不稳定，偶尔抖到 800 ms，一抖整条链就拖长。

第一版改造：allOf 并行

public ItemDetailVO getItemDetail(Long itemId) {
    CompletableFuture<ItemBase> baseF = CompletableFuture
            .supplyAsync(() -> itemService.getBase(itemId), pool);
    CompletableFuture<Stock> stockF = CompletableFuture
            .supplyAsync(() -> stockService.getStock(itemId), pool);
    CompletableFuture<Price> priceF = CompletableFuture
            .supplyAsync(() -> priceService.getPrice(itemId), pool);
    CompletableFuture<List<Promo>> promoF = CompletableFuture
            .supplyAsync(() -> promoService.listPromo(itemId), pool);
    CompletableFuture<Integer> commentF = CompletableFuture
            .supplyAsync(() -> commentService.countComment(itemId), pool);

    CompletableFuture.allOf(baseF, stockF, priceF, promoF, commentF).join();

    return assemble(baseF.join(), stockF.join(), priceF.join(),
                    promoF.join(), commentF.join());
}

上线后 P50 从 690 ms 降到 272 ms，P99 从 1.8 s 降到 510 ms。效果和预期一致：并行后总耗时取决于最慢的那个（营销 264 ms）。

然后就出事了。

上线的第三天，详情页集体超时

周三上午十点，告警：商品详情页 P99 从 510 ms 涨到 4.6 秒，错误率 3.7%。

我第一反应是营销服务又抖了，但去看营销服务的监控，人家一切正常，平均 258 ms。再看我们自己服务的线程栈：

$ jstack 18234 | grep -A 15 "ForkJoinPool.commonPool-worker"
"ForkJoinPool.commonPool-worker-13" #218 daemon prio=5 os_prio=0 tid=0x00007f...
   java.lang.Thread.State: RUNNABLE
        at com.xxx.task.ReportTask.lambda$rebuild$2(ReportTask.java:142)
        at com.xxx.task.ReportTask$$Lambda$923.run(Unknown Source)
        at java.util.concurrent.ForkJoinTask$RunnableExecuteAction.exec(...)

线程全在跑 ReportTask——一个每天上午十点跑的报表重建任务。它是另一位同事写的，也用了 CompletableFuture.supplyAsync()，但没传线程池。

没传线程池的 supplyAsync 会用 ForkJoinPool.commonPool()。这个池的大小是 Runtime.getRuntime().availableProcessors() - 1，我们容器是 4 核，池里只有 3 个线程。报表任务的 join() 又占着不放，3 个线程被吃干，详情页的任务全在外面排队。

严格说，我第一版代码里的 pool 是传了的，所以详情页本身没用 commonPool。但同进程的另一个聚合接口用了默认池，被拖垮后大量线程阻塞在等待上，把 Tomcat 的 200 个工作线程也占满了，最后整个服务不可用——这才是我真正踩的坑：同一进程里只要有一处用默认池，大家就一起完蛋。

线程池隔离

改法很直接：按下游服务拆线程池，一个慢下游最多拖垮它自己那个池，不会波及其他。

@Configuration
public class AsyncPoolConfig {

    private ThreadPoolExecutor buildPool(String name, int core, int queue) {
        return new ThreadPoolExecutor(core, core, 0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(queue),
                new ThreadFactoryBuilder().setNameFormat(name + "-%d").build(),
                new ThreadPoolExecutor.CallerRunsPolicy());
    }

    @Bean("itemPool")   public ThreadPoolExecutor itemPool()    { return buildPool("item",   8, 256); }
    @Bean("stockPool")  public ThreadPoolExecutor stockPool()   { return buildPool("stock",  16, 512); }
    @Bean("pricePool")  public ThreadPoolExecutor pricePool()   { return buildPool("price",  12, 512); }
    @Bean("promoPool")  public ThreadPoolExecutor promoPool()   { return buildPool("promo",  16, 512); }
    @Bean("commentPool") public ThreadPoolExecutor commentPool(){ return buildPool("comment", 8, 256); }
}

几个决定说明一下：

• 队列用 LinkedBlockingQueue 而不是 SynchronousQueue。Executors.newCachedThreadPool 用的是后者，任务来了就无限建线程，高并发下会 OOM。我见过一次，线上直接 unable to create new native thread。
• 拒绝策略用 CallerRunsPolicy。池满时让调用线程自己跑，相当于天然的降级——虽然慢，但不丢请求。聚合接口里这比直接抛异常好。
• 队列长度按"能撑住下游挂掉多久"算。库存 QPS 峰值 800，池 16 个线程，单线程每秒处理 5 个，队列 512 大约能撑 1.6 秒的积压，配合下面的超时正好。

顺便把报表任务也翻出来改了，全局搜了一遍 supplyAsync( 不带第二个参数的调用，一共 7 处，全部补上独立线程池。

超时控制：JDK 8 没有 orTimeout

第二个必须解决的是超时。CompletableFuture.orTimeout() 是 Java 9 才加的，我们线上还是 JDK 8，得自己实现。

private static final ScheduledExecutorService TIMER =
        Executors.newScheduledThreadPool(2,
            new ThreadFactoryBuilder().setNameFormat("cf-timeout-%d").build());

public static <T> CompletableFuture<T> withTimeout(
        CompletableFuture<T> future, long timeout, TimeUnit unit, String name) {

    CompletableFuture<T> timeoutFuture = new CompletableFuture<>();
    ScheduledFuture<?> timerTask = TIMER.schedule(() -> {
        if (timeoutFuture.completeExceptionally(
                new TimeoutException(name + " timeout " + unit.toMillis(timeout) + "ms"))) {
            future.cancel(true);
        }
    }, timeout, unit);

    future.whenComplete((r, e) -> {
        timerTask.cancel(false);
        if (e != null) {
            timeoutFuture.completeExceptionally(e);
        } else {
            timeoutFuture.complete(r);
        }
    });
    return timeoutFuture;
}

这里有个容易被忽略的点：future.cancel(true) 只是给正在执行的任务发中断信号，如果下游是 HTTP 调用且没配 read timeout，中断是没用的，线程照样卡着。所以 Dubbo 的 timeout 和 HTTP client 的 socketTimeout 必须配，这层超时是兜底不是主力。

我们的 Dubbo 配置：

dubbo:
  consumer:
    timeout: 800
    retries: 0        # 聚合接口一律不重试，重试会放大下游压力

降级：异常也要返回可用结果

五个下游里，只有商品基础信息是必需的，其他四个挂了都应该返回带默认值的页面，而不是整个报错。

public ItemDetailVO getItemDetail(Long itemId) {
    // 基础信息：失败就抛
    CompletableFuture<ItemBase> baseF = withTimeout(
            CompletableFuture.supplyAsync(() -> itemService.getBase(itemId), itemPool),
            300, TimeUnit.MILLISECONDS, "itemBase");

    // 其余四个：失败降级
    CompletableFuture<Stock> stockF = withTimeout(
            CompletableFuture.supplyAsync(() -> stockService.getStock(itemId), stockPool),
            500, TimeUnit.MILLISECONDS, "stock")
            .exceptionally(e -> { log.warn("stock fallback", e); return Stock.unknown(); });

    CompletableFuture<Price> priceF = withTimeout(
            CompletableFuture.supplyAsync(() -> priceService.getPrice(itemId), pricePool),
            500, TimeUnit.MILLISECONDS, "price")
            .exceptionally(e -> { log.warn("price fallback", e); return Price.empty(); });

    CompletableFuture<List<Promo>> promoF = withTimeout(
            CompletableFuture.supplyAsync(() -> promoService.listPromo(itemId), promoPool),
            600, TimeUnit.MILLISECONDS, "promo")
            .exceptionally(e -> { log.warn("promo fallback", e); return Collections.emptyList(); });

    CompletableFuture<Integer> commentF = withTimeout(
            CompletableFuture.supplyAsync(() -> commentService.countComment(itemId), commentPool),
            400, TimeUnit.MILLISECONDS, "comment")
            .exceptionally(e -> { log.warn("comment fallback", e); return 0; });

    CompletableFuture.allOf(baseF, stockF, priceF, promoF, commentF).join();
    return assemble(baseF.join(), stockF.join(), priceF.join(),
                    promoF.join(), commentF.join());
}

exceptionally 里一定要打日志并上报指标。我们给每个下游加了 fallback_count 的 counter，配了告警——不然降级生效了你都不知道，等用户反馈"价格怎么是空的"就晚了。

有依赖关系的情况用 thenCompose

不是所有场景都能全并行。比如要先拿类目 ID 再查类目属性，就得用 thenCompose（注意不是 thenApply，后者会得到 CompletableFuture<CompletableFuture<T>>）：

CompletableFuture<CategoryAttr> attrF = baseF.thenComposeAsync(
        base -> CompletableFuture.supplyAsync(
                () -> categoryService.getAttr(base.getCategoryId()), catPool),
        catPool);

我们详情页里有一处这种依赖，thenCompose 让它排在基础信息之后，总耗时是 300 + 150 = 450 ms，仍在可接受范围。

最终数据

P99 从 1.8 s 到 342 ms，主要是超时控制把长尾砍掉了。线程池隔离本身不提速，但它避免了那次"一个报表任务拖垮整个详情页"。

小结

• supplyAsync 不传线程池会掉进 ForkJoinPool.commonPool()，4 核容器只有 3 个线程。全项目搜一遍补上，别偷这个懒。
• 线程池按下游隔离，拒绝策略用 CallerRunsPolicy 做降级，别用 newCachedThreadPool。
• JDK 8 没有 orTimeout，用 ScheduledExecutorService + completeExceptionally 自己实现。但真正管用的是客户端自己的 read timeout。
• 非关键路径一定要 exceptionally 降级，并且给降级次数配监控告警。
• 有依赖用 thenCompose 不是 thenApply。

还有个我至今觉得没做好的地方：这五个线程池的参数是我按峰值 QPS 拍的，上线后没再调过。后来流量翻倍，促销池的队列经常积压到 300 多。复盘时我记了一条：池的活跃线程数和队列长度必须进监控，光看接口耗时，等发现的时候已经在排队了。

参考

• 官方文档：java-concurrent
• Baeldung 上的相关实战
• StackOverflow 上的相关讨论