spring - 有条件地具有指数退避的弹簧反应重试

Question

使用spring响应式WebClient，我使用一个API，如果响应状态为500，我需要使用指数退避重试。但是在 Mono 类中，我没有看到任何以 Predicate 作为输入参数的 retryBackoff。

这是我搜索的那种功能：

public final Mono<T> retryBackoff(Predicate<? super Throwable> retryMatcher, long numRetries, Duration firstBackoff)

现在我的实现如下（我没有使用 backOff 机制重试）：

client.sendRequest()
    .retry(e -> ((RestClientException) e).getStatus() == 500)
    .subscribe();

Answer 1

您可能想查看reactor-addons项目中的reactor-extra模块。在 Maven 中，您可以执行以下操作：

<dependency>
    <groupId>io.projectreactor.addons</groupId>
    <artifactId>reactor-extra</artifactId>
    <version>3.2.3.RELEASE</version>
</dependency>

然后像这样使用它：

client.post()
    .syncBody("test")
    .retrieve()
    .bodyToMono(String.class)
    .retryWhen(Retry.onlyIf(ctx -> ctx.exception() instanceof RestClientException)
                    .exponentialBackoff(firstBackoff, maxBackoff)
                    .retryMax(maxRetries))

Answer 2

我目前正在使用 Kotlin Coroutines + Spring WebFlux 进行尝试：

以下似乎不起作用：

suspend fun ClientResponse.asResponse(): ServerResponse =
    status(statusCode())
        .headers { headerConsumer -> headerConsumer.addAll(headers().asHttpHeaders()) }
        .body(bodyToMono(DataBuffer::class.java), DataBuffer::class.java)
        .retryWhen { 
            Retry.onlyIf { ctx: RetryContext<Throwable> -> (ctx.exception() as? WebClientResponseException)?.statusCode in retryableErrorCodes }
                .exponentialBackoff(ofSeconds(1), ofSeconds(5))
                .retryMax(3)
                .doOnRetry { log.error("Retry for {}", it.exception()) }
        )
        .awaitSingle()

Answer 3

我不确定，你使用的是什么 spring 版本，在 2.1.4 我有这个：

client.post()
    .syncBody("test")
    .retrieve()
    .bodyToMono(String.class)
    .retryBackoff(numretries, firstBackoff, maxBackoff, jitterFactor);

......所以这正是你想要的，对吧？