Camel 和 JMS 以正确的顺序从高级队列中消费消息

Posted 2023-03-10

【中文标题】Camel 和 JMS 以正确的顺序从高级队列中消费消息

【英文标题】：Consuming message in the correct order from an Advanced Queue by Camel and JMS

【发布时间】：2015-12-13 05:01:24

【问题描述】：

我在将 Apache Camel 与 Oracle 高级队列和 JMS 结合使用时遇到了问题。

这是关于分发消息的应用程序。在 Camel 的帮助下，消息在 Oracle 高级队列中被接收和排队。然后它们被 Camel 消费并转发到目标系统。 对于消息投递失败的情况，在Advanced Queue中定义了一个重试次数，以便重复投递消息。

如果 Camel 现在将消息出列并将其发送到不可用的目标系统，则会抛出 HttpOperationFailedException 或 NoSuchEndpointException。这些被捕获并执行回滚。

此时，期望消息传递将按照重试计数中定义的频率重试，然后移至异常队列。 但是，正在发生的是队列中的下一条消息正在发送。

因为消息的内容是部分相互依赖的，所以必须按顺序处理。

我认为 JMS 库的使用存在错误配置，但我不确定并且没有发现任何可以影响此行为的东西。

使用的 JMS 库是 Oracle AqApi v 11.2.0.3。

这是骆驼路线的代码：

from("jms-camel-component:myComponent.AQ?jmsMessageType=Text").routeId("deliveryToTarget")
        .transacted()                
        .setExchangePattern(ExchangePattern.InOut)                        
        .setHeader(Exchange.HTTP_QUERY, constant("throwExceptionOnFailure=false"))
        .setHeader(Exchange.CONTENT_TYPE, constant("application/xml; charset=UTF-8"))
        .doTry()
            .recipientList(header("endpointTarget"))
            .endDoTry()
            .process(ResponseProzessor.getInstance())
            .log("Message was delivererd.")
        .doCatch(HttpOperationFailedException.class, NoSuchEndpointException.class)    
            .process(ResponseProzessor.getInstance())
            .log("Error occured.")
            .rollback()
        .end();

JmsComponent 配置如下：

JmsComponent jmsComponent = new JmsComponent(scc);
jmsComponent.setConnectionFactory(connectionFactory);
jmsComponent.setTransactionManager(tm);
jmsComponent.setMaxConcurrentConsumers(1);            
jmsComponent.setMaxMessagesPerTask(1);                  
jmsComponent.setIncludeSentJMSMessageID(true);

提前感谢您的帮助！

更新

我想，我已经找到了所描述行为的原因。高级队列上配置了延迟。只要延迟持续，队列中的下一条消息就会出队。消息不是随机出列的，它们是根据优先级出列的。

我真的认为这是必须在消费者身上配置的东西。只要没有提交或移动到异常队列，是否有任何技巧可以配置 camel-jms-component 以使用队列中的第一条消息？ 没找到直接在camel上配置的选项……

【问题讨论】：

在 JMS 中，一般不保证消息的顺序。在 ActiveMQ 中，如果你只有一个消费者，并且客户端重新交付，你是相当安全的，但我不知道 Oracle AQ...

如果没有针对此问题的通用解决方案，我会对 Weblogic 应用程序服务器的解决方案感兴趣。

解决这个问题的（干净）解决方案始终是服务器端，因为客户端（应该/）不知道队列中的消息。根据您的 QoS，您需要读取队列中的所有消息以确定您首先想要的消息 - 根据您的选项（配置明智），哪个仍然可能是一个选项。在您在客户端实施此操作之前，我会在下面检查我的答案/服务器端的配置。 HTH

事后看来，问题似乎是在高级队列数据库端设置的延迟。这导致 Camel 框架在发生错误的情况下完全回滚之前的事务之前使用下一条消息。很高兴知道在这种情况下这是否是正确的事情或 Oracle AQ 实施的疏忽。

@PhilW：这是否意味着我必须在 Camel 和数据库之间放置一个像 Apache ActiveMQ 这样的消息代理？据我了解，这些消息代理的用途与 Oracle 高级队列相同。这不过是另一个队列。那是对的吗？那么没有一个已经过时了吗？

【参考方案1】：

我不是 Oracle AQ 专家，但据我所知，这是队列中的设置，而不是客户端。

sort_list 参数确定消息的顺序 出队。拥有后无法更改邮件排序顺序 创建了队列表

来自： http://docs.oracle.com/cd/B19306_01/server.102/b14257/aq_admin.htm

您很可能设置了 ENQ_TIME 或 COMMIT_TIME - 这可能已经满足您的需求。

当然，您的消费者必须是唯一的。

【讨论】：

根据我使用 AQ 的经验，Java 通常是简单的样板，而实际配置恰好在数据库中排队。可能还想查看相关消息的消息分组docs.oracle.com/cd/B10501_01/appdev.920/a96587/…

你是对的。真正的配置发生在数据库中。可以为队列配置延迟，但似乎消息只会在延迟时间内被阻塞，这意味着 jms 开始消费下一条消息。感谢您的链接，但消息分组似乎不是一个解决方案，因为消息不是在同一个事务中发送的。

感谢您帮助我们解决问题。我们最初的问题是 AQ 中设置的超时，这导致队列消息的无序处理。尽管您的小费并不是我们解决方案的重要原因，但我认为奖励您赏金是明智的。谢谢！