ActiveMQ Topic持久化订阅

MQ学习系列：

1. 消息队列概念与认知
2. ActiveMQ Topic消息重发
3. ActiveMQ Topic 消息持久化订阅
4. zookeeper+ActiveMQ集群实现高可用

---

一、持久化到文件（默认）

第一步：在${activemq.base}/conf/activemq.xml文件中配置持久化适配器。

<!--
    Configure message persistence for the broker. The default persistence
    mechanism is the KahaDB store (identified by the kahaDB tag).
    For more information, see:
    http://activemq.apache.org/persistence.html
-->
<persistenceAdapter>
    <kahaDB directory="${activemq.data}/kahadb"/>
</persistenceAdapter>

以上配置为默认，文件地址为E:ActiveMQapache-activemq-5.15.8datakahadb目录下。

第二步：发送消息时代码

//参数：消息， 持久化/不持久化，优先级，存储时长ms
producer.send(message,DeliveryMode.PERSISTENT,1,1000*60*5);

我们看看DeliveryMode这个类是什么鬼？

// IntelliJ API Decompiler stub source generated from a class file
// Implementation of methods is not available

package javax.jms;

public interface DeliveryMode {
    int NON_PERSISTENT = 1;
    int PERSISTENT = 2;
}

第三步：消息订阅者创建消费对象时

• connection start（）前添加客户端id

 //设置客户端id
 connection.setClientID("client-1");

• 创建客户端持久化订阅customer

//客户端持久化订阅
TopicSubscriber consumer = session.createDurableSubscriber(topic, "client1-sub");

经过如上三步，我们便可以通过文件来持久化我们的Topic订阅。

二、持久化到数据库（mysql）

第一步：将MySQL的数据库驱动复制到ActiveMQ的lib目录下。

第二步：在${activemq.base}/conf/activemq.xml文件中配置持久化配置。

<persistenceAdapter>
	<jdbcPersistenceAdapter dataDirectory="${activemq.base}/data" 
                            dataSource="#mysql-ds"/>
</persistenceAdapter>

第三步：在${activemq.base}/conf/activemq.xml文件中配置数据源。

<bean id="mysql-ds" class="org.apache.commons.dbcp2.BasicDataSource" 
      destroy-method="close">
    <property name="driverClassName" value="com.mysql.jdbc.Driver"/>
    <property name="url" 
              value="jdbc:mysql://localhost:3306/activemq?relaxAutoCommit=true"/>
    <property name="username" value="root"/>
    <property name="password" value="root"/>
    <property name="poolPreparedStatements" value="true"/>
</bean>

配置完成启动程序，系统自动生成三张表；

具体表结构参考 这里