使用RabbitMq MessageListenerContainer的反应堆通量超时

问题描述

使用热发布者模型，在云环境中大约有50％的时间发生以下超时异常：

print()

但是我不能在本地复制它。我的主要理论是，在通量产生期间发生原因，并且在[ERROR] reactor.core.scheduler.Schedulers - Scheduler worker in group main failed with an uncaught exception reactor.core.Exceptions$ErrorCallbackNotImplemented: java.util.concurrent.TimeoutException: Did not observe any item or terminal signal within 300000ms (and no fallback has been configured) Caused by: java.util.concurrent.TimeoutException: Did not observe any item or terminal signal within 300000ms (and no fallback has been configured) reactor.core.publisher.FluxTimeout$TimeoutMainSubscriber.handleTimeout(FluxTimeout.java:289) reactor.core.publisher.FluxTimeout$TimeoutMainSubscriber.doTimeout(FluxTimeout.java:274) reactor.core.publisher.FluxTimeout$TimeoutTimeoutSubscriber.onNext(FluxTimeout.java:396) reactor.core.publisher.StrictSubscriber.onNext(StrictSubscriber.java:89) datadog.trace.instrumentation.reactor.core.TracingSubscriber.onNext(TracingSubscriber.java:75) reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:73) datadog.trace.instrumentation.reactor.core.TracingSubscriber.onNext(TracingSubscriber.java:75) reactor.core.publisher.MonoDelay$MonoDelayRunnable.run(MonoDelay.java:117) reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:68) reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:28) java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304) java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) java.base/java.lang.Thread.run(Thread.java:834) 和Flux之间发生了酸味。创作看起来像：

MessageListenerContainer

这个异常感觉像是一个配置错误，但是我无法缩小导致此超时的原因，以确认这一点。任何帮助将不胜感激！

解决方法

引用：https://www.codota.com/code/java/methods/reactor.core.publisher.Flux/timeout 我怀疑问题在于创建masterFlux时没有指定超时。

引用：https://projectreactor.io/docs/core/release/api/reactor/core/publisher/Flux.html

create（）的静态方法具有以下定义： create（Consumer 发射器，FluxSink.OverflowStrategy背压）

因此，我正在猜测，因为尚未定义OverflowStrategy，如果消息传输失败，则代码不知道该怎么办并引发此异常。有关OverflowStrategy的更多信息，请参见： https://projectreactor.io/docs/core/release/api/reactor/core/publisher/FluxSink.OverflowStrategy.html

,

如果热通量和冷通量的行为不同，则可能是Flux默认配置中的某些内容（可能是您在create()调用期间使用的可覆盖方法）来识别错误。

如果冷热通量的行为相同，我将调查

另一个需要说明的是，是否有任何架构组件在起作用（例如，订户/发布者是否直接连接，或者它们是否经过各种负载均衡器和代理），因为这可以帮助隔离错误。