JMS 回滚和重新交付不遵守 RedeliveryDelay 配置

Question

我想让我的 Camel 路由与 ActiveMQ 进行事务处理。回滚和最大重新交付可以正常工作，但重新交付延迟不能正常工作，这应该是增量的。

例如，当我未能处理消息（引发异常）时，它被重新传递了 3 次（如预期的那样），但之间没有时间（不是）。

我的 Spring 配置：

<context:annotation-config/>
<context:component-scan base-package="fr.dush.poc.springplaceholder"/>

<spring:camelContext>
    <spring:package>fr.dush.poc.springplaceholder.routes</spring:package>
    <spring:contextScan/>
</spring:camelContext>

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

<bean id="PROPAGATION_REQUIRED" class="org.apache.camel.spring.spi.SpringTransactionPolicy">
    <property name="transactionManager" ref="jmsTransactionManager"/>
</bean>
<bean id="PROPAGATION_REQUIRES_NEW" class="org.apache.camel.spring.spi.SpringTransactionPolicy">
    <property name="transactionManager" ref="jmsTransactionManager"/>
    <property name="propagationBehaviorName" value="PROPAGATION_REQUIRES_NEW"/>
</bean>

Spring 配置在配置 bean 中继续：

@Component
public class CamelFactories {

private static final Logger LOGGER = LoggerFactory.getLogger(CamelFactories.class);

public static final int REDELIVERY_DELAY = 1000;
public static final int BACK_OFF_MULTIPLIER = 2;
public static final int HOUR = 3600000;
public static final int MAXIMUM_REDELIVERY_DELAY = 2 * HOUR;
public static final int MAXIMUM_REDELIVERIES = 3;

@Bean(name = "jmsConnectionFactory")
public ActiveMQConnectionFactory createFactory() {
    ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory();
    factory.setBrokerURL("tcp://localhost:61616");

    RedeliveryPolicy policy = new RedeliveryPolicy() {

        @Override
        public long getNextRedeliveryDelay(long previousDelay) {
            long nextDelay = super.getNextRedeliveryDelay(previousDelay);
            LOGGER.warn("Previous delay={} ; This delay={} ", previousDelay, nextDelay);
            return nextDelay;
        }
    };
    policy.setMaximumRedeliveries(MAXIMUM_REDELIVERIES);

    policy.setRedeliveryDelay(REDELIVERY_DELAY);
    policy.setBackOffMultiplier(BACK_OFF_MULTIPLIER);
    policy.setUseExponentialBackOff(true);
    policy.setMaximumRedeliveryDelay(MAXIMUM_REDELIVERY_DELAY);

    factory.setRedeliveryPolicy(policy);

    return factory;
}

@Bean(name = "activemq")
public JmsComponent createJmsComponent(JmsTransactionManager transactionManager,
        ActiveMQConnectionFactory connectionFactory) {

    ActiveMQComponent component = new ActiveMQComponent();
    component.setTransactionManager(transactionManager);
    component.setConnectionFactory(connectionFactory);
    component.setTransacted(true);

    return component;
}

我的路线很简单：

public class CamelRouteBuilder extends SpringRouteBuilder {

  @Override
  public void configure() throws Exception {

    Policy required = getApplicationContext().getBean("PROPAGATION_REQUIRED",
            SpringTransactionPolicy.class);

    from("activemq:queue:foo.bar")
            .transacted()
            .policy(required)

            .log(LoggingLevel.INFO, "fr.dush.poc", "Receive message: ${body}")
            .beanRef("serviceBean") // throw an exception
            .to("mock:routeEnd");
  }
}

在我的日志中，我有 3 次，之前的 delay=0：

CamelFactories:36 - Previous delay=0 ; This delay=1000

似乎不是我一个人有这个问题，但我仍然没有找到解决方案......

谢谢，

-杜什

Answer 1

这可能通过在 ActiveMQComponent 上设置 cacheLevelName=CACHE_CONSUMER 来解决。我有同样的症状，这为我解决了。在相关说明中，除非我使用 CACHE_CONSUMER，否则我也会使用事务处理组件来乱序传递消息。

Answer 2

我仍然没有找到解决方案。但我找到了另一种选择：从 CAMEL 本身重试 API。

配置非常相似。 Spring 配置示例：

<redeliveryPolicyProfile id="infiniteRedeliveryPolicy" 
                         asyncDelayedRedelivery="true"
                         redeliveryDelay="${camel.redelivery_delay}"
                         maximumRedeliveryDelay="${camel.maximum_redelivery_delay}"
                         maximumRedeliveries="${camel.infinite_redelivery}"
                         backOffMultiplier="${camel.back_off_multiplier}"
                         useExponentialBackOff="true"/>

<routeContext>
  <route>
    <!-- ... -->

    <!-- Define behaviour in case of technical error -->
    <onException redeliveryPolicyRef="infiniteRedeliveryPolicy">
       <exception>java.lang.Exception</exception>
       <handled>
           <constant>false</constant>
       </handled>

       <log message="Message can't be processed for now. I'll retry later!" />
    </onException>
  </route>
</routeContext>

如果您想在 ActiveMQ 队列中保留未处理的消息，即使您关闭了应用程序，消费者也应该是事务性的。