我一直在学习Spring Webflux和响应式编程，并且遇到了一个问题，我试图使用Spring Webclient解决重试逻辑。我已经创建了一个客户端，并成功调用了一个返回一些JSON数据的外部Web服务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 Builder

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

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

型

安装时间

我试图理解并使用Retry类的retryWhen方法，但不知道是否可以访问或传递响应头值。

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），直到先前建立的Retry-After值过期。

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))));
        })