Spring Boot监听ActiveMQ队列：如何保留已消费消息至确认删除？

当然可以实现这个需求！核心就是利用ActiveMQ的客户端手动确认机制，替代默认的自动确认模式。这样ActiveMQ只会在你的服务明确发送确认信号后才删除消息，完美解决你担心的外部服务处理失败导致消息丢失的问题。

一、核心原理 #

默认情况下，ActiveMQ使用AUTO_ACKNOWLEDGE模式：消息一旦被消费者接收，ActiveMQ就会自动标记为已消费并删除。而我们需要切换到CLIENT_ACKNOWLEDGE模式，此时消息会被ActiveMQ保留，直到你的服务显式调用确认方法才会被删除；如果服务崩溃或者没有发送确认，ActiveMQ会在一段时间后将消息重新投递（可配置重发策略）。

二、Spring Boot配置步骤 #

1. 配置ActiveMQ连接信息 #

在application.properties（或application.yml）里配置基础连接，同时关闭自动确认：

spring.activemq.broker-url=tcp://localhost:61616
spring.activemq.user=admin
spring.activemq.password=admin
# 指定为客户端手动确认模式
spring.jms.listener.acknowledge-mode=client

2. 自定义JMS监听容器（可选，用于精细控制） #

如果需要更灵活的配置（比如并发数、重发规则），可以自定义DefaultJmsListenerContainerFactory：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
import org.springframework.jms.connection.CachingConnectionFactory;
import javax.jms.ConnectionFactory;

@Configuration
public class JmsConfig {

    @Bean
    public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(CachingConnectionFactory connectionFactory) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        // 启用客户端手动确认
        factory.setSessionAcknowledgeMode(javax.jms.Session.CLIENT_ACKNOWLEDGE);
        // 可根据业务调整并发消费者数量
        factory.setConcurrency("1-5");
        return factory;
    }
}

3. 在消息监听器中手动确认 #

在你的消费方法里，完成业务逻辑（比如发送到外部队列）后手动确认消息；如果业务失败，不要调用确认，让ActiveMQ重新投递。

示例代码：

import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;
import javax.jms.Message;
import javax.jms.Session;

@Component
public class ActiveMQConsumer {

    @JmsListener(destination = "your-business-queue", containerFactory = "jmsListenerContainerFactory")
    public void consumeMessage(Message message, Session session) throws Exception {
        try {
            // 1. 解析消息内容
            String messageContent = message.getBody(String.class);
            
            // 2. 执行业务逻辑：发送到外部队列
            sendToExternalQueue(messageContent);
            
            // 3. 业务成功，手动确认消息，ActiveMQ会删除该消息
            message.acknowledge();
        } catch (Exception e) {
            // 业务失败，不确认消息，ActiveMQ会自动重发
            // 若遇到不可恢复的错误，可调用session.recover()将消息直接转入死信队列
            // session.recover();
            throw e; // 抛出异常触发重发机制
        }
    }

    private void sendToExternalQueue(String content) throws Exception {
        // 调用外部服务发送消息的逻辑
        // 如果这里抛出异常，就会进入catch块，消息不会被确认
    }
}

三、额外优化建议 #

• 重发策略配置：在ActiveMQ的activemq.xml中设置重发次数和间隔，避免无限重发：

<broker ...>
    <plugins>
        <redeliveryPlugin fallbackToDeadLetter="true" sendToDlqIfMaxRetriesExceeded="true">
            <redeliveryPolicyMap>
                <redeliveryPolicy queue="your-business-queue" maximumRedeliveries="3" initialRedeliveryDelay="1000" redeliveryDelay="2000"/>
            </redeliveryPolicyMap>
        </redeliveryPlugin>
    </plugins>
</broker>

• 死信队列（DLQ）：当消息重发达到最大次数后，会被移至死信队列，你可以后续批量处理这些失败消息，避免阻塞业务队列。

这样配置后，你的服务就能完全控制消息的确认时机，只有当外部服务处理成功时，ActiveMQ才会删除消息；如果失败，消息会被重新投递，彻底解决消息丢失的问题。

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