如何解决使用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()
调用期间使用的可覆盖方法)来识别错误。
如果冷热通量的行为相同,我将调查
另一个需要说明的是,是否有任何架构组件在起作用(例如,订户/发布者是否直接连接,或者它们是否经过各种负载均衡器和代理),因为这可以帮助隔离错误。
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。