Reactor 中的自定义重试

Question

我试图基于 Reactor 额外包的功能在 Kotlin 和 Reactor 中实现重试逻辑。我想要做的是传递一个持续时间列表，并且在每个context.iteration 上，我得到列表的第 (iteration-1) 个元素。它部分工作，我总是在最后一次迭代中得到一个IndexOutOfBoundsException，这比我想要的要多，尽管我提供了最大重试次数 - 列表的大小。虽然重试在给定的持续时间和“正确”的次数内运行（肯定是因为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，您会发现有一个特殊的BackoffDelay，名为RETRY_EXHAUSTED。它在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);
    }
}