Spring WebFlux(反应器).zipWith 时出错 - 由于缺少请求而无法发出滴答声 [英] Spring WebFlux (reactor). Error when zipWith - Could not emit tick due to lack of requests

问题描述

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

I have a Flux, for each object I should make an API call to the third party REST (about 1000 calls). To prevent to many requests per second Im using:

    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

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

The problem is, sometimes application fails with exception:

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)

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

Is there any logic that I can add to prevent an error, or just skip this tick?

推荐答案

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

This means that the consumer of the results doesn't consume the data fast enough: the interval being on a fixed frequency, it tries to emits but nobody listens.

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

There's a need for some sort of more advanced permit-based rate limiter built on Reactor I think. But in the meantime, another simple (simplistic?) approach that you could try is to individually ensure each call is delayed from the previous one by 10ms:

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

(此操作符是为了替换 zipWith(interval) 模式)

(this operator was made to replace the zipWith(interval) pattern)

这篇关于Spring WebFlux(反应器).zipWith 时出错 - 由于缺少请求而无法发出滴答声的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！