Project Reactor - 如何处理来自 Flux.interval 的溢出异常？

Question

我正在使用 Spring Webflux 构建一个 spring boot 应用程序，我想让应用程序完全非阻塞。应用程序本身有一些 REST 端点和需要每隔几秒运行一次的批处理作业。对于批处理作业，我试图Flux.interval(Duration.ofMillis(1000))生成我忽略的长值并运行我的预定作业。

Flux.interval(Duration.ofMillis(1000))
    .flatMap(ignore -> doSomething())
    .subscribe();

但是一段时间后我收到错误

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

有人能告诉我如何克服这个问题吗？

Answer 1

问题的原因很可能是doSomething()操作花费的时间比指定的 Flux 间隔长，这意味着一段时间后doSomething作业相互重叠并产生背压。因为Flux.interval是一个热源（意味着它不会按需发出信号）并且flatMap有一个默认的并发限制 (256)，操作员会不堪重负，这会导致OverflowException.

根据您的要求，此问题有几个潜在的解决方案：

1. 忽略溢出错误，丢弃会溢出的信号

这意味着有时，如果我们已经有很多（256）在进行中，我们会跳过一秒钟并且不会在间隔内安排作业。

Flux.interval(Duration.ofMillis(1000))
    .onBackpressureDrop()
    .flatMap(ignore -> doSomething())

2.将flatMap并发设置为更高的值

这仍然会在一段时间后导致溢出异常，但它会延迟问题的出现（可能不是最佳解决方案）。

Flux.interval(Duration.ofMillis(1000))
    .flatMap(ignore -> doSomething(), Integer.MAX_VALUE)

3. 不要让工作相互重叠

我们从热源切换到冷源，从而消除了溢出的可能性。然而，我们失去了每秒安排一个事件的保证。相反，它们将在上一个作业完成且至少经过 1 秒后按需安排。

Mono.just(1).repeat() // infinite Flux with backpressure
    .delayElements(Duration.ofMillis(1000))
    .concatMap(ignore -> doSomething())

如果您可以处理重叠作业并在flatMap调用中定义合理的并发级别，您还可以将此解决方案与前一个解决方案结合使用。