商品详情页要同时拿到价格、库存、促销、评价四份数据，四个下游接口串行调用 800ms 起步，改成并行之后 P99 掉到 220ms。但并行带来的新问题是：只要有一个接口卡住，整页就跟着卡住。给每个 future 单独 orTimeout 只能解决单点超时，它抛出的 TimeoutException 会让 allOf 立刻异常结束，其余还在跑的请求白白浪费，而且调用方拿不到任何已经成功的数据。这篇文章解决的就是这个组合问题：聚合、降级、统一兜底三件事怎么在一个方法里干净地做完。
先看清 allOf 的真实行为：它只负责”都完成”，不负责”结果”
很多同学第一次用 CompletableFuture.allOf 会踩两个坑，必须先说清楚。
第一个坑：allOf 返回的是 CompletableFuture<Void>，它不携带任何结果，你必须回到各个 future 上调用 join() 取值。第二个坑：只要有任意一个 future 以异常结束，allOf 返回的 future 也会以 CompletionException 结束，此时去 join() 其它 future 是安全的（它们照常返回结果或抛自己的异常），但你拿不到”整体成功”这个信号。
还有一个容易被忽略的点：allOf 的完成不会取消其它还在执行的 future。发生异常时，剩下的 HTTP 请求照样打到下游，线程照样占着。这就是为什么单纯在 allOf().join() 外面包一层 try-catch 是不够的——它既不兜底也不止损。
下面这段代码演示了原生行为，可以直接跑：
import java.util.concurrent.*;
public class AllOfBaseline {
public static void main(String[] args) throws Exception {
ExecutorService pool = Executors.newFixedThreadPool(4);
CompletableFuture<String> ok = CompletableFuture.supplyAsync(() -> {
sleep(100);
return "price=99";
}, pool);
CompletableFuture<String> boom = CompletableFuture.supplyAsync(() -> {
sleep(200);
throw new IllegalStateException("stock service 503");
}, pool);
CompletableFuture<String> slow = CompletableFuture.supplyAsync(() -> {
sleep(3000);
return "promo=none";
}, pool);
CompletableFuture<Void> all = CompletableFuture.allOf(ok, boom, slow);
try {
all.join();
} catch (CompletionException e) {
System.out.println("allOf 抛出的包装异常: " + e.getCause().getMessage());
}
// 关键点：slow 仍在跑，allOf 不会取消它
System.out.println("ok 的结果仍可读取: " + ok.join());
pool.shutdownNow();
}
static void sleep(long ms) {
try { Thread.sleep(ms); } catch (InterruptedException e) { Thread.currentThread().interrupt(); }
}
}
输出会是这样：
allOf 抛出的包装异常: stock service 503
ok 的结果仍可读取: price=99
另外注意，join() 抛的是非受检的 CompletionException，而 get() 抛的是受检的 ExecutionException，两者都会把原始异常包在 getCause() 里。如果你在 catch 里直接 e.getMessage() 打日志，日志里看到的是 java.lang.IllegalStateException: stock service 503 这一整串，容易被误判成异常类型。统一用 e.getCause() 取原始异常是更稳的写法。
单点 orTimeout 为什么不等于统一兜底
orTimeout(long, TimeUnit) 是 Java 9 引入的，内部基于 Delayer 和 ScheduledThreadPoolExecutor，超时时让 future 以 TimeoutException 完成。它比 completeOnTimeout 更适合我们，因为后者只给默认值不报警，前者能留下超时痕迹。但它有两个硬伤：
- 各自超时等于各自失败。四个接口都设 500ms，只要一个超时，
allOf立刻异常，另外三个即使 300ms 就成功了也白跑。 - 没有总时间预算。四个接口各自 500ms，理论上总耗时还是 500ms，但线程池排队、DNS 解析、GC 停顿这些不在接口超时里的开销会累积，实际可能远超预期。
正确的做法是分两层：个体超时负责控制单个依赖的最坏耗时，整体兜底负责控制调用方的总预算，并在总预算耗尽时把”已完成的部分”和”未完成的部分”分开处理。下面进入实现。
封装一个 Aggregator：个体超时 + 整体兜底 + 部分结果
设计目标有三个：任意一个任务失败或超时都返回默认值而不是让整体挂掉；整体总耗时不超过给定预算；调用方能拿到每个任务的状态（成功 / 失败 / 超时）。
第一步：定义结果容器与任务注册
用一个 Task<T> 描述”一个具名任务”，用一个 Item<T> 承载三种状态。这里不用泛型通配一把梭，而是保留每个任务自己的类型，避免调用方还要强转。
import java.util.concurrent.*;
import java.util.function.Supplier;
public final class Aggregator {
public enum Status { SUCCESS, FAILED, TIMEOUT }
public static final class Item<T> {
public final String name;
public final Status status;
public final T value; // SUCCESS 时非空
public final Throwable error; // FAILED 时非空
private Item(String name, Status status, T value, Throwable error) {
this.name = name;
this.status = status;
this.value = value;
this.error = error;
}
static <T> Item<T> ok(String name, T v) {
return new Item<>(name, Status.SUCCESS, v, null);
}
static <T> Item<T> failed(String name, Throwable e) {
return new Item<>(name, Status.FAILED, null, e);
}
static <T> Item<T> timeout(String name) {
return new Item<>(name, Status.TIMEOUT, null,
new TimeoutException(name + " timed out"));
}
@Override
public String toString() {
return "Item{" + name + ", " + status + ", value=" + value + "}";
}
}
/** 一个待聚合的具名任务 */
public static final class Task<T> {
final String name;
final Supplier<T> supplier;
final long timeoutMs;
final T fallback;
public Task(String name, Supplier<T> supplier, long timeoutMs, T fallback) {
this.name = name;
this.supplier = supplier;
this.timeoutMs = timeoutMs;
this.fallback = fallback;
}
}
}
注意 fallback 是必需的，不能给 null 就完事。业务上”价格为 0″和”价格未知”是两回事，让调用方显式提供默认值，可以在编译期逼着人想清楚降级语义。
第二步：核心聚合方法
要点：每个任务用 orTimeout 控个体时间，用 handle 把异常转成 Item（这样 allOf 永远不会因异常提前结束），整体再用 orTimeout 控总预算，最后用 getNow 收割已完成的结果，未完成的统一标记 TIMEOUT 并取消。
import java.util.*;
import java.util.concurrent.*;
public final class Aggregator {
// 省略上一节的状态枚举与 Item/Task 定义
/**
* 并行执行所有任务，带个体超时与整体兜底。
* 无论单个任务成功、失败还是超时，本方法都会在 totalTimeoutMs 内返回。
*/
public static <T> List<Item<?>> runAll(List<Task<?>> tasks,
long totalTimeoutMs,
Executor executor) {
// 每个任务包成 CompletableFuture>，异常已在 handle 中消化
List<CompletableFuture<Item<?>>> futures = new ArrayList<>(tasks.size());
List<CompletableFuture<?>> raw = new ArrayList<>(tasks.size());
for (Task<?> t : tasks) {
CompletableFuture<Item<?>> f = CompletableFuture
.supplyAsync(t.supplier::get, executor)
.orTimeout(t.timeoutMs, TimeUnit.MILLISECONDS)
.handle((v, ex) -> {
if (ex == null) {
return Item.ok(t.name, v);
}
Throwable cause = (ex instanceof CompletionException && ex.getCause() != null)
? ex.getCause() : ex;
if (cause instanceof TimeoutException) {
return Item.timeout(t.name);
}
return Item.failed(t.name, cause);
});
futures.add(f);
raw.add(f);
}
CompletableFuture<Void> all = CompletableFuture
.allOf(raw.toArray(new CompletableFuture<?>[0]));
try {
// 整体预算：即使某个 future 因为底层不响应取消而迟迟不结束，这里也一定会超时返回
all.get(totalTimeoutMs, TimeUnit.MILLISECONDS);
} catch (TimeoutException e) {
// 预期内：走兜底逻辑，不打印成异常栈
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} catch (ExecutionException e) {
// 正常情况下不会走到这里：handle 已消化所有异常
}
List<Item<?>> result = new ArrayList<>(futures.size());
for (int i = 0; i < futures.size(); i++) {
CompletableFuture<Item<?>> f = futures.get(i);
if (f.isDone() && !f.isCompletedExceptionally() && !f.isCancelled()) {
result.add(f.getNow(null)); // 不阻塞，已完成才会走到这里
} else {
Task<?> t = tasks.get(i);
result.add(Item.timeout(t.name)); // 未完成，按超时处理
f.cancel(true); // 尽力止损，能否真中断取决于底层实现
}
}
return result;
}
}
几个必须解释的细节：
- 为什么
allOf外面还要get(timeout)？ 因为orTimeout只保证 future 在超时后”完成”，但如果supplier内部是阻塞在Thread.sleep或者一个不响应中断的 socket 读上，orTimeout依然能按时触发，可如果线程池被打满、任务还排在队列里没开始执行，supplyAsync返回的 future 就不会有任何进展。整体get超时是兜底的兜底，保证调用方线程有确定的返回时间。 getNow(null)与isDone()配合。到这里所有 future 要么已完成要么已超时，getNow不会阻塞。不要用join()，否则一旦判断失误就会阻塞在调用线程上，兜底就失效了。cancel(true)的真实效果。它只向运行线程发中断信号，HttpClient的同步send、JDBC 查询、Thread.sleep会响应中断并抛InterruptedException；Netty/Reactor 这类基于回调的调用基本不响应。ExecutorService.shutdownNow同理。所以要清楚：cancel是尽力而为，不是保证。
第三步：独立线程池，别和业务线程池混用
聚合调用必须用专门的小线程池，原因有两个：聚合任务本身会阻塞（等下游 IO），不适合放 ForkJoinPool.commonPool()；一旦下游全部变慢，共用池会被撑爆，导致同一 Pool 里的其它业务一起排队，形成级联故障。
import java.util.concurrent.*;
public final class AggregatorPool {
private static final ThreadPoolExecutor POOL = new ThreadPoolExecutor(
16, // corePoolSize：按 P99 并发量定，不是按 QPS
16,
60L, TimeUnit.SECONDS,
new LinkedBlockingQueue<>(2000), // 有界队列，满了触发拒绝，快速失败
new ThreadFactory() {
private int seq = 0;
@Override
public synchronized Thread newThread(Runnable r) {
Thread t = new Thread(r, "aggregate-" + (seq++));
t.setDaemon(true);
return t;
}
},
new ThreadPoolExecutor.CallerRunsPolicy()); // 让调用线程自己跑，形成背压
public static Executor get() {
return POOL;
}
}
为什么选 CallerRunsPolicy：队列满时新任务由提交者线程执行，页面渲染变慢但不会丢数据，同时天然把压力传导回请求层，触发上游限流。相比 AbortPolicy 直接抛 RejectedExecutionException，在商品详情这种”宁可慢一点也别报错”的场景更合适。
第四步：组装调用
import java.util.*;
public class ProductDetailService {
public Map<String, Object> load(long skuId) {
List<Aggregator.Task<?>> tasks = Arrays.asList(
new Aggregator.Task<>("price",
() -> priceClient.query(skuId), 300, 0L),
new Aggregator.Task<>("stock",
() -> stockClient.query(skuId), 300, -1),
new Aggregator.Task<>("promo",
() -> promoClient.query(skuId), 500, Collections.emptyList()),
new Aggregator.Task<>("review",
() -> reviewClient.count(skuId), 500, 0L)
);
// 总预算 600ms：给最慢的接口留 100ms 裕量，超出部分一律降级
List<Aggregator.Item<?>> items = Aggregator.runAll(tasks, 600, AggregatorPool.get());
Map<String, Object> data = new HashMap<>();
for (Aggregator.Item<?> item : items) {
if (item.status == Aggregator.Status.SUCCESS) {
data.put(item.name, item.value);
} else {
// 记录降级，方便后续埋点统计哪些接口最常超时
System.out.println("degrade " + item.name + " status=" + item.status);
data.put(item.name, fallbackOf(item.name));
}
}
return data;
}
private Object fallbackOf(String name) {
switch (name) {
case "price": return 0L;
case "stock": return -1;
case "promo": return Collections.emptyList();
case "review": return 0L;
default: return null;
}
}
}
这里 600ms 的总预算怎么定：先看各接口的 P99，取”最慢两个并行”+ 200ms 左右的裕量。如果四个接口 P99 分别是 120/150/480/450ms，并行后理论 P99 约 480ms，加上线程调度和序列化，600ms 是合理的。不要拍脑袋写 1000ms，兜底时间越长，雪崩时打回来的请求积压越久。
验证与踩坑：三个真实报错
报错一：java.util.concurrent.TimeoutException 被当成业务异常上报
现象：监控里 TimeoutException 的告警量暴涨，但接口成功率没掉。原因就是 orTimeout 抛出的 TimeoutException 在 handle 里被归类为普通失败，直接进了 FAILED 分支并上报。修法就是前面代码里的显式判断：
Throwable cause = (ex instanceof CompletionException && ex.getCause() != null)
? ex.getCause() : ex;
if (cause instanceof TimeoutException) {
return Item.timeout(t.name); // 单独状态，走降级指标而不是异常指标
}
return Item.failed(t.name, cause);
顺带一提，Java 9 的 orTimeout 抛的就是 java.util.concurrent.TimeoutException，而 CompletableFuture.get(timeout) 抛的是同一个类，两者在日志里长得一模一样，务必在异常信息里带上任务名（上面 Item.timeout 的 new TimeoutException(name + " timed out") 就是为了这个）。
报错二：java.util.concurrent.CompletionException: java.util.concurrent.RejectedExecutionException
现象：大促压测时大量请求直接 500。栈顶是 supplyAsync 里的 RejectedExecutionException，说明线程池拒绝策略生效了。这里有个坑：supplyAsync 提交被拒时抛的 RejectedExecutionException 是在调用 supplyAsync 的那一行同步抛出的，不会进到 handle 里，所以 Aggregator.runAll 的循环会直接中断，前面的 future 成了孤儿。解决办法是在提交处单独 try-catch，把提交失败转成 Item.failed：
CompletableFuture<Item<?>> f;
try {
f = CompletableFuture.supplyAsync(t.supplier::get, executor)
.orTimeout(t.timeoutMs, TimeUnit.MILLISECONDS)
.handle((v, ex) -> { /* 同上 */ return Item.ok(t.name, v); });
} catch (RejectedExecutionException ree) {
f = CompletableFuture.completedFuture(Item.failed(t.name, ree));
}
当然更根本的做法是配合前面 CallerRunsPolicy 让拒绝变成背压。但如果用的是 AbortPolicy，这段 try-catch 就是必须的。
报错三：压测时聚合方法整体耗时 3s+，日志显示 getNow 返回了 null
现象：日志里出现 Item{price, SUCCESS, value=null}，业务侧把 null 当成”价格 0 元”展示，引发资损。原因是 supplier 正常返回了 null，handle 里 ex == null 判断成立，于是转成 SUCCESS。修复：handle 里判空，把 null 视为降级。
.handle((v, ex) -> {
if (ex != null) { /* 处理异常 */ }
if (v == null) {
return Item.failed(t.name, new IllegalStateException(t.name + " returned null"));
}
return Item.ok(t.name, v);
});
还有一类耗时问题来自 orTimeout 的实现机制：它内部用 CompletableFuture.delayer 这个静态的 ScheduledThreadPoolExecutor，在 Java 9~11 早期版本中，大量 future 同时挂超时任务会在这个调度线程上产生竞争。JDK-8248038 修复过相关问题。若你的环境是 JDK 8，根本没有 orTimeout，需要自己用 ScheduledExecutorService 手动 completeExceptionally，写法如下：
static <T> CompletableFuture<T> withTimeout(CompletableFuture<T> f,
long ms,
ScheduledExecutorService scheduler) {
ScheduledFuture<?> timer = scheduler.schedule(() -> {
f.completeExceptionally(new TimeoutException("timeout after " + ms + "ms"));
}, ms, TimeUnit.MILLISECONDS);
// 正常完成时取消定时任务，避免调度队列堆积
f.whenComplete((v, e) -> timer.cancel(false));
return f;
}
注意 whenComplete 里取消定时器这一步不能省，否则每次调用都在调度器里留一个待触发的任务，高频下会看到 ScheduledThreadPoolExecutor 的队列长度持续增长。
小结
三个层次各司其职：orTimeout 控单接口最坏耗时，handle 把异常和超时转成可降级的 Item 从而让 allOf 永远不异常，get(totalTimeout) 给出调用方看到的总时延上界。getNow 收割已完成的部分，没完成的按超时计并 cancel 止损。线程池要独立、有界、带背压策略。这套组合下来，聚合接口在任意下游抖动时都能在预算内返回可用数据，业务侧看到的是”某个字段降级”而不是”整页 500″。
FAQ
Q1：所有 future 都成功了，还需要调 get(totalTimeout) 吗？
需要，但代价极低。因为 allOf 在所有 future 完成时立即完成，get 会立刻返回，不会真的等满 totalTimeoutMs。这一行的作用是在异常路径上给调用方一个确定的时间上界，正常路径几乎零开销。
Q2：为什么用 get 而不是 join？
join 无法指定超时，一旦前面有 future 因为线程池排队始终没开始执行，join 会无限阻塞，兜底完全失效。另外 join 抛的是 CompletionException，在需要区分超时和中断的场景不如 get 清晰。
Q3：cancel(true) 能真正停掉下游 HTTP 请求吗？
取决于客户端。JDK 11+ 的 java.net.http.HttpClient 同步 send 在中断时会抛 InterruptedException 并关闭连接；HttpURLConnection 在某些阶段不响应中断；基于 Netty 的异步客户端通常需要显式 cancel 掉自己的 future 才能断开连接。所以 cancel 只能当”降低无用功”的优化，不能当资源回收的保证，真正的资源兜底靠线程池上限和连接池上限。
Q4：任务数量不确定（比如按 SKU 数量动态展开）时还能用吗？
可以用，但要注意两件事。第一，线程池上限决定了真正并行的任务数，任务数量远大于线程数时，排在队列里的任务可能整体超时前一个都没开始，表现为所有 item 都是 TIMEOUT。第二，Task 里的 timeoutMs 应该按”单任务最长耗时”设，而不是平均耗时，否则大量任务会正常超时。任务数量超过 20 个时建议先用 CompletableFuture 分批（比如 8 个一组），再对批次做聚合。
Q5：能不能直接在 allOf 上 orTimeout 省掉 get？
可以，allOf(...).orTimeout(total, MILLISECONDS).join() 在语义上接近，但有两个差别：orTimeout 到点后 join 会抛 CompletionException 包装的 TimeoutException，需要额外解包；更关键的是，orTimeout 依赖内部 delayer 线程按时调度，极端情况下（调度器繁忙、系统时钟被调整）不如 get(timeout) 直接基于条件等待来得可靠。生产代码里我更倾向 get(timeout)。