Spring WebClient - 如何根据响应标头延迟重试

问题描述

一点背景

我一直在学习 Spring Webflux 和响应式编程，但遇到了一个问题，我试图使用 Spring Webclient 解决重试逻辑。我已经创建了一个客户端并成功调用了一个返回一些 JSON 数据的外部网络服务 GET 端点。

问题

当外部服务以 503 - Service Unavailable 状态响应时，响应包含一个 Retry-After 标头，其中包含一个值，该值指示在重试请求之前我应该​​等待多长时间。我想在 Spring Webflux/Reactor 中找到一种方法来告诉 webClient 在 X 周期后重试它的请求，其中 X 是现在与我从响应标头中解析出的 DateTime 之间的差异。

简单的 WebClient GET 请求

public <T> Mono<T> get(final String url,Class<T> clazz) {
        return webClient
                .get().uri(url)
                .retrieve()
                .bodyToMono(clazz);
    }

WebClient 生成器

我使用构建器创建了上述方法中使用的 webClient 变量，并将其作为实例变量存储在类中。

webClientBuilder = WebClient.builder();
webClientBuilder.codecs(clientCodecConfigurer -> {
    clientCodecConfigurer.defaultCodecs();
    clientCodecConfigurer.customCodecs().register(new Jackson2JsonDecoder());
    clientCodecConfigurer.customCodecs().register(new Jackson2JsonEncoder());
});
webClient = webClientBuilder.build();

重试时间

我已尝试了解并使用 retryWhen 类的 Retry 方法，但不知道我是否可以访问或传递那里的响应标头值。

public <T> Mono<T> get(final String url,Class<T> clazz) {
        return webClient
                .get().uri(url)
                .retrieve()
                .bodyToMono(clazz);
                .retryWhen(new Retry() {
                    @Override
                    public Publisher<?> generateCompanion(final Flux<RetrySignal> retrySignals) {
                        // Can I use retrySignals or retryContext to find the response header somehow?
                        // If I can find the response header,how to return a "yes-retry" response?
                    }
                })
    }

具有额外逻辑和数据库交互的过滤器

我还尝试做一些额外的逻辑并在 WebClient.Builder 中使用过滤器，但这只会让我停止新请求（调用 #get），直到之前建立的重试-价值结束后。

webClientBuilder = WebClient.builder();
webClientBuilder.codecs(clientCodecConfigurer -> {
    clientCodecConfigurer.defaultCodecs();
    clientCodecConfigurer.customCodecs().register(new Jackson2JsonDecoder());
    clientCodecConfigurer.customCodecs().register(new Jackson2JsonEncoder());
});
webClientBuilder.filter(ExchangeFilterFunction.ofRequestProcessor(clientRequest -> {
            final Clock clock = Clock.systemUTC();
            final int id = (int) clientRequest.attribute("id"); // id is saved as an attribute for the request,pull it out here
            final long retryAfterEpochMillis = // get epoch millisecond from DB for id
            if(epoch is in the past) {
                return Mono.just(clientRequest);
            } else { // have to wait until epoch passes to send request
                return Mono.just(clientRequest).delayElement(Duration.between(clock.instant(),Instant.ofEpochMilli(retryAfterEpochMillis)));
            }
        })
);
webClient = webClientBuilder.build();

.onStatus(HttpStatus::isError,response -> {
                    final List<String> retryAfterHeaders = response.headers().header("Retry-After");
                    if(retryAfterHeaders.size() > 0) {
                        final long retryAfterEpochMillis = // parse millisecond epoch time from header
                        // Save millisecond time to DB associated to specific id
                    }
                    return response.bodyToMono(String.class).flatMap(body ->
                            Mono.error(new RuntimeException(
                                            String.format("Request url {%s} failed with status {%s} and reason {%s}",url,response.rawStatusCode(),body))));
                })

感谢任何帮助，如果我能提供更多上下文数据来提供帮助，我会的。

解决方法

1.在重试构建器中检索标头

public class WebClientStatefulRetry3 {
    public static void main(String[] args) {
        WebClient webClient = WebClient.create();

        call(webClient)
                .retryWhen(Retry.indefinitely()
                        .filter(ex -> ex instanceof WebClientResponseException.ServiceUnavailable)
                        .doBeforeRetryAsync(signal -> Mono.delay(calculateDelay(signal.failure())).then()))
                .block();
    }

    private static Mono<String> call(WebClient webClient) {
        return webClient.get()
                .uri("http://mockbin.org/bin/b2a26614-0219-4018-9446-c03bc1868ebf")
                .retrieve()
                .bodyToMono(String.class);
    }

    private static Duration calculateDelay(Throwable failure) {
        String headerValue = ((WebClientResponseException.ServiceUnavailable) failure).getHeaders().get("Retry-After").get(0);

        return // calculate delay here from header and current time;
    }
}

2.使用扩展运算符访问前一个响应并生成下一个

public class WebClientRetryWithExpand {
    public static void main(String[] args) {
        WebClient webClient = WebClient.create();

        call(webClient)
                .expand(prevResponse -> {
                    List<String> header = prevResponse.headers.header("Retry-After");
                    if (header.isEmpty()) {
                        return Mono.empty();
                    }
                    long delayInMillis = // calculate delay from header and current time

                    return Mono.delay(Duration.ofMillis(delayInMillis))
                            .then(call(webClient));
                })
                .last()
                .block();
    }

    private static Mono<ResponseWithHeaders> call(WebClient webClient) {
        return webClient.get()
                .uri("https://example.com")
                .exchangeToMono(response -> response.bodyToMono(String.class)
                        .map(rawResponse -> new ResponseWithHeaders(rawResponse,response.headers())));
    }

    @Data
    static class ResponseWithHeaders {
        private final String rawResponse;
        private final ClientResponse.Headers headers;
    }
}