🔰 学习视频 🔰

尚硅谷ActiveMQ教程(MQ消息中间件快速入门)

集数：48—57

🔰 学习格言 🔰

练拳不练功，到老一场空；基础不牢，地动山摇。

rm activemq.xml
# 输入y表示确定

mv activemq.xml.bk activemq.xml

一、简介

官网：https://activemq.apache.org/persistence

MQ的高可用：事务、持久、签收、可持久化。

事务、持久、签收是MQ本身的。可持久化需要借助外力进行物理备份。

为了避免意外宕机以后丢失信息，需要做到重启后可以恢复消息队列，消息系统一般都会采用持久化机制。ActiveMQ的消息持久化机制有JDBC，AMQ， KahaDB和LevelDB，无论使用哪种持久化方式，消息的存储逻辑都是一致的。

就是在发送者将消息发送出去后，消息中心首先将消息存储到本地数据文件、内存数据库或者远程数据库等再试图将消息发送给接收者，成功则将消息从存储中删除，失败则继续尝试发送。

消息中心启动以后首先要检查指定的存储位置，如果有未发送成功的消息，则需要把消息发送出去。

二、持久化机制

2.1 AMQ Message Store（了解）

AMQ是一种文件存储形式，它具有写入速度快和容易恢复的特点。消息存储在一个个文件中，文件的默认大小为32M，当一个存储文件中的消息已经全部被消费，那么这个文件将被标识为可删除，在下一个清除阶段，这个文件被删除。AMQ适用于ActiveMQ5.3之前的版本。

2.2 Kaha DB

官网说明：https://activemq.apache.org/kahadb

基于日志文件，从ActiveMQ5.4开始默认的持久化插件。

🔶 查看

打开配置文件activemq.xml查看。

进入/activeMQ/apache-activemq-5.15.15/data目录，可以查看到

KahaDB是目前默认的存储方式,可用于任何场景提高了性能和恢复能力。消息存储使用一个事务日志（正文）和仅仅用一个索引文件（目录）来存储它所有的地址。KahaDB是一个专门针对消息持久化的解决方案，它对典型的消息使用模式进行了优化。数据被追加到data logs中。当不再需要log文件中的数据的时候，log文件会被丢弃。

🔶 存储机制

kahadb在消息保存目录中只有4类文件和一个lock，跟ActiveMQ的其他几种文件存储引擎相比这就非常简洁了。

组成 = 4个文件 + 一把锁

文件1：db-<Number>.log

KahaDB存储消息到预定义大小的数据记录文件中，文件命名为db-<Number> .log。 当数据文件已满时，一个新的文件会随之创建，number数值也会随之递增，它随着消息数量的增多，如每32M一个文件，文件名按照数字进行编号，如db-1.log、db-2.log、 db-3.log 。当不再有引用到数据文件中的任何消息时，文件会被删除或归档。

文件2：db.data

该文件包含了持久化的BTree索引，索引了消息数据记录中的消息,它是消息的索引文件，本质上是B-Tree (B树)，使用B-Tree作为索引指向db-<Number>.log里面存储的消息。

文件3：db.free

当前db.data文件里哪些页面是空闲的，文件具体内容是所有空闲页的ID.

文件4：db.redo

db.redo用来进行消息恢复，如果KahaDB消息存储在强制退出后启动，用于恢复BTree索引。

文件5：lock

文件锁，表示当前获得kahadb读写权限的broker

2.3 levelDB（了解）

这种文件系统是从ActiveMQ5.8之后引进的，它和KahaDB非常相似，也是基于文件的本地数据库储存形式，但是它提供比KahaDB更快的持久性。

但它不使用自定义B-Tree实现来索引预写日志，而是使用基于LevelDB的索引。

默认配置如下:

<persistenceAdapter><levelDB directory="activemq-data"/>
</persistenceAdapter>

三、JDBC（主要）

🔹 步骤1：分析

🔹 步骤2：拷贝驱动

拷贝MySQL驱动Jar包到activemq的lib目录下。

授权lib目录拷贝权限。
chmod 777 /activeMQ/apache-activemq-5.15.15/lib

🔹 步骤3：修改配置文件

将activemq.xml配置文件的81-83行进行修改。

修改前：

<persistenceAdapter><kahaDB directory="${activemq.data}/kahadb"/>
</persistenceAdapter>

修改后：

<persistenceAdapter><jdbcPersistenceAdapter dataSource="#mysql-ds"/>
</persistenceAdapter>

dataSource指定将要引用的持久化数据库的bean名称，createTablesOnStartup是否在启动的时候创建数据表，默认值是true，这样每次启动都会去创建数据表了，一般是第一次启动的时候设置为true之后改成false。

🔹 步骤4：数据库配置

添加数据库配置到配置文件，通常情况，MQ服务器和数据库服务器是分开的 ，所以这里将MySQL数据库创建在Windows系统上。

官网复制的配置如下：

在<broker>的标签外面，在<import resource="jetty.xml"/>之前。

<broker>...
</broker><!-- MySql DataSource Sample Setup --> 
<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/activemq?relaxAutoCommit=true"/> <property name="username" value="activemq"/> <property name="password" value="activemq"/> <property name="poolPreparedStatements" value="true"/> 
</bean><import resource="jetty.xml"/> 

根据自己的数据库进行配置。

<!-- MySql DataSource Sample Setup --> 
<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://192.168.56.1:3306/activemq?relaxAutoCommit=true&amp;serverTimezone=UTC&amp;useUnicode=true&amp;characterEncoding=utf-8&amp;useSSL=false"/> <property name="username" value="root"/> <property name="password" value="123456"/> <property name="poolPreparedStatements" value="true"/> 
</bean> 

为了防止报错，url后面增加了这么多配置。

jdbc:mysql://192.168.56.1:3306/activemq?relaxAutoCommit=true&serverTimezone=UTC&useUnicode=true&characterEncoding=utf-8&useSSL=false

特别注意：需要开启Windows本地数据库的远程访问：https://blog.csdn.net/ethan__xu/article/details/89320614，否则在步骤5时将启动失败。

注意：如果要使用druid数据库连接池，需要在步骤2引入druid的jar包。

🔹 步骤5：建仓SQL和建表说明

建立一个名为activemq的数据库

CREATE DATABASE activemq;

重新启动activemq，可以查看到数据库中出现三张表格。

我在运行时出现错误没法启动，看到这个评论后进行了修改，终于启动成功了~

💨 activemq_msgs

ACTIVEMQ_ ACKS表存储持久订阅的信息和最后一个持久订阅接收的消息ID。

• ID：自增的数据库主键
• CONTAINER：消息的Destination
• MSGID_PROD：消息发送者的主键
• MSG_SEQ：是发送消息的顺序，MSGID_PROD+MSG_ SEQ可以组成JMS的MessagelD
• EXPIRATION：消息的过期时间，存储的是从1970-01-01到现在的毫秒数
• MSG：消息本体的Java序列化对象的二进制数据
• PRIORITY：优先级，从0-9， 数值越大优先级越高

💨 activemq_acks

ACTIVEMQ_ ACKS表用于存储订阅关系，存储持久订阅的信息和最后一个持久订阅接收的消息ID。如果是持久化Topic，订阅者和服务器的订阅关系在这个表保存。数据库字段如下:

• CONTAINER：消息的Destination
• SUB_DEST：如果是使用Static集群，这个字段会有集群其他系统的信息
• CLIENT_ID：每个订阅者都必须有一个唯.-的客户端ID用以区分
• SUB_NAME：订阅者名称
• SELECTOR：选择器，可以选择只消费满足条件的消息。条件可以用自定义属性实现，可支持多属性AND和OR操作。
• LAST_ACKED_ID：记录消费过的消息的ID。

💨 activemq_lock

ACTIVEMQ_LOCK表在集群环境中才有用，只有一个Broker可以获得消息，称为Master Broker，其他的只能作为备份等待Master Broker不可用，才可能成为下一个Master Broker。这个表用于记录哪个Broker是当前的Master Broker。

🔹 步骤6：代码运行验证

注意，要开启持久化

messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);

💨 队列生产者

import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;public class JmsProduce {public static final String ACTIVEMQ_URL = "tcp://192.168.150.101:61616";public static final String QUEUE_NAME = "jdbc01";public static void main(String[] args) throws JMSException {// 1 创建连接工厂,按照给定的url地址，采用默认的用户名和密码ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);// 2 通过连接工厂，获得连接connection并启动访问Connection connection = activeMQConnectionFactory.createConnection();connection.start();// 3 创建会话session// 参数1：事务// 参数2：签收Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);// 4 创建目的地（队列还是主题,这里选择队列）Queue queue = session.createQueue(QUEUE_NAME);// 5 创建消息的生产者MessageProducer messageProducer = session.createProducer(queue);messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);// 6 通过使用messageProducer生产3条消息发送到MQ队列中for (int i = 0; i < 3; i++) {// 7 创建消息 按照要求写好的消息TextMessage textMessage = session.createTextMessage("jdbc msg----" + i);//一个字符串// 8 通过messageProducer发给mqmessageProducer.send(textMessage);}// 9 关闭资源messageProducer.close();session.close();connection.close();System.out.println("---------消息发布到MQ完成--------");}
}

MQ服务器控制台：

mysql数据库ACTIVEMQ_MSGS表：

💨 队列消费者

import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;
import java.io.IOException;public class JmsConsumer {public static final String ACTIVEMQ_URL = "tcp://192.168.150.101:61616";public static final String QUEUE_NAME = "jdbc01";public static void main(String[] args) throws JMSException, IOException {// 1 创建连接工厂,按照给定的url地址，采用默认的用户名和密码ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);// 2 通过连接工厂，获得连接connection并启动访问Connection connection = activeMQConnectionFactory.createConnection();connection.start();// 3 创建会话session// 参数1：事务// 参数2：签收Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);// 4 创建目的地（队列还是主题,这里选择队列）Queue queue = session.createQueue(QUEUE_NAME);// 5 创建消息的消费者MessageConsumer messageConsumer = session.createConsumer(queue);// 6 处理消息/*方式1：同步阻塞方式(receive())订阅者或接收者调用MessageConsumer的receive()方法来接收消息，receive方法在能够接收到消息之前（或超时之前）将一直堵塞。receive(xxxx),如果添加参数表示等待一定时间，超过时间将关闭。*///        while (true) {
//            TextMessage textMessage = (TextMessage) messageConsumer.receive();
//            if (textMessage!=null) {
//                System.out.println("-------消费者接收到消息： " + textMessage.getText());
//            } else {
//                break;
//            }
//        }/*方式2： 异步非阻塞方式（监听器onMessage()）订阅者或接收者通过MessageConsumer的setMessageListener(MessageListener listener)注册一个消息监听器，当消息到达后，系统自动调用监听器MessageListener的onMessage(Message message)方法。*/messageConsumer.setMessageListener(new MessageListener() {@Overridepublic void onMessage(Message message) {if (message != null && message instanceof  TextMessage) {TextMessage textMessage = (TextMessage) message;try {System.out.println("-------消费者接收到消息： " + textMessage.getText());} catch (JMSException e) {e.printStackTrace();}}}});System.in.read();  // 保证控制台不灭，消费者启动之后将一直等待。如果关闭的话，消费者启动后还没监听到就立刻关闭了。// 7 资源关闭messageConsumer.close();session.close();connection.close();}
}

MQ服务器控制台：

mysql数据库ACTIVEMQ_MSGS表，消息被消费后，数据清空:

在点对点类型中：

当DeliveryMode设置为NON_PERSISTENCE时，消息被保存在内存中，不会保存在数据库中;
当DeliveryMode设置为PERSISTENCE时，消息保存在broker的相应的文件或者数据库中。
而且点对点类型中消息一旦被Consumer消费就从broker中删除。

💨 主题消费者

先启动消费者

import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;
import java.io.IOException;public class JmsConsumerTopicPersist {public static final String ACTIVEMQ_URL = "tcp://192.168.150.101:61616";public static final String TOPIC_NAME = "Topic-jdbc";public static void main(String[] args) throws JMSException, IOException {System.out.println("customer z3");ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);Connection connection = activeMQConnectionFactory.createConnection();connection.setClientID("z3");connection.start();Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);Topic topic = session.createTopic(TOPIC_NAME);TopicSubscriber topicSubscriber = session.createDurableSubscriber(topic, "mq-jdbc");connection.start();Message message = topicSubscriber.receive();while (message != null) {TextMessage textMessage = (TextMessage) message;System.out.println("------收到的持久Topic：" + textMessage.getText());message = topicSubscriber.receive(1000L);}session.close();connection.close();}
}

MQ服务器控制台：

查看数据库activemq_acks：

💨 主题生产者

import org.apache.activemq.ActiveMQConnectionFactory;import javax.jms.*;public class JmsProduceTopicPersist {public static final String ACTIVEMQ_URL = "tcp://192.168.150.101:61616";public static final String TOPIC_NAME = "Topic-jdbc";public static void main(String[] args) throws JMSException {ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);Connection connection = activeMQConnectionFactory.createConnection();Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);Topic topic = session.createTopic(TOPIC_NAME);MessageProducer messageProducer = session.createProducer(topic);messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);connection.start();for (int i = 0; i < 3; i++) {TextMessage textMessage = session.createTextMessage("TOPIC_NAME----" + i);         //一个字符串messageProducer.send(textMessage);}// 9 关闭资源messageProducer.close();session.close();connection.close();System.out.println("---------消息发布到MQ完成--------");}
}

MQ服务器控制台，有3条消息，并且都已经被处理了。

处理完消息后，消费者程序结束，离线了。

查看数据库activemq_acks：

🔹 小总结

queue：在没有消费者消费的情况下会将消息保存到activemq_msgs表中，只要有任意一一个消费者已经消费过了,消费之后这些消息将会立即被删除。

topic：一般是先启动消费订阅然后再生产的情况下会将订阅者保存到activemq_acks。

四、JDBC with Journal

这种方式克服了JQBC Store的不足，JDBC每次消息过来，都需要去写库和读库。

ActiveMQ Journal, 使用高速缓存写入技术，大大提高了性能。

当消费者的消费速度能够及时跟上生产者消息的生产速度时，journal文件能够大大减少需要写入到DB中的消息。

举个例子，生产者生产了1000条消息，这1000条消息会保存到journal文件， 如果消费者的消费速度很快的情况下，在journal文件还没有同步到DB之前，消费者已经消费了90%的以上的消息，那么这个时候只需要同步剩余的10%的消息到DB。如果消费者的消费速度很慢，这个时候journal文件可以使消息以批量方式写到DB。

🔹 修改配置文件

注释掉原来的：

<persistenceAdapter><jdbcPersistenceAdapter dataSource="#mysql-ds"/>
</persistenceAdapter>

换成：

<persistenceFactory><journalPersistenceAdapterFactory journalLogFiles="4"journalLogFileSize="32768"useJournal="true"useQuickJournal="true"dataSource="#mysql-ds"dataDirectory="activemq-data"/>
</persistenceFactory>

重启activemq

🔹 测试

首先启动上一章节的队列生产者，可以在MQ控制台查看到新增3条消息，但是在mysql数据库的activemq_msgs表格内没有任何数据。此时，启动队列消费者，刚刚新增的3条消息竟被处理，数据库同样没有任何变化。

五、总结

🔹 持久化消息主要指

MQ所在的服务器down了，消息不会丢失的机制。

🔹 持久化机制演化过程

从最初的AMQ Message Store 方案到ActiveMQ V4版本中推出的High performance journal (高性能事务支持)附件，并且同步推出了关于关系型数据库的存储方案。ActiveMQ 5.3版本中又推出了对KahaDB的支持(V5 4版本后称为ActiveMQ默认的持久化方案)，后来ActiveMQ V5.8版本开始支持LevelDB，到现在，V5.9+版本提供了标准的Zookeeper+LevelDB集群化方案。我们重点介绍了KahaDB、LevelDB 和mysq|数据库这三种持久化存储方案。

🔹 ActiveMQ的消息持久化机制

• AMQ：基于日志文件
• KahaDB：基于日志文件，从ActiveMQ 5.4开始默认的持久化插件
• JDBC：基于第3方数据库
• LevelDB：基于文件的本地数据库储存，从ActiveMQ 5.8版本之后又推出了LeveIDB的持久化引擎性能高于KahaDB
• Replicated LevelDB Store：从ActiveMQ 5.9提供了基于LevelDB和Zookeeper的数据复制方式，用于Master-slave方式的首选数据复制方案。

🔹 消息的存储逻辑一致

就是在发送者将消息发送出去后，消息中心首先将消息存储到本地数据文件、内存数据库或者远程数据库等，然后试图将消息发送给接收者，发送成功则将消息从存储中删除失败则继续尝试。消息中心启动以后首先要检查指定的存储位置，如果有未发送成功的消息，则需要把消息发送出去。