商品详情页 800 毫秒，我把它改成了 180 毫秒
10 月底做性能优化，商品详情页是重点。这个页面要聚合 6 个数据源，原来的代码是串行调的：
public ProductDetailVO getDetail(Long skuId) {
ProductVO product = productService.get(skuId); // 45 ms
StockVO stock = stockService.getStock(skuId); // 38 ms
PriceVO price = priceService.getPrice(skuId); // 120 ms
List<CouponVO> coupons = couponService.listBySku(skuId); // 210 ms
List<CommentVO> comments = commentService.listTop(skuId); // 310 ms
RecommendVO recommend = recommendService.get(skuId); // 75 ms
return assemble(product, stock, price, coupons, comments, recommend);
}
6 个加起来 798 毫秒，实测接口 P99 是 860 毫秒。这些调用之间没有任何依赖，完全可以并行。改成 CompletableFuture 之后 P99 降到 320 毫秒（取决于最慢的那个 310 毫秒）。
过程中踩了几个坑，一并记下。
第一个坑：默认的线程池
第一版我写得很快：
CompletableFuture<ProductVO> f1 = CompletableFuture.supplyAsync(() -> productService.get(skuId));
CompletableFuture<StockVO> f2 = CompletableFuture.supplyAsync(() -> stockService.getStock(skuId));
没传 Executor，跑起来也正常。但这是有问题的。看 supplyAsync 的源码：
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier) {
return asyncSupplyStage(asyncPool, supplier);
}
// asyncPool 的选择逻辑
private static final Executor asyncPool = useCommonPool ?
ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();
// useCommonPool 的判断
private static final boolean useCommonPool =
(ForkJoinPool.getCommonPoolParallelism() > 1);
也就是说，默认用的是 ForkJoinPool.commonPool()，和 parallelStream 同一个池，并行度是 CPU 核数 - 1。我们 4 核机器，并行度 3。
问题在于：这些 RPC 调用是 IO 密集的，线程大部分时间在等网络响应。如果只有 3 个线程，同时只能并发 3 个请求，第 4 个就得排队。6 个数据源要分两批，跟串行相比收益有限。更糟的是这个池是全局共享的，别的地方用并行流会跟我们抢。
正确做法：必须自己传线程池，而且按业务隔离。
@Configuration
public class AsyncPoolConfig {
/** 商品详情聚合专用：IO 密集，线程数给多一点 */
@Bean("detailPool")
public ExecutorService detailPool() {
return new ThreadPoolExecutor(
32, 32, 0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<>(500),
new ThreadFactoryBuilder().setNameFormat("detail-pool-%d").build(),
new ThreadPoolExecutor.AbortPolicy());
}
/** 消息推送：独立池，避免被详情页拖垮 */
@Bean("pushPool")
public ExecutorService pushPool() {
return new ThreadPoolExecutor(
8, 16, 60L, TimeUnit.SECONDS,
new SynchronousQueue<>(),
new ThreadFactoryBuilder().setNameFormat("push-pool-%d").build(),
new ThreadPoolExecutor.CallerRunsPolicy());
}
}
隔离的意义我后来才体会到：有一次推荐服务响应变慢（从 75 毫秒涨到 3 秒），因为详情页用了独立线程池，只是详情页变慢，其他功能的异步任务不受影响。如果共用池子，线程会被慢调用占满，全站异步任务一起挂。
thenApply 还是 thenApplyAsync
这是最容易搞混的一对。区别是：不带 Async 的方法在"触发它的那个线程"上执行，带 Async 的才提交到线程池。
CompletableFuture<String> f = CompletableFuture.supplyAsync(() -> {
log.info("supply in {}", Thread.currentThread().getName()); // detail-pool-1
return "hello";
}, detailPool);
f.thenApply(s -> {
log.info("thenApply in {}", Thread.currentThread().getName());
// ↑ detail-pool-1！和上游同一个线程（也可能是主线程，看时序）
return s + " world";
});
f.thenApplyAsync(s -> {
log.info("thenApplyAsync in {}", Thread.currentThread().getName());
// ↑ ForkJoinPool.commonPool-worker-1（没传 executor 的话）
return s + " world";
}, detailPool);
关键点是：thenApply 的执行线程不确定。如果调用它的时候上游已经完成，那它就在当前线程（比如 Tomcat 线程）同步执行；如果上游还没完成，那它在完成上游的那个线程上执行。
这个不确定性会导致一个隐蔽的问题：如果你的转换逻辑很重，用 thenApply 可能会阻塞上游的 IO 线程，或者阻塞主线程。我的规则是：转换逻辑轻（几十行内的纯计算）用 thenApply，重的话用 thenApplyAsync 并指定线程池。
thenCompose 和 thenCombine
两个名字很像，语义完全不同。
thenCompose 用来串联，解决"下一步依赖上一步的结果，且下一步本身也是异步的"这种嵌套。它的作用相当于 Stream 的 flatMap，把 CompletableFuture<CompletableFuture<T>> 拍平。
// 不用 thenCompose，会嵌套
CompletableFuture<CompletableFuture<Address>> nested = getUserAsync(userId)
.thenApply(user -> getAddressAsync(user.getAddressId()));
// 用 thenCompose，直接拿到 CompletableFuture<Address>
CompletableFuture<Address> flat = getUserAsync(userId)
.thenCompose(user -> getAddressAsync(user.getAddressId()));
签名上的区别：thenApply 接收 Function<T, U>，thenCompose 接收 Function<T, CompletionStage<U>>。
thenCombine 用来并联，两个独立的 Future 都完成后，把结果合并：
CompletableFuture<ProductVO> productF = getProductAsync(skuId);
CompletableFuture<PriceVO> priceF = getPriceAsync(skuId);
CompletableFuture<ProductWithPrice> combined = productF.thenCombine(priceF,
(product, price) -> new ProductWithPrice(product, price));
注意 thenCombine 只有两个参数版本，三个以上要嵌套或者用 allOf。
allOf：等所有完成，但它不给你结果
最终版的代码用的是 allOf：
@Autowired
@Qualifier("detailPool")
private ExecutorService detailPool;
public ProductDetailVO getDetail(Long skuId) {
CompletableFuture<ProductVO> productF = CompletableFuture
.supplyAsync(() -> productService.get(skuId), detailPool)
.exceptionally(this::defaultProduct);
CompletableFuture<StockVO> stockF = CompletableFuture
.supplyAsync(() -> stockService.getStock(skuId), detailPool)
.exceptionally(e -> defaultStock());
CompletableFuture<PriceVO> priceF = CompletableFuture
.supplyAsync(() -> priceService.getPrice(skuId), detailPool)
.exceptionally(e -> defaultPrice());
CompletableFuture<List<CouponVO>> couponF = CompletableFuture
.supplyAsync(() -> couponService.listBySku(skuId), detailPool)
.exceptionally(e -> Collections.emptyList());
CompletableFuture<List<CommentVO>> commentF = CompletableFuture
.supplyAsync(() -> commentService.listTop(skuId), detailPool)
.exceptionally(e -> Collections.emptyList());
CompletableFuture<RecommendVO> recommendF = CompletableFuture
.supplyAsync(() -> recommendService.get(skuId), detailPool)
.exceptionally(e -> RecommendVO.empty());
CompletableFuture<Void> all = CompletableFuture.allOf(
productF, stockF, priceF, couponF, commentF, recommendF);
try {
// 等所有任务完成，最多等 500 毫秒
all.get(500, TimeUnit.MILLISECONDS);
} catch (TimeoutException e) {
log.warn("getDetail timeout, skuId={}", skuId);
// 不抛异常，用已经完成的部分组装
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} catch (ExecutionException e) {
log.error("getDetail failed", e);
}
// 逐个 join / getNow，没完成的会拿到默认值
return assemble(
productF.getNow(defaultProduct(null)),
stockF.getNow(defaultStock()),
priceF.getNow(defaultPrice()),
couponF.getNow(Collections.emptyList()),
commentF.getNow(Collections.emptyList()),
recommendF.getNow(RecommendVO.empty()));
}
三个要点：
一、allOf 返回的是 CompletableFuture<Void>，它不聚合结果，只表示"都完成了"。想拿结果还是得逐个 join() 或 getNow()。我第一次用的时候以为能拿到 List，翻了半天 API。
二、getNow(defaultValue) 不会阻塞，完成了就返回值，没完成就返回你给的默认值。这是超时的关键——配合上面的 get(500, ms)，超时后用 getNow 能拿到"已经完成的那部分"，页面降级显示。如果直接用 join()，超时后照样会阻塞等剩下的任务。
三、JDK 8 没有 orTimeout。orTimeout 和 completeOnTimeout 是 JDK 9 才加的，我们跑的是 JDK 8u212，只能用 get(timeout) 这种老办法。所以上面那段 try-catch 有点啰嗦，没办法。
异常处理
CompletableFuture 的异常不会自动往外抛，它会被"封印"在 Future 里，只有你调 get() 或 join() 时才以 ExecutionException / CompletionException 的形式冒出来。如果不处理，异常就静默丢失了。
三个处理方法：
// exceptionally：只处理异常，相当于 catch
CompletableFuture<String> f1 = future.exceptionally(ex -> {
log.error("failed", ex);
return "default"; // 必须返回一个同类型的值
});
// handle：不管成功失败都调用，能拿到结果和异常
CompletableFuture<String> f2 = future.handle((result, ex) -> {
if (ex != null) {
log.warn("failed, use default", ex);
return "default";
}
return result;
});
// whenComplete：类似 finally，能拿到结果但改不了（返回值还是原结果）
CompletableFuture<String> f3 = future.whenComplete((result, ex) -> {
if (ex != null) {
monitor.increment("xxx_fail");
} else {
monitor.increment("xxx_success");
}
});
我在详情页里用的是 exceptionally，因为每个数据源的降级策略不同：优惠券挂了返回空列表（用户看不到券，但能下单），价格挂了返回默认价（宁可不显示也不要显示错的）。
有个坑：handle 和 whenComplete 里，如果正常完成，第二个参数 ex 是 null；如果异常完成，第一个参数 result 是 null。判断的时候别写反。
再一个坑：异常传播链。如果 thenApply 里抛了异常，后续所有依赖它的阶段都会被"短路"，直接以异常完成。这段是好的，符合直觉。但 allOf 不一样——只要有一个任务异常完成，allOf 的 future 就异常完成，而其他任务仍然在跑。所以我在每个任务上都单独加了 exceptionally，把它们各自兜住，allOf 就不会因为某一个失败而整体失败。
数据
改完之后压测，100 并发跑 5 分钟：
| 版本 | P50 | P99 | 最大耗时 | 单机 QPS |
|---|---|---|---|---|
| 串行 | 798 ms | 860 ms | 2,140 ms | 124 |
| 并行（公共池） | 412 ms | 1,320 ms | 4,800 ms | 238 |
| 并行（独立池 32 线程） | 318 ms | 340 ms | 512 ms | 311 |
注意中间那行：用公共池（并行度 3）的时候，虽然 P50 降了一半，但 P99 反而涨到 1320 毫秒。因为 6 个任务抢 3 个线程，排队等待时间被放大了，而且和其他用到公共池的任务互相干扰。这个数据说明：IO 密集的异步任务必须配独立的、线程数充足的线程池。
线程数怎么定？我们按 线程数 = CPU核数 × (1 + 平均等待时间/平均计算时间) 估算。详情页这几个调用平均等待 120 毫秒、本地计算不到 2 毫秒，算出来理论值很大，但受限于下游服务的承受能力，最后拍板 32。压测 32 和 64 差别不大，说明瓶颈已经不在线程数上了。
不适合用 CompletableFuture 的场景
改造过程中我也退回了几个改动，说说哪些不该用：
- 两个步骤之间有数据依赖。比如"先查订单，再用订单里的 userId 查用户"，这种只能串行，硬拆成异步再
thenCompose，代码可读性变差，性能也没提升。 - 在事务里。Spring 的
@Transactional靠 ThreadLocal 传递数据库连接，异步线程里拿不到。我在异步任务里调数据库 mapper，结果报"no transaction in progress"。要异步就不能有事务，或者把事务边界拆开。 - 需要 ThreadLocal 传递上下文。我们用了
MDC存 traceId，异步线程里全丢了，日志串不起来。得手动传：
Map<String, String> mdcContext = MDC.getCopyOfContextMap();
CompletableFuture.supplyAsync(() -> {
if (mdcContext != null) {
MDC.setContextMap(mdcContext);
}
try {
return productService.get(skuId);
} finally {
MDC.clear();
}
}, detailPool);
这段代码写了 6 遍（6 个数据源），很啰嗦。后来我抽了个工具方法 AsyncWrappers.withMdc(Supplier)，稍微好一点，但还是比同步代码丑。
小结
supplyAsync不传 Executor 就用ForkJoinPool.commonPool()，并行度是 CPU 核数 - 1，和 parallelStream 共用。IO 密集任务必须自己配线程池。- 不同业务用不同的线程池做隔离。慢的下游不该拖垮其他异步任务。
thenApply在触发线程或上游线程上执行，线程不确定；thenApplyAsync才提交到池子。转换逻辑重用 Async 版本。thenCompose串联（扁平化嵌套的 Future），thenCombine并联（合并两个 Future 的结果）。allOf返回Void，不聚合结果。要拿数据得逐个getNow(默认值)，配合get(timeout)就能实现超时降级。JDK 8 没有orTimeout，那是 JDK 9 加的。- 异常不会自动外抛，要用
exceptionally/handle/whenComplete处理。allOf里只要有一个失败就整体失败，所以每个子任务都要单独兜底。 - 事务和 ThreadLocal（比如 MDC 的 traceId）在异步线程里会失效，需要手动传递。