ActiveMQ - 保留已使用的消息，直到确认删除答案

【问题标题】：ActiveMQ - keep consumed messages until it's acknowledge to deleteActiveMQ - 保留已使用的消息，直到确认删除
【发布时间】：2018-03-13 16:22:37
【问题描述】：

我有一个正在侦听 ActiveMQ 队列的 Spring Boot 应用程序。

有没有办法告诉 ActiveMQ 保留已使用的消息，直到收到从我的服务中删除消息的确认？

我的场景是我从队列中读取消息，一旦使用，我将其发送到外部队列。如果外部服务无法处理消息，我会丢失消息。有没有我可以告诉活动 mq 保留消费的消息，直到它收到删除的确认。

【问题讨论】：

标签： java spring-boot activemq

【解决方案1】：

最安全的选择是使用 XA 事务以原子方式执行消费/发送（假设您发送消息的代理支持 XA）。我不确定这在 Spring 中是容易还是难，但我知道它在 Java EE 容器中非常简单。

【讨论】：

【解决方案2】：

您可以使用单个确认模式来确认单个消息。以下是相同的代码：

    import javax.jms.Connection;
    import javax.jms.JMSException;

    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.apache.activemq.ActiveMQMessageConsumer;
    import org.apache.activemq.ActiveMQSession;
    import org.apache.activemq.command.ActiveMQTextMessage;

    public class SimpleConsumer {

        public static void main(String[] args) throws JMSException {
            Connection conn = null;
            try {
                ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory("tcp://localhost:61616");
                conn = cf.createConnection("consumer", "consumer");
                ActiveMQSession session = (ActiveMQSession) conn.createSession(false,
                        ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE);
                ActiveMQMessageConsumer consumer = (ActiveMQMessageConsumer) session
                        .createConsumer(session.createQueue("QUEUE"));
                conn.start();
                ActiveMQTextMessage msg = null;
                while ((msg = (ActiveMQTextMessage) consumer.receive()) != null) {
                    System.out.println("Received message is: " + msg.getText());
               // Call your service and ack the message if successfully processed                    
                 msg.acknowledge();
                }
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (conn != null) {
                    try {
                        conn.close();
                    } catch (Exception e) {
                    }
                }
            }
        }
    }

【讨论】：

作者原来的问题是，当消费消息被确认但发送消息失败时，消息“丢失”。该解决方案为相反的问题打开了大门，即消息可以成功发送但无法被确认（例如由于网络中断），这会导致重新传递和发送重复的消息。如果发送消息是幂等的，那么这不是问题，但如果它是消费者需要能够检测重复或整个过程需要是原子的（例如使用 XA）。