Spring WebFlux 中实现基于响应头的动态重试退避时间

本文介绍如何在 Spring WebFlux 中根据 HTTP 响应（如 Retry-After 头或自定义状态码 106 及其携带的 retryTime）动态调整重试间隔，替代固定退避策略，提升对服务端限流/降级策略的适应性。

本文介绍如何在 Spring WebFlux 中根据 HTTP 响应（如 `Retry-After` 头或自定义状态码 106 及其携带的 `retryTime`）动态调整重试间隔，替代固定退避策略，提升对服务端限流/降级策略的适应性。

在 Spring WebFlux 的响应式编程模型中，RetryBackoffSpec 提供了简洁的指数退避重试能力，但其退避时长（如 backoff(5, ofSeconds(10))）是静态预设的，无法感知上游服务返回的动态重试建议（例如 HTTP 429 Too Many Requests 响应中的 Retry-After 头，或自定义错误码 106 中附带的 retryTime: 37 秒）。为实现真正“服务端驱动”的弹性重试，需绕过 RetryBackoffSpec，转而使用底层 Retry 抽象并自定义 generateCompanion 逻辑。

核心思路是：基于 RetrySignal 中的异常信息（如 WebClientResponseException），解析响应头或响应体，提取服务端指定的重试延迟值，并据此动态生成 Mono.delay(...) 作为下一次重试的触发时机。

以下是一个生产就绪的 DynamicRetrySpec 实现，支持两种常见场景：

• ✅ 标准 429 Too Many Requests + Retry-After（秒级整数）
• ✅ 自定义业务错误码（如 106）+ 自定义响应体字段（如 { "code": 106, "retryTime": 45 }）

private static class DynamicRetrySpec extends Retry {
    private final int maxRetries;
    private final Duration defaultBackoff;
    private final ObjectMapper objectMapper; // 用于解析 JSON 响应体（可选）

    public DynamicRetrySpec(int maxRetries, Duration defaultBackoff) {
        this(maxRetries, defaultBackoff, new ObjectMapper());
    }

    public DynamicRetrySpec(int maxRetries, Duration defaultBackoff, ObjectMapper objectMapper) {
        this.maxRetries = maxRetries;
        this.defaultBackoff = defaultBackoff;
        this.objectMapper = objectMapper;
    }

    @Override
    public Publisher<?> generateCompanion(Flux<RetrySignal> retrySignals) {
        return retrySignals.flatMap(this::getRetry);
    }

    private Mono<Long> getRetry(Retry.RetrySignal rs) {
        Throwable failure = rs.failure();
        int currentRetry = rs.totalRetries();

        if (currentRetry >= maxRetries) {
            log.warn("Max retries ({}) exhausted for error: {}", maxRetries, failure.getMessage());
            throw Exceptions.propagate(failure);
        }

        Duration delay = Duration.ZERO;
        if (failure instanceof WebClientResponseException webEx) {
            // 场景1：检查 Retry-After 头（标准 429）
            String retryAfter = webEx.getHeaders().getFirst("Retry-After");
            if (retryAfter != null && !retryAfter.trim().isEmpty()) {
                try {
                    delay = Duration.ofSeconds(Long.parseLong(retryAfter.trim()));
                    log.debug("Using Retry-After header: {}s", delay.getSeconds());
                } catch (NumberFormatException ignored) {
                    // 若非数字，忽略，走默认回退
                }
            }

            // 场景2：自定义错误码 106，解析响应体中的 retryTime
            if (delay.isZero() && webEx.getRawStatusCode() == 106) {
                delay = parseRetryTimeFromBody(webEx.getResponseBodyAsString());
            }
        }

        // 若未从响应中提取到有效 delay，则使用默认退避
        if (delay.isZero()) {
            delay = defaultBackoff;
        }

        log.info("Retry #{} scheduled in {}ms (status: {}, reason: {})",
                currentRetry + 1, delay.toMillis(), 
                failure instanceof WebClientResponseException ? 
                    ((WebClientResponseException) failure).getRawStatusCode() : "N/A",
                failure.getClass().getSimpleName());

        return Mono.delay(delay).thenReturn((long) currentRetry);
    }

    private Duration parseRetryTimeFromBody(String body) {
        if (body == null || body.trim().isEmpty()) return Duration.ZERO;
        try {
            JsonNode node = objectMapper.readTree(body);
            int retryTime = node.path("retryTime").asInt(-1);
            if (retryTime > 0) {
                return Duration.ofSeconds(retryTime);
            }
        } catch (Exception e) {
            log.warn("Failed to parse retryTime from response body", e);
        }
        return Duration.ZERO;
    }
}

使用方式（集成到 WebClient 链路）：

webClient.post()
    .uri("/v1/device/grant")
    .bodyValue(params)
    .retrieve()
    .onStatus(status -> status.value() == 106, clientResponse ->
        clientResponse.bodyToMono(String.class)
            .map(body -> new WebClientResponseException(106, "Custom Retry Required", 
                clientResponse.headers().asHttpHeaders(), body.getBytes(), StandardCharsets.UTF_8)))
    .bodyToMono(MyResponse.class)
    .retryWhen(new DynamicRetrySpec(3, Duration.ofSeconds(5)))
    .doOnSuccess(result -> handleResponse(result, grantMapEntry, deviceInfo))
    .onErrorResume(throwable -> {
        log.error("All retries failed for device request", throwable);
        return Mono.empty(); // 或发布失败事件
    })
    .subscribe();

⚠️ 关键注意事项：

• RetrySignal.failure() 必须是 WebClientResponseException 类型才能获取响应头/体；确保你的错误传播路径未被意外包装（如避免 Mono.error(new RuntimeException(ex))）。
• 对于非 4xx/5xx 状态码（如 106），需显式调用 onStatus 将其转换为 WebClientResponseException，否则 retryWhen 不会触发。
• Mono.delay() 返回的是 Mono<Long>，仅作计时用途，其发射值不影响后续流程；真正的重试由 Reactor 内部调度器驱动。
• 动态延迟不参与 jitter 计算（因 jitter() 是 RetryBackoffSpec 特有 API），如需抖动，需手动在 parseRetryTimeFromBody 或 getBackOffDelayFromHeaders 中添加随机偏移（例如 delay.plusMillis(ThreadLocalRandom.current().nextLong(0, delay.toMillis() / 10))）。

通过该方案，客户端重试行为完全与服务端治理策略对齐，既避免盲目轮询加重服务压力，又保障了请求最终可达性，是构建高韧性响应式微服务的关键实践之一。