project-reactor - Spring WebFlux( react 器)。 zipWith 时出错 - 由于缺少请求而无法发出滴答声

Question

我有一个 Flux，对于每个对象，我应该对第三方 REST 进行 API 调用(大约 1000 次调用)。为了防止每秒有很多请求，我使用:

    Flux<Calls> callsIntervalFlux=
            Flux.interval(Duration.ofMillis(100))
                    .zipWith(callsFlux, (i, call) -> call);

// and now Calls emits every 10ms, and REST API is not overloaded

问题是，有时应用程序会因异常而失败:

reactor.core.Exceptions$ErrorCallbackNotImplemented: reactor.core.Exceptions$OverflowException: Could not emit tick 32 due to lack of requests (interval doesn't support small downstream requests that replenish slower than the ticks)
Caused by: reactor.core.Exceptions$OverflowException: Could not emit tick 32 due to lack of requests (interval doesn't support small downstream requests that replenish slower than the ticks)

是否可以添加任何逻辑来防止错误，或者只是跳过此勾选？

Answer 1

这意味着结果的消费者没有足够快地消费数据:interval 处于固定频率，它尝试发出但没有人听。

我认为需要某种基于 Reactor 的更高级的基于许可的速率限制器。但与此同时，您可以尝试的另一种简单(简单？)方法是单独确保每个调用都比前一个调用延迟 10 毫秒:

Flux<Calls> callsIntervalFlux = callsFlux.delayElements(Duration.ofMillis(10));

(这个运算符是为了替换 zipWith(interval) 模式)