ActiveMQ专题2: 持久化

AMQ的持久化问题

前言

​ 前面一篇AMQ专题中，咱们发现对于Topic这种类型的消息，即便将deliveryMode设置为持久化，只要生产者在消费者以前启动。消息生产者发布的消息仍是会丢失。这是符合JMS规范的。
固然，做为一个如此活跃的开源消息中间件，在实现JMS基本规范以后，必然会经过扩展的方式来实现Topic的持久化订阅。
而所谓的deliveryMode持久化和订阅持久化仍是两个不一样的概念。本篇博客咱们就经过实例来一探究竟。

DeliveryMode持久化

​ 在前面一篇中，咱们经过producer.setDeliveryMode(DeliveryMode.PERSISTENT);将消息传递特性置为持久化，可是当消息类型是topic的时候，无论该值设置为啥，只要先启动Producer,那么对于后启动的Consumer都没法获取原来发布的主题。

​ 那么这个DeliveryMode到底是用来干啥的呢？

• DeliveryMode中的是否持久化，指的是当重启activeMQ以后，原来队列或者主题中未被消费的消息是否仍然保留

​ 我这里本身经过代码进行了以下测试，测试步骤和结果以下：

1. 建立producer，并将producer的deliveryMode设置成持久化，运行producer
2. 在消息被consumer消费以前，重启activeMQ
3. 运行consumer，发现接收到了activeMQ重启以前Producer发送的消息
4. 修改producer，将producer的deliveryMode设置成非持久化，运行producer
5. 在消息被consumer消费以前，重启activeMQ
6. 运行consumer，没有接收到任何消息，原producer产生的消息丢失

​ 持久化和非持久化最终队列控制台分别以下:

至此，不难发现，deliveryMode的是否持久化是针对activeMQ服务器是否重启而言的。对于不支持持久化的设置，当mq重启以后，没有被消费的消息就会丢失。而支持持久化的设置，只要消息没有被消费，重启mq，仍然能被新加入的consumer消费。

订阅持久化

​ JMS的规范是没有要求实现订阅持久化的。所幸的是activeMQ实现了这个特性。我的认为所谓的订阅持久化相对于消息的持久化，不过是一种伪持久化。先不作太多说明，咱们直接看一个示例代码：

生产者

public class SimpleProducer {
    public static void main(String[] args) {
        // STEP1: 获得链接工厂
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD, ActiveMQConnection.DEFAULT_BROKER_URL);
        
        Connection connection = null;
        Session session = null;
        MessageProducer topicProducer = null;
        Destination topicDestination = null;
        try {
            // STEP2: 从链接工厂获得链接而且启动链接
            connection = connectionFactory.createConnection();
            connection.start();
            
            // STEP3: 获取会话
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            
            // STEP4: 建立主题 
            topicDestination = session.createTopic("KiDe-topic-Demo");
            
            // STEP5: 建立消息生产者
            topicProducer = session.createProducer(topicDestination);
            topicProducer.setDeliveryMode(DeliveryMode.PERSISTENT);     // 设置为持久化
            
            
            // STEP6: 发送消息
            for (int i=0; i<20; i++) {
                TextMessage message = session.createTextMessage("Producer message:" + i);
                topicProducer.send(message);
            }
            
            // STEP7: 若是开启了事务 ，此时须要调用session提交操做
            // session.commit();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (connection != null) {
                try {
                    connection.close();
                } catch (JMSException e) {
                }
            }
        }
    }
}

消费者

public class SimpleConsumer {
    public static void main(String[] args) {
        // STEP1: 建立链接工厂
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD, ActiveMQConnection.DEFAULT_BROKER_URL);

        Connection connection = null;
        Session session = null;
        MessageConsumer topicConsumer = null;
        try {
            // STEP2: 从链接工厂获得链接而且启动链接
            connection = connectionFactory.createConnection();
            connection.setClientID("1");        // 若是要进行持久化订阅，必须对链接设置clientID
            connection.start();

            // STEP3: 获取会话
            session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            
            // STEP4: 建立持久化订阅者
            TopicSubscriber topicSubscriber = session.createDurableSubscriber(session.createTopic("KiDe-topic-Demo"), "1");
                        
            // STEP5: 设置消息接收监听
            topicSubscriber.setMessageListener(new MessageListener() {
                
                @Override
                public void onMessage(Message paramMessage) {
                    TextMessage message = (TextMessage) paramMessage;
                    try {
                        System.out.println("消费者接收到主题消息：" + message.getText());
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
            TimeUnit.SECONDS.sleep(200);    // 睡眠200秒，使得客户端能够接收到对应消息
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (connection != null) {
                try {
                    connection.close();
                } catch (JMSException e) {
                }
            }
        }
    }
}

​ 最终个人验证步骤和结果以下:

1. 运行producer，向activeMQ发送主题消息
2. 运行consumer，发现未收到任何消息
3. 运行producer，此时运行中的consumer接收到了topic消息
4. 中止运行consumer，从新运行producer
5. 从新运行consumer，此时consumer接收到了刚刚producer产生的消息
6. 建立consumer的session的时候，同时建立两个同clientId的session时会报同一通道已被占用的错误

​ 分析以上步骤，我最终对这种伪持久化订阅的总结以下：

• 要实现伪持久化订阅，必须先向activeMQ发布持久化订阅消息，经过clientId来标识不一样的订阅渠道。

  若是在发布持久化订阅消息以前producer就向mq发送了topic消息，那么consumer仍是无法接收

• activeMQ肯定是不是同一持久化订阅者的依据条件有两个:connection.setClientID("3")中的clientId

  以及session.createDurableSubscriber(session.createTopic("KiDe-topic-Demo"), "12")中的name

总结

• deliveryMode的持久化和订阅持久化是两个不一样的概念，两者互不干扰，组合实现业务需求
• 须要弄清参数的实际意义第一步本身动手写实例，看运行结果是否与本身预期一致。第二步则是状况容许的时候，多看源码，掌握好的代码和设计