5.2 指数退避重试与熔断信号

// Flux<T> 上只有这一个重载
public final Flux<T> retryWhen(Retry retry);   // reactor.util.retry.Retry

AtomicInteger subscribes = new AtomicInteger();

Flux<Integer> alwaysFails = Flux.defer(() -> {
    int n = subscribes.incrementAndGet();
    System.out.println("subscribe#" + n);
    return Flux.error(new RuntimeException("boom" + n));
});

alwaysFails
    .retryWhen(Retry.withThrowable(errors ->
        errors.take(2).delayElements(Duration.ofMillis(20))))
    .subscribe(
        v -> {},
        e -> System.out.println("ONERROR: " + e.getMessage()),
        () -> System.out.println("ONCOMPLETE"));

subscribe#1
subscribe#2
subscribe#3
ONCOMPLETE

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.retry.Retry;

import java.time.Duration;
import java.util.concurrent.atomic.AtomicInteger;

/** 工具类中的静态方法：每个订阅独立计数；第 maxRetries+1 次失败把原错误交给下游 */
static <T> Flux<T> retryWithBackoff(Flux<T> source, int maxRetries, long baseMillis) {
    return source.retryWhen(Retry.withThrowable(errors -> {
        AtomicInteger failures = new AtomicInteger();

        return errors.concatMap(error -> {
            int attempt = failures.incrementAndGet();

            if (attempt > maxRetries) {
                // 关键：companion 以 onError 结束，原始异常才会到达下游
                return Mono.<Long>error(error);
            }

            // 指数退避：base, 2*base, 4*base ...
            long delay = baseMillis * (1L << (attempt - 1));
            System.out.println("第 " + attempt + " 次重试，等待 " + delay + "ms");

            // companion 的 onNext 触发重订阅
            return Mono.delay(Duration.ofMillis(delay))
                       .thenReturn((long) attempt);
        });
    }));
}

subscribe#1 at +0ms
subscribe#2 at +105ms     // 100ms 退避
subscribe#3 at +308ms     // 200ms 退避
subscribe#4 at +712ms     // 400ms 退避
ONERROR: boom4            // 第 4 次失败，原样交给下游

subscribe#1 at +0ms
subscribe#2 at +107ms
subscribe#3 at +310ms
subscribe#4 at +715ms
ONERROR at +715ms: Retries exhausted: 3/3

Mono<Response> call = client.get()
    .retryWhen(Retry.backoff(3, Duration.ofMillis(200))
        .maxBackoff(Duration.ofSeconds(5))     // 单次退避上限
        .multiplier(2.0)                       // 200, 400, 800
        .jitter(0.5)                           // 随机抖动，打散重试时刻
        .filter(this::isTransient)             // 只重试瞬态错误
        .doBeforeRetry(s -> log.warn("第 {} 次重试，原因：{}",
            s.totalRetries() + 1, s.failure().toString()))
        .onRetryExhaustedThrow((spec, signal) ->
            new ServiceUnavailableException(
                "已重试 " + signal.totalRetries() + " 次", signal.failure())));

请求失败？
    ├── 4xx / 参数错误 / 业务校验失败 → 不重试，直接返回
    ├── 瞬态错误（连接重置、5xx、超时）→ 有限次退避重试
    └── 连续多次瞬态失败 → 熔断，走降级

import io.github.resilience4j.reactor.circuitbreaker.operator.CircuitBreakerOperator;

// 写法 A：熔断器在重试外层，每次逻辑调用只记一次结果
client.get()
    .retryWhen(Retry.backoff(3, Duration.ofMillis(200)).filter(this::isTransient))
    .transformDeferred(CircuitBreakerOperator.of(circuitBreaker))
    .onErrorResume(this::fallback);

// 写法 B：熔断器在重试内层，每一次重试尝试都会被记录
client.get()
    .transformDeferred(CircuitBreakerOperator.of(circuitBreaker))
    .retryWhen(Retry.backoff(3, Duration.ofMillis(200)).filter(this::isTransient))
    .onErrorResume(this::fallback);

// 危险：companion 永远不完成，失败一次就重试一次
flux.retryWhen(Retry.withThrowable(errors ->
    errors.flatMap(error -> Mono.delay(Duration.ofSeconds(1)))));

// 1 秒一次的无限重试，由 timeout 兜住总时长，避免无延迟忙等
flux.retryWhen(Retry.fixedDelay(Long.MAX_VALUE, Duration.ofSeconds(1))
        .filter(this::isTransient))
    .timeout(Duration.ofSeconds(30))
    .onErrorResume(e -> fallback());

public final class RetryPolicies {

    private RetryPolicies() {
    }

    public static Retry transientBackoff() {
        return Retry.backoff(3, Duration.ofMillis(200))
            .maxBackoff(Duration.ofSeconds(5))
            .multiplier(2.0)
            .jitter(0.5)
            .filter(RetryPolicies::isTransient)
            .doBeforeRetry(signal -> log.warn(
                "retry {}/3 after {}: {}",
                signal.totalRetries() + 1,
                signal.failure().getClass().getSimpleName(),
                signal.failure().getMessage()))
            .onRetryExhaustedThrow((spec, signal) ->
                new ServiceUnavailableException(
                    "retry exhausted after " + signal.totalRetries() + " retries",
                    signal.failure()));
    }

    private static boolean isTransient(Throwable t) {
        return t instanceof java.io.IOException
            || t instanceof java.util.concurrent.TimeoutException
            || t instanceof TransientServiceException;  // 应用自定义，例如 429 / 5xx 包装
    }
}

Mono<Response> call = client.get()
    .retrieve()
    .bodyToMono(Response.class)
    .retryWhen(RetryPolicies.transientBackoff())
    .onErrorResume(ServiceUnavailableException.class, e -> fallback());