AMQ的持久化问题 前言
前面一篇AMQ专题中，我们发现对于Topic这种类型的消息，即使将deliveryMode设置为持久化，只要生产者在消费者之前启动。消息生产者发布的消息还是会丢失。这是符合JMS规范的。
当然，作为一个如此活跃的开源消息中间件，在实现JMS基本规范之后，必然会通过扩展的方式来实现Topic的持久化订阅。
而所谓的deliveryMode持久化和订阅持久化还是两个不同的概念。本篇博客我们就通过实例来一探究竟。
在前面一篇中，我们通过producer.setDeliveryMode(DeliveryMode.PERSISTENT);将消息传递特性置为持久化，当时尝试过当消息类型是topic的时候，不管该值设置为啥，只要先启动Producer,那么对于后启动的Consumer都无法获取原来发布的主题。
那么这个DeliveryMode究竟是用来干啥的呢？
DeliveryMode中的是否持久化，指的是当重启activeMQ之后，原来队列或者主题中未被消费的消息是否仍然保留
我这里自己通过代码进行了如下测试，测试步骤和结果如下：
创建producer，并将producer的deliveryMode设置成持久化，运行producer
在消息被consumer消费之前，重启activeMQ
运行consumer，发现接收到了activeMQ重启之前Producer发送的消息
修改producer，将producer的deliveryMode设置成非持久化，运行producer
在消息被consumer消费之前，重启activeMQ
运行consumer，没有接收到任何消息，原producer产生的消息丢失
持久化和非持久化最终队列控制台分别如下:
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) { } } } } } 最终我的验证步骤和结果如下:
运行producer，向activeMQ发送主题消息
运行consumer，发现未收到任何消息
运行producer，此时运行中的consumer接收到了topic消息
停止运行consumer，重新运行producer
重新运行consumer，此时consumer接收到了刚刚producer产生的消息
创建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的持久化和订阅持久化是两个不同的概念，二者互不干扰，组合实现业务需求
需要弄清参数的实际意义第一步自己动手写实例，看运行结果是否与自己预期一致。第二步则是情况允许的时候，多看源码，掌握好的代码和设计