我正在尝试找出是否有任何方法可以在消息回滚到队列时影响消费者的消息处理顺序。以下是一些简单的代码,可帮助我复制该问题。我只是按特定顺序将具有不同JMSXGroupIds的消息推送到队列中:
- "A1" (JMSXGroupId: 1) - "B1" (JMSXGroupId: 2) - "A2" (JMSXGroupId: 1) - "C1" (JMSXGroupId: 3) - "B2" (JMSXGroupId: 2)
代码使A1回滚(它最初重试了3次消息),并以延迟的方式返回队列。然而,消费者会等待直到可以再次获取A1(等待延迟的时间后),这意味着B1和C1组被阻塞在A1后面,无法被处理。
理想情况下,我希望当A1放回队列并告知其等待时,消费者会选择B1和C1... 我最终要做的就是防止一个JMSXGroup阻塞其他JMSXGroup。此外,值得注意的是,我需要保留A的消息顺序(A1、A2、A3...),并希望通过将它们保留在队列上来实现,而不是构建某些异常管理解决方案。
我正在使用基于Java的Apache Camel微服务和事务路由。没有什么特别的,但如果需要,我可以提供更多详细配置信息。
提前感谢。
- "A1" (JMSXGroupId: 1) - "B1" (JMSXGroupId: 2) - "A2" (JMSXGroupId: 1) - "C1" (JMSXGroupId: 3) - "B2" (JMSXGroupId: 2)
代码使A1回滚(它最初重试了3次消息),并以延迟的方式返回队列。然而,消费者会等待直到可以再次获取A1(等待延迟的时间后),这意味着B1和C1组被阻塞在A1后面,无法被处理。
理想情况下,我希望当A1放回队列并告知其等待时,消费者会选择B1和C1... 我最终要做的就是防止一个JMSXGroup阻塞其他JMSXGroup。此外,值得注意的是,我需要保留A的消息顺序(A1、A2、A3...),并希望通过将它们保留在队列上来实现,而不是构建某些异常管理解决方案。
onException(Exception.class)
.log("Exception Caught !! ")
.redeliveryDelay("1000")
.maximumRedeliveries(3)
.handled(false)
.markRollbackOnly()
.log("log:output");
from("amq:queue:mailbox?concurrentConsumers=1")
.to(logEndpoint)
.process(exchange -> {
if(exchange.getIn().getBody(String.class).contains("A")) {
throw new Exception("Found A");
}
});
我正在使用基于Java的Apache Camel微服务和事务路由。没有什么特别的,但如果需要,我可以提供更多详细配置信息。
提前感谢。