承接的问题
上一节把异步链的异常与取消讲完了，但生产里的调用还有两个维度：时间和失败后的重复。一个借阅通知要扇出到 mail、sms、push，每个下游都可能慢，也都可能连错三次。如果不给整条链路一个总时限，重试会把延迟乘起来：单次 300 ms、重试 3 次，最坏就是 900 ms 再加退避，而调用方的超时可能只有 500 ms。
这一节不用新的 JUC 类，只用 13 章的 CompletableFuture 和两个原子类，把「预算、重试、熔断」拼成一个能落地的组合，且只用 java.util.concurrent 与 java.base。
截止时间只有一个，其余都是推导值
重试逻辑最容易被写错的地方，是把「每次尝试的超时」当成固定值。正确做法是在入口处算一个绝对截止时刻，之后每一步都问「还剩多少」：
long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(budgetMillis);
long remaining = deadline - System.nanoTime();
用 System.nanoTime() 而不是 currentTimeMillis()，是因为前者是单调时钟，不受 NTP 校时和手工改表影响。用墙钟算预算，遇到一次向后校时就会把预算算成负数，或者凭空多出几小时。
有了 deadline，三个量都由它推导：
| 量 | 推导方式 | 不这样做的后果 |
|---|---|---|
| 单次尝试超时 | min(perAttemptTimeout, remaining) | 最后一次尝试可能远超总预算 |
| 退避时长 | min(backoff, remaining - 最小尝试时间) | 退避吃掉全部预算，最后一次根本没跑 |
| 是否继续 | remaining > 0 | 预算耗尽后还在 sleep 和重试 |
下面是完整实现。总预算 300 ms、最多 3 次尝试、退避基数 15 ms 翻倍：
import java.io.IOException;
import java.util.concurrent.*;
public final class RetryBudget {
static final long BUDGET_MILLIS = 300;
static final int MAX_ATTEMPTS = 3;
static final long BASE_BACKOFF_MILLIS = 15;
static final long MIN_ATTEMPT_MILLIS = 10;
static void runRetry(ExecutorService worker) {
long start = System.nanoTime();
long deadline = start + TimeUnit.MILLISECONDS.toNanos(BUDGET_MILLIS);
Exception last = null;
for (int attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
long remaining = deadline - System.nanoTime();
if (remaining <= 0) {
System.out.printf("attempt %d skipped: budget exhausted%n", attempt);
break;
}
if (attempt > 1) {
long backoff = BASE_BACKOFF_MILLIS << (attempt - 2);
long cap = Math.max(0, remaining - TimeUnit.MILLISECONDS.toNanos(MIN_ATTEMPT_MILLIS));
long sleep = Math.min(TimeUnit.MILLISECONDS.toNanos(backoff), cap);
System.out.printf("backoff requested=%d ms capped=%d ms remaining=%.2f ms%n",
backoff, TimeUnit.NANOSECONDS.toMillis(sleep), remaining / 1e6);
sleepNanos(sleep);
remaining = deadline - System.nanoTime();
if (remaining <= 0) {
System.out.printf("attempt %d skipped after backoff%n", attempt);
break;
}
}
long attemptStart = System.nanoTime();
System.out.printf("attempt %d start budget=%.2f ms remaining=%.2f ms%n",
attempt, remaining / 1e6, remaining / 1e6);
try {
String value = invokeOnce(worker, attempt, remaining);
System.out.printf("attempt %d ok used=%.2f ms value=%s%n",
attempt, (System.nanoTime() - attemptStart) / 1e6, value);
return;
} catch (Exception e) {
last = e;
System.out.printf("attempt %d fail used=%.2f ms remaining=%.2f ms cause=%s%n",
attempt, (System.nanoTime() - attemptStart) / 1e6,
(deadline - System.nanoTime()) / 1e6, root(e).getClass().getSimpleName());
}
}
System.out.printf("give up after %.2f ms, last=%s%n",
(System.nanoTime() - start) / 1e6,
last == null ? "none" : root(last).getClass().getSimpleName());
}
static String invokeOnce(ExecutorService worker, int attempt, long budgetNanos)
throws Exception {
Callable<String> call = () -> {
Thread.sleep(switch (attempt) { case 1 -> 80; case 2 -> 90; default -> 120; });
throw new IOException("downstream 503");
};
try {
return CompletableFuture.supplyAsync(() -> {
try { return call.call(); }
catch (Exception e) { throw new RuntimeException(e); }
}, worker).orTimeout(budgetNanos, TimeUnit.NANOSECONDS).get();
} catch (ExecutionException e) {
throw new Exception(root(e));
}
}
static Throwable root(Throwable t) {
while (t.getCause() != null && t.getCause() != t) t = t.getCause();
return t;
}
static void sleepNanos(long nanos) {
long end = System.nanoTime() + nanos;
boolean interrupted = false;
while (System.nanoTime() < end) {
try { TimeUnit.NANOSECONDS.sleep(end - System.nanoTime()); }
catch (InterruptedException e) { interrupted = true; }
}
if (interrupted) Thread.currentThread().interrupt();
}
}
本机真实输出（单次运行，机器 load average 113，耗时会随调度波动，不作基准）：
attempt 1 start budget=299.98 ms remaining=299.98 ms
attempt 1 fail used=112.94 ms remaining=187.04 ms cause=IOException
backoff requested=15 ms capped=15 ms remaining=186.68 ms
attempt 2 start budget=170.22 ms remaining=170.22 ms
attempt 2 fail used=91.62 ms remaining=78.60 ms cause=IOException
backoff requested=30 ms capped=30 ms remaining=78.28 ms
attempt 3 start budget=42.24 ms remaining=42.24 ms
attempt 3 fail used=50.01 ms remaining=-7.76 ms cause=TimeoutException
give up after 308.11 ms, last=TimeoutException
三处值得对照着看。
第一，假下游第一次只 sleep(80)，实测却用了 112.94 ms。这就是为什么预算必须从 nanoTime 实测里扣，而不能按「每次 80 ms、三次 240 ms」推算。
第二，第三次尝试的预算只有 42.24 ms，而代码里写的是 sleep(120)。orTimeout 在 42 ms 处把 Future 置为超时，多出来的 7.76 ms 是定时任务被调度执行的延迟。于是「最多 3 次」在预算面前自动降级成「2 次失败 + 1 次几乎肯定失败」。
第三，总耗时 308.11 ms，超出预算 8.11 ms。超出的部分来自超时定时器本身的调度延迟，不是预算算错。要求严格不超时，只能把总预算设得比调用方超时少一截（常见的 10% 余量）。
退避要用剩余预算裁剪
指数退避的公式是 base << (attempt - 2)：15、30、60、120……如果不裁剪，第 4 次重试前就要睡 120 ms，而总预算可能只剩 30 ms。裁剪的写法是：
long cap = Math.max(0, remaining - MIN_ATTEMPT_MILLIS_NANOS);
long sleep = Math.min(backoffNanos, cap);
MIN_ATTEMPT_MILLIS 是「留给这次尝试的最低预算」。没有它，退避可以精确吃掉所有剩余时间，然后第三次尝试以 0 ms 预算发起、必然超时——等于用一次真实的下游调用换了一个必然的失败。留 10 ms 虽然也短，但至少给了快速失败的下游一次机会。
上一条输出里退避没有触发裁剪（15 和 30 都完整睡完），是因为前两次尝试只花了 204 ms。如果下游再慢一点，就会看到 capped 小于 requested。
还有一个细节：Thread.sleep 的精度不保证。上面 sleepNanos 用循环重算剩余时间，是因为一次 TimeUnit.NANOSECONDS.sleep(30ms) 在高负载下可能睡 45 ms；单次 sleep 后直接减预算，会让「剩余」比实际偏大，后续尝试又会超时。
orTimeout 只负责「结果不等人」
JDK 25 的 orTimeout 只有两行：
// JDK 25 源码摘录，已删减
public CompletableFuture<T> orTimeout(long timeout, TimeUnit unit) {
arrangeTimeout(unit.toNanos(timeout), new Timeout<T>(this, null, true));
return this;
}
private <U> void arrangeTimeout(long nanoDelay, Timeout<U> onTimeout) {
ForkJoinPool e = ASYNC_POOL;
if (result == null) {
ScheduledForkJoinTask<Void> t =
new ScheduledForkJoinTask<Void>(nanoDelay, 0L, true, onTimeout, null, e);
whenComplete(new Canceller(t));
e.scheduleDelayedTask(t);
}
}
static final class Timeout<U> implements Runnable {
public void run() {
if (f != null && !f.isDone()) {
if (exceptional) f.completeExceptionally(new TimeoutException());
else f.complete(value);
}
}
}
Timeout.run() 只做一件事：如果 Future 还没完成，就把它标记为异常完成。它不会中断正在执行的任务，也不会取消它——任务在哪个线程跑、跑到哪一步，CompletableFuture 完全不知道。Canceller 的作用是反方向：当结果正常返回时，取消那个已经没用的定时任务，避免定时器堆积。
本机验证（同一个单线程执行器，任务睡 400 ms，orTimeout(100 ms)）：
[main] got TimeoutException at 104.82 ms, future cancelled=false
[worker] task still running, finished at 410.39 ms
[main] after wait, future done=true cancelled=false
主线程在 104.82 ms 拿到 TimeoutException，工作线程在 410.39 ms 仍然把任务跑完并打印了日志。isCancelled() 全程是 false。
这个语义对重试有直接影响：超时的那次尝试并没有停止。如果它写数据库、发消息、扣库存，重试就会与它并发执行，产生重复副作用。所以重试的下游操作必须幂等（带请求 ID 去重），或者像 13.3 讲的那样显式持有 Future 并 cancel(true)——但 CompletableFuture.cancel(true) 同样不会中断底层任务，只是把结果标记为取消。真正能中断的只有自己持有 Thread 或在任务内部检查中断标志。
熔断器是三个状态加两个计数
熔断器解决的问题是「下游已经坏了，别再拿重试去打它」。三态语义：
CLOSED：正常放行，累计失败数；OPEN：全部拒绝，直到冷却期结束；HALF_OPEN：只放行一个试探请求，成功回CLOSED，失败回OPEN。
状态用 AtomicReference<State>，计数和开闸时刻用 AtomicLong：
enum State { CLOSED, OPEN, HALF_OPEN }
final class CircuitBreaker {
private final AtomicReference<State> state = new AtomicReference<>(State.CLOSED);
private final AtomicLong failures = new AtomicLong();
private final AtomicLong openUntil = new AtomicLong();
private final int failureThreshold;
private final long openNanos;
CircuitBreaker(int failureThreshold, long openMillis) {
this.failureThreshold = failureThreshold;
this.openNanos = TimeUnit.MILLISECONDS.toNanos(openMillis);
}
boolean tryAcquire() {
State now = state.get();
if (now == State.CLOSED) return true;
if (now == State.OPEN) {
if (System.nanoTime() - openUntil.get() < 0) return false;
return state.compareAndSet(State.OPEN, State.HALF_OPEN);
}
return false; // HALF_OPEN 只放行抢到迁移的那一个线程
}
void onFailure() {
if (state.get() == State.HALF_OPEN) {
if (state.compareAndSet(State.HALF_OPEN, State.OPEN)) {
openUntil.set(System.nanoTime() + openNanos);
}
return;
}
if (state.get() == State.CLOSED) {
long f = failures.incrementAndGet();
if (f >= failureThreshold && state.compareAndSet(State.CLOSED, State.OPEN)) {
failures.set(0);
openUntil.set(System.nanoTime() + openNanos);
}
}
}
void onSuccess() {
failures.set(0);
state.compareAndSet(State.HALF_OPEN, State.CLOSED);
}
}
状态迁移必须用 CAS，不能先读再写。 冷却期结束时可能同时有几十个线程看到 OPEN 已过期，谁 CAS 成功谁去试探，其余全部返回 false。如果写成 if (state.get() == OPEN && expired) { state.set(HALF_OPEN); return true; }，所有线程都会放行，半开就变成了全开。
failures 用 AtomicLong 而不是普通计数，并配合阈值判断。 阈值判断本身有竞态：多个线程可能同时看到 failures >= 3，但只有一个能 CAS 成功，其余继续按 CLOSED 处理。多记一两次失败无害，错误地关闭熔断才有害。
本机状态迁移日志（阈值 3 次、冷却 200 ms，单次运行）：
[ 26.6 ms] CLOSED request 2 executing (simulated failure)
[ 27.5 ms] OPEN CLOSED -> OPEN, failures=3
[ 27.7 ms] OPEN request 4 rejected
[ 27.9 ms] OPEN request 5 rejected
[ 248.6 ms] HALF_OPEN OPEN -> HALF_OPEN, one trial allowed
[ 249.3 ms] OPEN HALF_OPEN -> OPEN, trial failed
[ 477.0 ms] HALF_OPEN OPEN -> HALF_OPEN, one trial allowed
[ 477.7 ms] CLOSED HALF_OPEN -> CLOSED, failures reset from 0
[ 478.1 ms] CLOSED CLOSED, failure counted, failures=1
[ 478.6 ms] CLOSED CLOSED, success, failures reset from 1
注意日志前缀打的是迁移之后的状态，所以 OPEN -> HALF_OPEN 那行前缀是 HALF_OPEN。半开试探失败后冷却时间重新计算，这一点很重要：否则一个持续故障的下游会被每 200 ms 探测一次，等于没有熔断。
在 libraryd-conc 里落地
借阅成功的通知扇出最适合套这套组合。三个下游性格不同：mail 慢但可靠，sms 便宜但偶发 503，push 由第三方托管、故障时整段时间不可用。给每个下游独立的预算和熔断器，而不是全局一个：
package com.plume.conc.core;
import java.util.concurrent.*;
public record DownstreamGuard(String name, CircuitBreaker breaker,
ExecutorService worker, long budgetNanos) {
public void notify(String isbn, Runnable action) {
if (!breaker().tryAcquire()) {
throw new IllegalStateException(name() + " circuit open, dropped: " + isbn);
}
long deadline = System.nanoTime() + budgetNanos();
try {
for (int attempt = 1; ; attempt++) {
long remaining = deadline - System.nanoTime();
if (remaining <= 0 || attempt > 3) {
throw new IllegalStateException(name() + " budget exhausted");
}
try {
CompletableFuture.runAsync(action, worker())
.orTimeout(remaining, TimeUnit.NANOSECONDS).join();
breaker().onSuccess();
return;
} catch (CompletionException e) {
long backoff = Math.min(15L << (attempt - 1), 200L) * 1_000_000L;
sleepQuietly(Math.min(backoff, Math.max(0, remaining - 10_000_000L)));
}
}
} catch (RuntimeException e) {
breaker().onFailure();
throw e;
}
}
private static void sleepQuietly(long nanos) {
try {
Thread.sleep(nanos / 1_000_000L, (int) (nanos % 1_000_000L));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
mail 可以给 2 s 预算、5 次失败跳闸；sms 给 800 ms、3 次跳闸；push 给 500 ms、2 次跳闸。预算和阈值都应该按下游的 SLA 定，不要全站一个值。
注意 join() 抛的是 CompletionException，捕获它比捕获 ExecutionException 更贴近 runAsync 的语义；orTimeout 注入的 TimeoutException 也会被包在 CompletionException 里。这个包装关系在 13.3 已经拆过。
这个设计的代价
预算是一种全局资源，重试会放大下游压力。 一次 300 ms 的预算里塞 3 次尝试，等于在故障时把 QPS 放大 3 倍。熔断器是对冲手段，但它的阈值要按「下游能承受的探测频率」设，不是按业务失败率设。
orTimeout 的超时不是中止。 前面已经验证：超时后任务还在跑。对非幂等的写操作，重试必须配去重键，否则会写出两条借阅记录。这一点在第 15 章的组合陷阱里还会再出现一次。
熔断器按实例计数。 每个 JVM 一个熔断器，10 个实例就是 10 份独立判断，下游已经挂了，每个实例仍要各自失败 5 次才跳闸。要全局一致，需要把状态外置到共享存储，那就超出了 JUC 的范围。
什么时候不要用
不要在单次调用本身就是幂等且廉价的场景上加熔断。读缓存、健康检查这类操作，失败一次重试一次就够了，熔断只会把一次抖动放大成一段拒绝服务。
不要把重试和熔断同时配成「激进」参数：3 次重试 + 3 次失败跳闸，意味着一次真实的用户请求就能把下游打成 OPEN，恢复期内所有请求都被拒。经验值是 failureThreshold >= maxAttempts + 1。
不要用 orTimeout 去实现「取消」。它还不了你一个停止的任务，只还你一个提前完成的结果。要真正停止，得让任务自己响应中断，或者用 13 章讲过的显式 Future.cancel(true) 加任务内检查。
最后，预算不要设在业务层的最外层。最外层设的是用户可接受的延迟（例如 800 ms），下游每一跳分到的是剩余预算的一部分；所有层都用同一个 800 ms，总延迟会变成层数乘以 800 ms。
小结
超时重试的核心不是 API，而是「一个截止时刻、处处推导」。nanoTime 提供单调时间，剩余预算裁剪单次超时和退避，orTimeout 只负责让结果提前返回、不负责停任务。熔断器是 AtomicReference 上的三态 CAS 加两个计数器，半开状态靠一次 CAS 保证只放一个试探。把预算和熔断按下游拆开，才算真正落地。
阅读导航：上一节 13.3 异常传播与取消 ｜ 下一节 14.2 批量聚合与背压
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。