ActiveMQ Artemis服务中断时未处理消息丢失的配置解决方案

已为JMS消费者配置事务管理器，同时实现了从死信队列（DLQ）自动将消息重发回原队列的功能，但存在以下问题：只有当标注@Transactional的事务方法抛出异常导致事务失败时，消息才会被移入DLQ；若处理消息时消费者服务意外关闭，消息会直接从队列中消失，希望这类消息能回到队列重新处理，需补充哪些配置？

现有配置代码 #

1. 消费者监听器容器工厂与事务管理器 #

@Bean
public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(
        @Qualifier("jmsSimpleConnectionFactory") ConnectionFactory connectionFactory,
        DefaultJmsListenerContainerFactoryConfigurer containerFactoryConfigurer,
        PlatformTransactionManager transactionManager) {

    DefaultJmsListenerContainerFactory listenerContainerFactory = new DefaultJmsListenerContainerFactory();

    containerFactoryConfigurer.configure(listenerContainerFactory, connectionFactory);

    listenerContainerFactory.setMessageConverter(this.jmsMessageConverter);
    listenerContainerFactory.setSessionAcknowledgeMode(Session.SESSION_TRANSACTED);
    listenerContainerFactory.setCacheLevel(DefaultMessageListenerContainer.CACHE_CONNECTION);
    listenerContainerFactory.setTransactionManager(transactionManager);

    // 省略其他配置
}

@Bean
public PlatformTransactionManager transactionManager(ConnectionFactory connectionFactory) {
    return new JmsTransactionManager(connectionFactory);
}

2. DLQ重发相关配置 #

@Bean
public DefaultJmsListenerContainerFactory dlqListenerContainerFactory(PlatformTransactionManager transactionManager) {
    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
    factory.setConnectionFactory(jmsSimpleConnectionFactory);
    factory.setTransactionManager(transactionManager);
    factory.setSessionTransacted(true);
    return factory;
}

@JmsListener(destination = "DLQ", containerFactory = "dlqListenerContainerFactory")
public void processDLQMessage(String message) {
    jmsTemplate.convertAndSend(sourceQueue, message);
}

要解决服务意外关闭时的消息丢失问题，核心是让未处理完成的消息始终处于MQ服务器的事务管控下，确保服务中断后消息能自动重回队列。具体需调整以下配置：

1. 确保消息为持久化模式 #

只有持久化消息才会被MQ服务器持久存储，服务意外关闭后不会丢失。如果使用JmsTemplate发送消息，默认已是持久化模式，可显式确认配置：

jmsTemplate.setDeliveryMode(DeliveryMode.PERSISTENT);

2. 调整监听器容器的缓存与事务配置 #

当前设置的CACHE_CONNECTION会复用连接和会话，可能导致事务边界模糊，服务关闭时未提交的事务无法正常回滚。需修改缓存级别：

listenerContainerFactory.setCacheLevel(DefaultMessageListenerContainer.CACHE_NONE);

同时显式开启会话事务，配合事务管理器确保消息处理完成后才提交事务：

listenerContainerFactory.setSessionTransacted(true);

3. 配置MQ服务器的预取与重发策略 #

以ActiveMQ为例，限制预取消息数量（建议设为1），避免一次性拉取过多消息到消费者内存，导致服务关闭时这些消息脱离MQ管控：

ActiveMQConnectionFactory connectionFactory = (ActiveMQConnectionFactory) jmsSimpleConnectionFactory;
// 设置队列预取数为1
connectionFactory.setPrefetchPolicy(new ActiveMQPrefetchPolicy().setQueuePrefetch(1));

同时配置消息重发规则，避免无限循环重发，超过次数后移入DLQ：

RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
redeliveryPolicy.setMaximumRedeliveries(3); // 最多重发3次
redeliveryPolicy.setInitialRedeliveryDelay(1000); // 首次重发延迟1秒
connectionFactory.setRedeliveryPolicy(redeliveryPolicy);

4. 强化DLQ处理的事务一致性 #

在DLQ消息处理方法上添加@Transactional注解，确保从DLQ删除消息和发送回原队列的操作在同一事务中执行，避免消息转移过程中丢失：

@Transactional
@JmsListener(destination = "DLQ", containerFactory = "dlqListenerContainerFactory")
public void processDLQMessage(String message) {
    jmsTemplate.convertAndSend(sourceQueue, message);
}

内容的提问来源于stack exchange，提问作者skyho