Camel JMS事务异常：消息未转入ActiveMQ.DLQ问题求助

问题背景 #

我最近在实现一条带事务的Camel路由（JMS->HTTP4），遇到了一个棘手的问题：当REST服务端宕机触发ConnectException时，日志明明显示事务已回滚，但消息却直接被消费移除，没有转入预期的ActiveMQ.DLQ死信队列。我原本以为Camel结合ActiveMQ的事务模式会默认处理这类失败消息，翻了《Camel in Action》第9章和不少资料都没找到原因——虽然知道可以自定义TransactionErrorHandler指定存储队列，但这应该是事务模式下的默认行为才对呀。

环境信息 #

• ActiveMQ 5.15.2（默认独立安装配置）
• Camel 2.20.1
• Java 8_144
• MacOS 10.13.2

相关代码 #

配置类

@Configuration
public class Config {
    /**
     * The Camel context.
     */
    final CamelContext camelContext;
    /**
     * The Broker url.
     */
    @Value("${jms.broker.url}")
    private String brokerURL;
    /**
     * Instantiates a new Config.
     *
     * @param camelContext the sisyfos context
     * @param metricRegistry the metric registry
     */
    @Autowired
    public Config(final CamelContext camelContext, final MetricRegistry metricRegistry) {
        this.camelContext = camelContext;
        this.metricRegistry = metricRegistry;
    }
    @Bean
    public ActiveMQConnectionFactory activeMQConnectionFactory() {
        final ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory();
        activeMQConnectionFactory.setBrokerURL(brokerURL);
        return activeMQConnectionFactory;
    }
    /**
     * Pooled connection factory pooled connection factory.
     *
     * @return the pooled connection factory
     */
    @Bean
    @Primary
    public PooledConnectionFactory pooledConnectionFactory() {
        final PooledConnectionFactory pooledConnectionFactory = new PooledConnectionFactory();
        pooledConnectionFactory.setMaxConnections(8);
        pooledConnectionFactory.setMaximumActiveSessionPerConnection(500);
        pooledConnectionFactory.setConnectionFactory(activeMQConnectionFactory());
        return pooledConnectionFactory;
    }
    /**
     * Jms configuration jms configuration.
     *
     * @return the jms configuration
     */
    @Bean
    public JmsConfiguration jmsConfiguration() {
        final JmsConfiguration jmsConfiguration = new JmsConfiguration();
        jmsConfiguration.setConnectionFactory(pooledConnectionFactory());
        jmsConfiguration.setTransacted(true);
        jmsConfiguration.setTransactionManager(transactionManager());
        jmsConfiguration.setConcurrentConsumers(10);
        return jmsConfiguration;
    }
    /**
     * Transaction manager jms transaction manager.
     *
     * @return the jms transaction manager
     */
    @Bean
    public JmsTransactionManager transactionManager() {
        final JmsTransactionManager transactionManager = new JmsTransactionManager();
        transactionManager.setConnectionFactory(pooledConnectionFactory());
        return transactionManager;
    }
    /**
     * Active mq component active mq component.
     *
     * @return the active mq component
     */
    @Bean
    public ActiveMQComponent activeMQComponent(JmsConfiguration jmsConfiguration, PooledConnectionFactory pooledConnectionFactory, JmsTransactionManager transactionManager) {
        final ActiveMQComponent activeMQComponent = new ActiveMQComponent();
        activeMQComponent.setConfiguration(jmsConfiguration);
        activeMQComponent.setTransacted(true);
        activeMQComponent.setUsePooledConnection(true);
        activeMQComponent.setConnectionFactory(pooledConnectionFactory);
        activeMQComponent.setTransactionManager(transactionManager);
        return activeMQComponent;
    }
}

路由类

@Component
public class SendToCore extends SpringRouteBuilder {
    @Override
    public void configure() throws Exception {
        Logger.getLogger(SendToCore.class).info("Sending to CORE");
        //No retries if first fails due to connection error
        interceptSendToEndpoint("http4:*")
                .choice()
                .when(header("JMSRedelivered").isEqualTo("false"))
                .throwException(new ConnectException("Cannot connect to CORE REST"))
                .end();
        from("activemq:queue:myIncomingQueue")
                .transacted()
                .setHeader(Exchange.CONTENT_TYPE, constant("application/xml"))
                .to("http4:localhost/myRESTservice")
                .log("${header.CamelHttpResponseCode}")
                .end();
    }
}

问题原因分析 #

这里的核心误解是：Camel事务回滚并不会直接将消息送入DLQ，而是依赖ActiveMQ自身的重投策略和死信机制。具体来说：

1. 当事务回滚时，Camel只是将消息放回原队列，等待下一次投递，并不会主动触发DLQ逻辑。
2. ActiveMQ的死信队列触发条件是：消息重投次数达到配置的最大值，或者消息被明确拒绝（比如调用session.reject()）。如果没配置重投次数限制，消息会一直循环投递，直到被处理或手动干预。
3. 你的路由中使用了interceptSendToEndpoint主动抛出异常，可能干扰了ActiveMQ对重投次数的计数（JMSXDeliveryCount属性），导致重投逻辑未正确触发。

解决方案 #

1. 配置ActiveMQ的重投与死信策略

修改ActiveMQ的activemq.xml，添加明确的重投次数限制和死信队列配置：

<broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="${activemq.data}">
  <!-- 其他配置 -->
  <destinationPolicy>
    <policyMap>
      <policyEntries>
        <!-- 对所有队列生效 -->
        <policyEntry queue=">">
          <!-- 配置死信策略：每个队列对应独立的DLQ，或统一使用ActiveMQ.DLQ -->
          <deadLetterStrategy>
            <individualDeadLetterStrategy 
                queuePrefix="ActiveMQ.DLQ." 
                useQueueForQueueMessages="true"
                processExpired="true"
                processNonPersistent="true"/>
          </deadLetterStrategy>
          <!-- 配置重投策略：最大3次重投，每次间隔1秒 -->
          <redeliveryPolicy>
            <redeliveryPolicy 
                maximumRedeliveries="3" 
                redeliveryDelay="1000"
                useExponentialBackOff="false"/>
          </redeliveryPolicy>
        </policyEntry>
      </policyEntries>
    </policyMap>
  </destinationPolicy>
  <!-- 其他配置 -->
</broker>

2. 调整Camel路由的错误处理逻辑

去掉自定义的interceptSendToEndpoint，让HTTP4组件自身抛出连接异常，确保ActiveMQ能正确计数重投次数。同时可以在路由中明确配置事务错误处理器，指定死信队列：

@Override
public void configure() throws Exception {
    Logger.getLogger(SendToCore.class).info("Sending to CORE");
    
    // 配置事务错误处理器，指定死信队列
    errorHandler(transactionErrorHandler()
            .deadLetterQueue("activemq:queue:ActiveMQ.DLQ")
            .maximumRedeliveries(3));

    from("activemq:queue:myIncomingQueue")
            .transacted()
            .setHeader(Exchange.CONTENT_TYPE, constant("application/xml"))
            .to("http4:localhost/myRESTservice")
            .log("${header.CamelHttpResponseCode}")
            .end();
}

3. 清理冗余的事务配置

你的配置类中同时在JmsConfiguration和ActiveMQComponent中设置了transacted=true，建议统一配置，避免冲突。可以保留JmsConfiguration中的事务设置，移除ActiveMQComponent中的重复配置：

@Bean
public ActiveMQComponent activeMQComponent(JmsConfiguration jmsConfiguration, PooledConnectionFactory pooledConnectionFactory, JmsTransactionManager transactionManager) {
    final ActiveMQComponent activeMQComponent = new ActiveMQComponent();
    activeMQComponent.setConfiguration(jmsConfiguration);
    // 移除重复的transacted=true
    activeMQComponent.setUsePooledConnection(true);
    activeMQComponent.setConnectionFactory(pooledConnectionFactory);
    activeMQComponent.setTransactionManager(transactionManager);
    return activeMQComponent;
}

4. 验证重投计数

通过ActiveMQ控制台查看消息的JMSXDeliveryCount属性，确认每次事务回滚后该计数是否递增。如果计数未变化，说明连接或事务配置存在问题，需要检查PooledConnectionFactory是否正确关联了事务管理器。

内容的提问来源于stack exchange，提问作者Mikael Andersson Wigander