1

我有一个 spring 应用程序，它构建了一个响应式管道，如下所示：

buildPipeline(). // returns a flux based on changeStreamEvents or Kafka receives
.bufferTimeout( capacity, Duration.ofSeconds(1))
. flatMap( r -> {
   element x = r.get(r.size()-1)
   //some processing on element and the batch obtained
})
.doOnError( e-> log.info("error occurred:" + e.toString())
.subscribe()

但是，我看到我的应用程序中间抛出以下错误 -

java.lang.illegalArgumentException:3.9 While the Subscription is not cancelled, Subscription.request(long n) MUST throw a java.lang.illegalArgumentException if argument <= 0
at com.mongodb.reactivestreams.client.internal.ObservableToPublisher$1
$1.request(ObservableToPublisher.java:43)
at reactor.core.publisher.FluxMap$MapSubscriber.request(FluxMap.java:155)
at reactor.core.publisher.FluxBufferTimeout
$BufferimeoutSubscriber.requestMore(FluxBufferTimeout.java:317)

我无法确定出了什么问题，以及为什么流以这个错误终止。任何帮助将不胜感激。

在我添加“bufferTimeout”以添加批处理功能后，应用程序开始抛出此错误。在此之前，我从未遇到过这个异常。也不确定如何复制该问题，因为它不是在本地或 UAT 中发生，而是仅在应用程序的生产环境中发生。任何线索都会有所帮助。

谢谢！

4

1 回答 1

1

尝试添加一个 onBackPressureBuffer()，以便在需求低的情况下，此操作符缓冲请求，并以受控方式发出项目。

于 2020-12-12T11:02:00.440 回答