Reactor 中的自定义重试

Question

我试图根据 Reactor extra 包的功能在 Kotlin 和 Reactor 中实现重试逻辑。我想要做的是传递一个持续时间列表，并且在每个 context.iteration 上我得到列表的第（iteration-1）个元素。它工作部分，我总是在最后一次迭代中得到 IndexOutOfBoundsException，这比我想要的要多，尽管我已经提供了最大重试次数 - 大小的名单。但是，重试是运行，在给定的持续时间内和 "correct" 次（肯定是因为 IndexOutOfBoundsException 阻止了更多），只有这个异常（这是根本原因）困扰着我。

这是我自定义的 BackOff 界面：

interface MyCustomBackoff : Backoff {
    companion object {
        fun getBackoffDelay(backoffList: List<Duration>): (IterationContext<*>) -> BackoffDelay {
            return { context -> BackoffDelay(backoffList[(context.iteration() - 1).toInt()]) }
        }
    }
}

我的 Kotlin 扩展是：

fun <T> Mono<T>.retryCustomBackoffs(backoffList: List<Duration>, doOnRetry: ((RetryContext<T>) -> Unit)? = null): Mono<T> {
    val retry = Retry.any<T>().retryMax(backoffList.size.toLong()).backoff(MyCustomBackoff.getBackoffDelay(backoffList))

    return if (doOnRetry == null) {
        this.retryWhen(retry)
    }
    else {
        this.retryWhen(retry.doOnRetry(doOnRetry))
    }
}

我在这里错过了什么？

Answer 1

如果你查看 reactor.retry.AbstractRetry#calculateBackoff，你会发现有一个名为 RETRY_EXHAUSTED 的特殊 BackoffDelay。它 returns 当 retryContext.iteration() > maxIterations （不是 >=）在 backoff.apply(retryContext)

之后

if (retryContext.iteration() > maxIterations || Instant.now(clock).plus(jitteredBackoff).isAfter(timeoutInstant))
    return RETRY_EXHAUSTED;

因此，如果列表中有 2 个自定义退避延迟，则 calculateBackoff 会产生 3 个退避延迟。

您可以像这样更改您的 MyCustomBackoff（对不起 Java，我对 Kotlin 不熟悉）：

public interface MyCustomBackoff extends Backoff {
    static Backoff getBackoffDelay(List<Duration> backoffList) {
        return context -> context.iteration() <= backoffList.size() ?
                new BackoffDelay(backoffList.get(Long.valueOf(context.iteration() - 1).intValue())) :
                new BackoffDelay(Duration.ZERO);
    }
}

Reactor 中的自定义重试

Custom retry in Reactor

kotlin

project-reactor

retry-logic

spring-webflux