1 持久化订阅概述

在上篇文章中我们看到通过发布订阅模式的时候要先启动消费者，再启动生产者才能消费消息，这种订阅类型是非持久化订阅
持久化订阅的时候只要消费者启动注册后，之后消费者如果掉线然后再上线，那么消费者仍然可以消费到掉线期间的消息。

2 代码测试

2.1 容器中注入一个持久化消费者ActiveMqContext

@Bean
public ActiveMqContext persistentTopicContext() throws JMSException {return ActiveMQUtil.getTopicContext(Session.AUTO_ACKNOWLEDGE, DeliveryMode.PERSISTENT, persistentTopicName);
}

2.2 TopicProcedure中添加持久发送的方法

@Resource(name = "persistentTopicContext")
private ActiveMqContext persistentTopicContext;public void sendPersistentMsg(String msg) throws JMSException {ActiveMQUtil.sendMsg(persistentTopicContext, msg);
}

2.3 TopicConsumer添加持久化消费方法

@Resource(name = "persistentTopicContext")
private ActiveMqContext persistentContext;public void receivePersistentMsg() throws JMSException {ActiveMQUtil.receiveMsg(persistentContext);
}

2.4 TopicListener添加持久化监听方法

@Resource(name = "persistentTopicContext")
private ActiveMqContext persistentContext;public void receivePersistentMsg() throws JMSException {ActiveMQUtil.listenMsg(persistentContext);
}

2.5 测试代码

2.5.1 发送消息

@Test
public void sendPersistentMsg() throws JMSException, InterruptedException {for (int i = 0; i < 10; i ++) {procedure.sendPersistentMsg("msg-" + i);Thread.sleep(100);}
}

输出
2020-07-17 09:14:50.687 - [ActiveMQ Transport: tcp://localhost/127.0.0.1:61616@62623] INFO org.apache.activemq.jms.pool.PooledProducer : 106 - send mq msg success, msg:ActiveMQTextMessage {commandId = 0, responseRequired = false, messageId = ID:LAPTOP-7KJFACUD-62614-1594948489750-1:9:1:1:1, originalDestination = null, originalTransactionId = null, producerId = null, destination = topic://demo-persistent-topic, transactionId = null, expiration = 0, timestamp = 1594948490687, arrival = 0, brokerInTime = 0, brokerOutTime = 0, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = false, readOnlyBody = false, droppable = false, jmsXGroupFirstForConsumer = false, text = msg-0}
2020-07-17 09:14:50.796 - [ActiveMQ Transport: tcp://localhost/127.0.0.1:61616@62623] INFO org.apache.activemq.jms.pool.PooledProducer : 106 - send mq msg success, msg:ActiveMQTextMessage {commandId = 0, responseRequired = false, messageId = ID:LAPTOP-7KJFACUD-62614-1594948489750-1:9:1:1:2, originalDestination = null, originalTransactionId = null, producerId = null, destination = topic://demo-persistent-topic, transactionId = null, expiration = 0, timestamp = 1594948490796, arrival = 0, brokerInTime = 0, brokerOutTime = 0, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = false, readOnlyBody = false, droppable = false, jmsXGroupFirstForConsumer = false, text = msg-1}
2020-07-17 09:14:50.906 - [ActiveMQ Transport: tcp://localhost/127.0.0.1:61616@62623] INFO org.apache.activemq.jms.pool.PooledProducer : 106 - send mq msg success, msg:ActiveMQTextMessage {commandId = 0, responseRequired = false, messageId = ID:LAPTOP-7KJFACUD-62614-1594948489750-1:9:1:1:3, originalDestination = null, originalTransactionId = null, producerId = null, destination = topic://demo-persistent-topic, transactionId = null, expiration = 0, timestamp = 1594948490906, arrival = 0, brokerInTime = 0, brokerOutTime = 0, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = false, readOnlyBody = false, droppable = false, jmsXGroupFirstForConsumer = false, text = msg-2}
2020-07-17 09:14:51.016 - [ActiveMQ Transport: tcp://localhost/127.0.0.1:61616@62623] INFO org.apache.activemq.jms.pool.PooledProducer : 106 - send mq msg success, msg:ActiveMQTextMessage {commandId = 0, responseRequired = false, messageId = ID:LAPTOP-7KJFACUD-62614-1594948489750-1:9:1:1:4, originalDestination = null, originalTransactionId = null, producerId = null, destination = topic://demo-persistent-topic, transactionId = null, expiration = 0, timestamp = 1594948491016, arrival = 0, brokerInTime = 0, brokerOutTime = 0, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = false, readOnlyBody = false, droppable = false, jmsXGroupFirstForConsumer = false, text = msg-3}
2020-07-17 09:14:51.125 - [ActiveMQ Transport: tcp://localhost/127.0.0.1:61616@62623] INFO org.apache.activemq.jms.pool.PooledProducer : 106 - send mq msg success, msg:ActiveMQTextMessage {commandId = 0, responseRequired = false, messageId = ID:LAPTOP-7KJFACUD-62614-1594948489750-1:9:1:1:5, originalDestination = null, originalTransactionId = null, producerId = null, destination = topic://demo-persistent-topic, transactionId = null, expiration = 0, timestamp = 1594948491125, arrival = 0, brokerInTime = 0, brokerOutTime = 0, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = false, readOnlyBody = false, droppable = false, jmsXGroupFirstForConsumer = false, text = msg-4}
2020-07-17 09:14:51.234 - [ActiveMQ Transport: tcp://localhost/127.0.0.1:61616@62623] INFO org.apache.activemq.jms.pool.PooledProducer : 106 - send mq msg success, msg:ActiveMQTextMessage {commandId = 0, responseRequired = false, messageId = ID:LAPTOP-7KJFACUD-62614-1594948489750-1:9:1:1:6, originalDestination = null, originalTransactionId = null, producerId = null, destination = topic://demo-persistent-topic, transactionId = null, expiration = 0, timestamp = 1594948491234, arrival = 0, brokerInTime = 0, brokerOutTime = 0, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = false, readOnlyBody = false, droppable = false, jmsXGroupFirstForConsumer = false, text = msg-5}
2020-07-17 09:14:51.344 - [ActiveMQ Transport: tcp://localhost/127.0.0.1:61616@62623] INFO org.apache.activemq.jms.pool.PooledProducer : 106 - send mq msg success, msg:ActiveMQTextMessage {commandId = 0, responseRequired = false, messageId = ID:LAPTOP-7KJFACUD-62614-1594948489750-1:9:1:1:7, originalDestination = null, originalTransactionId = null, producerId = null, destination = topic://demo-persistent-topic, transactionId = null, expiration = 0, timestamp = 1594948491344, arrival = 0, brokerInTime = 0, brokerOutTime = 0, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = false, readOnlyBody = false, droppable = false, jmsXGroupFirstForConsumer = false, text = msg-6}
2020-07-17 09:14:51.453 - [ActiveMQ Transport: tcp://localhost/127.0.0.1:61616@62623] INFO org.apache.activemq.jms.pool.PooledProducer : 106 - send mq msg success, msg:ActiveMQTextMessage {commandId = 0, responseRequired = false, messageId = ID:LAPTOP-7KJFACUD-62614-1594948489750-1:9:1:1:8, originalDestination = null, originalTransactionId = null, producerId = null, destination = topic://demo-persistent-topic, transactionId = null, expiration = 0, timestamp = 1594948491453, arrival = 0, brokerInTime = 0, brokerOutTime = 0, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = false, readOnlyBody = false, droppable = false, jmsXGroupFirstForConsumer = false, text = msg-7}
2020-07-17 09:14:51.562 - [ActiveMQ Transport: tcp://localhost/127.0.0.1:61616@62623] INFO org.apache.activemq.jms.pool.PooledProducer : 106 - send mq msg success, msg:ActiveMQTextMessage {commandId = 0, responseRequired = false, messageId = ID:LAPTOP-7KJFACUD-62614-1594948489750-1:9:1:1:9, originalDestination = null, originalTransactionId = null, producerId = null, destination = topic://demo-persistent-topic, transactionId = null, expiration = 0, timestamp = 1594948491562, arrival = 0, brokerInTime = 0, brokerOutTime = 0, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = false, readOnlyBody = false, droppable = false, jmsXGroupFirstForConsumer = false, text = msg-8}
2020-07-17 09:14:51.672 - [ActiveMQ Transport: tcp://localhost/127.0.0.1:61616@62623] INFO org.apache.activemq.jms.pool.PooledProducer : 106 - send mq msg success, msg:ActiveMQTextMessage {commandId = 0, responseRequired = false, messageId = ID:LAPTOP-7KJFACUD-62614-1594948489750-1:9:1:1:10, originalDestination = null, originalTransactionId = null, producerId = null, destination = topic://demo-persistent-topic, transactionId = null, expiration = 0, timestamp = 1594948491672, arrival = 0, brokerInTime = 0, brokerOutTime = 0, correlationId = null, replyTo = null, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = false, readOnlyBody = false, droppable = false, jmsXGroupFirstForConsumer = false, text = msg-9}

2.5.2 消费消息

@Test
public void receivePersistentMsg() throws JMSException {consumer.receivePersistentMsg();
}

2020-07-17 09:15:52.278 - [main] INFO com.jms.activemq.ActiveMQUtil : 150 - receive msg:msg-0
2020-07-17 09:15:52.278 - [main] INFO com.jms.activemq.ActiveMQUtil : 150 - receive msg:msg-1
2020-07-17 09:15:52.278 - [main] INFO com.jms.activemq.ActiveMQUtil : 150 - receive msg:msg-2
2020-07-17 09:15:52.278 - [main] INFO com.jms.activemq.ActiveMQUtil : 150 - receive msg:msg-3
2020-07-17 09:15:52.278 - [main] INFO com.jms.activemq.ActiveMQUtil : 150 - receive msg:msg-4
2020-07-17 09:15:52.278 - [main] INFO com.jms.activemq.ActiveMQUtil : 150 - receive msg:msg-5
2020-07-17 09:15:52.278 - [main] INFO com.jms.activemq.ActiveMQUtil : 150 - receive msg:msg-6
2020-07-17 09:15:52.278 - [main] INFO com.jms.activemq.ActiveMQUtil : 150 - receive msg:msg-7
2020-07-17 09:15:52.278 - [main] INFO com.jms.activemq.ActiveMQUtil : 150 - receive msg:msg-8
2020-07-17 09:15:52.278 - [main] INFO com.jms.activemq.ActiveMQUtil : 150 - receive msg:msg-9

2.5.3 监听消息

@Test
public void listenerPersistentMsg() throws JMSException {listener.receivePersistentMsg();
}

输出
2020-07-17 09:17:05.373 - [ActiveMQ Session Task-1] INFO com.jms.activemq.ActiveMQUtil : 168 - listen msg:msg-0
2020-07-17 09:17:05.373 - [ActiveMQ Session Task-1] INFO com.jms.activemq.ActiveMQUtil : 168 - listen msg:msg-1
2020-07-17 09:17:05.373 - [ActiveMQ Session Task-1] INFO com.jms.activemq.ActiveMQUtil : 168 - listen msg:msg-2
2020-07-17 09:17:05.373 - [ActiveMQ Session Task-1] INFO com.jms.activemq.ActiveMQUtil : 168 - listen msg:msg-3
2020-07-17 09:17:05.373 - [ActiveMQ Session Task-1] INFO com.jms.activemq.ActiveMQUtil : 168 - listen msg:msg-4
2020-07-17 09:17:05.373 - [ActiveMQ Session Task-1] INFO com.jms.activemq.ActiveMQUtil : 168 - listen msg:msg-5
2020-07-17 09:17:05.373 - [ActiveMQ Session Task-1] INFO com.jms.activemq.ActiveMQUtil : 168 - listen msg:msg-6
2020-07-17 09:17:05.373 - [ActiveMQ Session Task-1] INFO com.jms.activemq.ActiveMQUtil : 168 - listen msg:msg-7
2020-07-17 09:17:05.373 - [ActiveMQ Session Task-1] INFO com.jms.activemq.ActiveMQUtil : 168 - listen msg:msg-8
2020-07-17 09:17:05.373 - [ActiveMQ Session Task-1] INFO com.jms.activemq.ActiveMQUtil : 168 - listen msg:msg-9

3 持久订阅小结

1 持久订阅的时候一定要先启动一次消费者进行注册，否则消费不到之前的消息
2 持久订阅的消费者要设置clentId，每个clientId对应的消费者都会拿到一份全量的消息

ActiveMQ学习三-持久化订阅相关推荐

ActiveMQ（三）：ActiveMQ的安全机制、api及订阅模式demo
一.ActiveMQ安全机制 ActiveMQ是使用jetty部署的,修改密码需要到相应的配置文件配置文件是这个: 在其第123行添加用户名和密码,添加配置如下: <plugins>& ...
ActiveMQ学习笔记03 - 消息持久化
2019独角兽企业重金招聘Python工程师标准>>> ActiveMQ的消息持久化机制有JDBC,AMQ,KahaDB和LevelDB,还有一种内存存储的方式,由于内存不属于持久化 ...
ROS与Arduino学习（三）订阅与发布
ROS与Arduino学习(三)订阅与发布 Tutorial Level:HelloWorld Next Tutorial:用ROS的Cmake编译程序本节介绍如何在arduino中发布与订阅消息. ...
ActiveMQ学习（七）
2019独角兽企业重金招聘Python工程师标准>>> Destination的高级特性 1.组合destinations(虚拟queue) 指的是组合队列,就是虚拟destinat ...
Docker学习三：Docker 数据管理
前言本次学习来自于datawhale组队学习: 教程地址为: https://github.com/datawhalechina/team-learning-program/tree/master/ ...
ActiveMQ专题2: 持久化
AMQ的持久化问题前言前面一篇AMQ专题中,我们发现对于Topic这种类型的消息,即使将deliveryMode设置为持久化,只要生产者在消费者之前启动.消息生产者发布的消息还是会丢失.这是符 ...
单线激光雷达（Lidar）学习三：使用雷达数据/scan转/PointCloud后生成鸟瞰图
单线激光雷达(Lidar)学习三:使用雷达数据/scan转/PointCloud后生成鸟瞰图前言: 雷达广泛应用于自动驾驶中,作用非常重要,是自动驾驶无人车中的作为"眼睛"的一环 ...
ActiveMQ学习笔记07 - 优缺点
2019独角兽企业重金招聘Python工程师标准>>> 优点: 可以用JDBC 虽然使用JDBC会降低ActiveMQ的性能,但是数据库一直都是开发人员最熟悉的存储介质.将消息存到数 ...
统计学习三要素模型＋策略＋算法
统计学习方法都是由模型. 策略和算法构成的. 即统计学习方法由三要素构成, 可以简单地表示为:方法＝模型+策略+算法模型统计学习首要考虑的问题是学习什么样的模型. 在监督学习过程中, 模型就是所要 ...
深度学习三巨头也成了大眼萌，这个一键转换动画电影形象的网站竟因「太火」而下线...
机器之心报道作者:魔王.杜伟想不想在动画电影中拥有自己的角色?这个网站一键满足你的需求,不过竟因流量太大成本过高而下线. 近期热映的电影<花木兰>总是让人回想起 1998 年上映的同名 ...

ActiveMQ学习三-持久化订阅