如何处理 JMS 重新传递到我的骆驼路线，但仍允许删除消息

Question

我有一个 Camel Route，它从 ActiveMQ JMS 队列中读取数据，进行一些处理，然后将结果传递到远程目的地。如果此过程的通信部分失败，我想无限期地重试（直到目的地“启动”）。我可以通过路由transacted 并设置redeliveryPolicy 来处理这个问题。在 Camel（通过 camel.processor.RedeliveryPolicy 重新尝试路由的失败部分）和 ActiveMQConnectionFactory（通过 org.apache.activemq.RedeliveryPolicy 重新尝试整个路由）中都设置了重新传递。

我还要求我们应该能够从 JMS 队列中删除一个条目（我通过一个通过 JMX 与 ActiveMQ 对话的应用程序来执行此操作），并且处理应该移至下一条消息。

问题是我可以允许删除消息（通过将使用者设置为cacheLevelName=CACHE_NONE），或者让连接句柄重试（通过将使用者设置为cacheLevelName=CACHE_CONSUMER），但不能同时允许。

这是我目前的设置：

<bean id="jmsTransactionManager" class="org.springframework.jms.connection.JmsTransactionManager">
  <property name="connectionFactory" ref="jmsConnectionFactory"/>
</bean>

<bean id="jmsConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
  <property name="brokerURL" value="${jms.brokerUrl}" />
  <property name="redeliveryPolicy" ref="amqRedeliveryPolicy" />
  <property name="prefetchPolicy">
    <bean class="org.apache.activemq.ActiveMQPrefetchPolicy">
      <property name="all" value="0"/>
    </bean>
  </property>
</bean> 

<bean id="jmsConfig" class="org.apache.camel.component.jms.JmsConfiguration">
  <property name="connectionFactory" ref="jmsConnectionFactory"/>
  <property name="transactionManager" ref="jmsTransactionManager"/>
  <property name="transacted" value="true"/>
  <property name="concurrentConsumers" value="1"/>
</bean>

<bean id="jms" class="org.apache.camel.component.jms.JmsComponent" >
  <property name="configuration" ref="jmsConfig" />
</bean>

<!-- Redelivery Policy for ActiveMQ (the Broker), so it will perform retries of the entire route -->
<bean id="amqRedeliveryPolicy" class="org.apache.activemq.RedeliveryPolicy">
  <property name="initialRedeliveryDelay" value="5000" />
  <property name="redeliveryDelay" value="5000" />
  <property name="maximumRedeliveries" value="-1" />
  <property name="queue" value=">" />
</bean>

<!-- Redelivery Policy for Camel Internal, so it will retry from the part of the exchange that's failed in case of comms issues -->
<bean id="redeliveryProfile" class="org.apache.camel.processor.RedeliveryPolicy">
  <property name="maximumRedeliveries" value="2"/>
  <property name="redeliveryDelay" value="500"/>
</bean>

<bean id="PROPAGATION_REQUIRED" class="org.apache.camel.spring.spi.SpringTransactionPolicy">
  <constructor-arg>
    <bean class="org.springframework.transaction.support.TransactionTemplate">
      <property name="transactionManager" ref="jmsTransactionManager"/>
    </bean>
  </constructor-arg>
</bean>

<camel:camelContext id="testContext">
  <camel:routeBuilder ref="transactedRoute" />
</camel:camelContext>

用一个非常简单的路线：

@Component
public class TransactedRoute extends SpringRouteBuilder {
  @Override
  public void configure() {
    onException(Exception.class) // we would handle comms exceptions here
      .redeliveryPolicyRef("redeliveryProfile")
      .rollback();

    from("jms:queue:myqueue?cacheLevelName=CACHE_CONSUMER")
      .transacted("PROPAGATION_REQUIRED")
      .bean(Bean1.class)
      .to(DESTINATION); // this could throw a comms exception
  }
}

使用此设置，路由从 jms:queue:myqueue 读取，并且在 Comms 异常的情况下，Camel 在内部重试来自 to 部分的路由两次 (redeliveryProfile)。然后在 5 秒延迟后，消息再次通过整个路由发送 (amqRedeliveryPolicy)。这种情况一直持续到我停止测试为止。

但是，如果我从 ActiveMQ 中删除消息，它将继续由路由处理，尽管它不再在队列中。

如果我将消费者更改为：

from("jms:queue:myqueue?cacheLevelName=CACHE_NONE")

我现在可以从 ActiveMQ 中删除消息并且路由停止处理它...但是 amqRedeliveryPolicy 被忽略，消息会立即重试（无 5 秒延迟）并在 6 次尝试后（AMQ 的默认值）它被放入死信队列。

那么，有没有办法通过修改我的配置来实现这两者？

还是我完全错过了沿线的某个地方？

Answer 1

如果您需要人工干预以通过 JMX 删除队列中的特定消息，这听起来有点像一个糟糕的设计。

也许你可以让 AMQ 在尝试 X 次失败后将消息移动到 DLQ - 永远重新传递也是一个糟糕的设计。但是如果 6 次尝试很少，您可以将默认值增加到更高的值。

然后，您可以从 DLQ 队列中检查消息并尝试了解消息失败的原因，并且您始终可以通过 JMX 或其他工具安全地删除或清除 DLQ 队列。

AMQ 允许每个队列有一个 DLQ，您可以将其配置为使用 DLQ 前缀等，而不是一个常见的 DLQ 队列。