如何处理JMS重新传送到我的Camel Road，但仍然允许消息删除

Question

我有一个骆驼路由，它从ActiveMQ JMS队列中读取，进行一些处理，并将结果传递到远程目标。如果此过程的通信部分失败，我想无限期地重试（直到目标“启动”）。我可以通过使路由事务处理并设置重新交付策略来处理这个问题。重新交付在骆驼（通过camel.processor.重新尝试路由的失败部分）和ActiveMQConntionWorks（通过org.apache.activemq.重新尝试整个路由）中设置。

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

问题是，我可以允许消息删除(通过将使用者设置为< code > CACHE level name = CACHE _ NONE )，也可以让连接处理重试(通过将使用者设置为< code > CACHE level name = CACHE _ CONSUMER )，但不能同时允许这两者。

这是我目前的设置：

xml prettyprint-override"><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读取，并且在通信异常的情况下，Camel在内部重试路由从到部分两次（redeliveryProfile）。然后在 5 秒延迟后，消息将再次通过整个路由发送 （amqRedeliveryPolicy）。这一直持续到我停止测试。

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

如果我将消费者更改为：

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

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

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

还是我完全错过了这一点？

Answer 1

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

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

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

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