Apache Camel - Seda 端点、recipientList、BlockWhenFull=true 和 Queue full IllegalStateException答案

【问题标题】：Apache Camel - Seda endpoint, recipientList, BlockWhenFull=true and Queue full IllegalStateExceptionApache Camel - Seda 端点、recipientList、BlockWhenFull=true 和 Queue full IllegalStateException
【发布时间】：2023-01-19 17:55:37
【问题描述】：

我有一个代码可以从 RabbitMQ 队列中检索消息，聚合它们，然后将聚合分发到另一个路由，该路由将通过 recipientList 组件将它们分派到不同的路由。问题在于，尽管在生产者端添加了 blockWhenFull=true 属性，但后者仍会抛出“由于队列已满而导致执行反应性工作时出错”异常。

from("direct:rabbitmq-ids-aggregate")
        .aggregate(constant(true), new UpdatesAggregationStrategy())
            .completionInterval("{{updates.aggregation.completionInterval}}")
            .completionSize("{{updates.aggregation.completionSize}}")
            .setHeader(CORRELATION_HEADER, simple("${exchangeId}"))
        .to("seda:dispatch?blockWhenFull=true");

from("seda:dispatch")
        .recipientList(simple("{{routes.hr-data}},{{routes.comments-data}},{{routes.legacy-hr-data}},{{routes.ranking-data}}"))
        .end();

2023-01-10 19:46:35,944 WARN o.a.c.i.e.DefaultReactiveExecutor [Camel (integration-core) thread #5 - seda://dispatch] Error executing reactive work due to Queue full. This exception is ignored.
java.lang.IllegalStateException: Queue full
    at java.util.AbstractQueue.add(AbstractQueue.java:98) ~[?:?]
    at org.apache.camel.component.seda.SedaProducer.addToQueue(SedaProducer.java:251) ~[camel-seda-3.15.0.jar:3.15.0]
    at org.apache.camel.component.seda.SedaProducer.process(SedaProducer.java:149) ~[camel-seda-3.15.0.jar:3.15.0]
    at org.apache.camel.processor.errorhandler.RedeliveryErrorHandler$SimpleTask.run(RedeliveryErrorHandler.java:471) ~[camel-core-processor-3.15.0.jar:3.15.0]
    at org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:187) [camel-base-engine-3.15.0.jar:3.15.0]
    at org.apache.camel.impl.engine.DefaultReactiveExecutor.scheduleMain(DefaultReactiveExecutor.java:64) [camel-base-engine-3.15.0.jar:3.15.0]
    at org.apache.camel.processor.Pipeline.process(Pipeline.java:184) [camel-core-processor-3.15.0.jar:3.15.0]
    at org.apache.camel.impl.engine.CamelInternalProcessor.process(CamelInternalProcessor.java:399) [camel-base-engine-3.15.0.jar:3.15.0]
    at org.apache.camel.component.seda.SedaConsumer.sendToConsumers(SedaConsumer.java:269) [camel-seda-3.15.0.jar:3.15.0]
    at org.apache.camel.component.seda.SedaConsumer.doRun(SedaConsumer.java:187) [camel-seda-3.15.0.jar:3.15.0]
    at org.apache.camel.component.seda.SedaConsumer.run(SedaConsumer.java:130) [camel-seda-3.15.0.jar:3.15.0]
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
    at java.lang.Thread.run(Thread.java:833) [?:?]

【问题讨论】：

标签： java apache-camel

【解决方案1】：

这blockWhenFull=真生产者端的属性不起作用，因为它被设置在错误的端点上（直接：rabbitmq-ids-聚合）而不是 seda:dispatch。要解决此问题，您应该添加blockWhenFull=真上的财产塞达：派遣像这样的端点：

from("seda:dispatch?blockWhenFull=true")
.recipientList(simple("{{routes.hr-data}},{{routes.comments-data}},
{{routes.legacy-hr-data}},{{routes.ranking-data}}")).end();

这将告诉 Camel 在向其添加新消息之前阻塞并等待 seda:dispatch 队列有可用空间，这应该可以防止抛出“由于队列已满而导致执行反应性工作时出错”异常。

【讨论】：