ActiveMQ系列：ActiveMQ的持久化机制-阿里云开发者社区

前面有介绍与基础、应用实践部分，有兴趣的可以移步：

初步认识了ActiveMQ：https://blog.csdn.net/qq_26975307/article/details/98875098

结合JavaSE进行初尝试：https://blog.csdn.net/qq_26975307/article/details/98968854

详细讲讲JMS：https://blog.csdn.net/qq_26975307/article/details/99408962

JMS的可靠性：https://phubing.blog.csdn.net/article/details/99412285

结合 Spring，基于配置文件的使用 ActiveMQ：https://phubing.blog.csdn.net/article/details/99413883

结合 SpringBoot，基于 application.xml 使用ActiveMQ：https://blog.csdn.net/qq_26975307/article/details/99415899

ActiveMQ传输协议：https://blog.csdn.net/qq_26975307/article/details/100147542

此篇开始探寻 ActiveMQ 的持久化机制，有兴趣可以留言共同探讨

1、官网拜读

有兴趣的可以到官网拜读拜读，毕竟官网才是真理：http://activemq.apache.org/persistence

2、是什么

2.1、当然是关注下面试题先

ActiveMQ 的持久化机制？MQ 服务器 down 机了，消息不会丢失的机制（类比下 Redis 的持久化机制？）

2.2、持久化说明

为了避免意外宕机以后丢失信息，需要做到重启后可以恢复消息队列，消息系统一般都会采用持久化机制。

ActiveMQ的消息持久化机制有JDBC，AMQ，KahaDB和LevelDB，无论使用哪种持久化方式，消息的存储逻辑都是一致的就是在发送者将消息发送出去后，消息中心首先将消息存储到本地数据文件、内存数据库或者远程数据库等再试图将消息发送给接收者，成功则将消息从存储中删除，失败则继续尝试发送。

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

2.3、ActiveMQ 的持久化机制有哪些？

2.3.1、AMQ Message Store（没什么卵用，不信你看官网）

基于文件的存储方式，是以前的默认消息存储，现在不用了

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

AMQ适用于ActiveMQ5.3之前的版本（所以这玩意没卵用了）

2.3.2、KahaDB消息存储（基于日志文件，从ActiveMQ5.4开始默认的持久化插件）

2.3.2.1、说明

看官网去：http://activemq.apache.org/kahadb

KahaDB is a file based persistence database that is local to the message broker that is using it. It has been optimized for fast persistence. It is the the default storage mechanism since ActiveMQ 5.4. KahaDB uses less file descriptors and provides faster recovery than its predecessor, the AMQ Message Store.

帮你翻译下：

KahaDB是一个基于文件的持久性数据库，它位于使用它的消息代理的本地。它已经为快速持久性进行了优化。

它是ActiveMQ 5.4以来的默认存储机制。KahaDB使用更少的文件描述符，并且比它的前身AMQ消息存储提供更快的恢复

（上面的所述文件夹在：apache-activemq-5.15.9/data/ 下）

KahaDB 是目前默认的存储方式，可用于任何场景，提高了性能和恢复能力。

消息存储使用一个事务日志和仅仅用一个索引文件来存储它所有的地址。

KahaDB 是一个专门针对消息持久化的解决方案，它对典型的消息使用模式进行了优化。

数据被追加到 data logs 中。当不再需要 log 文件中的数据的时候，log 文件会被丢弃。

ActiveMQ传输协议支持的：http://activemq.apache.org/configuring-version-5-transports.html

配置参考文档：activemq.apache.org/tcp-transport-reference

auto关键字：activemq.apache.org/auto

<transportConnector name="auto+nio" uri="auto+nio://0.0.0.0:61608?maximumConnections=1000&amp;wireFormat.maxFrameSize=104857600&amp;org.apache.activemq.transport.nio.SelectorManager.corePoolSize=20&amp;org.apache.activemq.transport.nio.SelectorManager.maximumPoolSize=50"/>

2.3.2.2、KahaDB 目录简要说明

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

1、db-.log

KahaDB 存储消息到预定义大小的数据记录文件中，文件命名为 db-.log 。当数据文件已满时，一个新的文件会随之创建，number数值也会随之递增，它随着消息数量的增多，如每32M一个文件，文件名按照数字进行编号，如

db-1.log、db-2.log、db-3.log …。

当不再有引用到数据文件中的任何消息时，文件会被删除或归档。

2、db.data 该文件包含了持久化的 BTree 索引，索引了消息数据记录中的消息，它是消息的索引文件，本质上是 B-Tree（B树），使用B-Tree作为索引指向db-.log里面存储的消息。（不要问我 B-Tree 原理，我也不知道）

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

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

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

2.3.3、JDBC消息存储（消息基于JDBC存储的）

2.3.3、LevelDB消息存储（这玩意大佬们还没决定，不信你再去看官网，看看这个东西的更新说明）

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

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

默认配置如下：

<persistenceAdapter>
    <levelDBdirectory="activemq-data"/>
</persistenceAdapter>

In ActiveMQ 5.8, the LevelDB Store was introduced. The LevelDB Store is a file based persistence database. It has been optimized to provide even faster persistence than KahaDB. Although not yet the default message store, we expect this store implementation become the default in future releases.

2.3.3、JDBC Message store with ActiveMQ Journal

To enable JDBC persistence of JMS messages without journaling, we need to change the message broker's default persistence configuration from AMQ4.x

<persistenceAdapter>
    <journaledJDBC journalLogFiles="5"dataDirectory="../activemq-data"/>
</persistenceAdapter>

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

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

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

举个栗子

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

怎么使用呢？

在/myactiveMQ/apache-activemq-5.15.9/conf路径下修改activemq.xml配置文件，按照如下修改：

修改前：

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

修改后：

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

以前是已启动生产者并发送，消息就存在于ActiveMQ——msgs表中，百分之百能够直接在表中直接看到。

因为目前链接数少，链接、验证、取数据、存数据几乎都是纳秒级别。

但现在增加了告诉缓存之后，一时间是看不到msgs的表数据的，因为是MQ生产者先跟缓存打交道，把数据先写到缓存中，之后再从缓存慢慢的同步到数据库中。

先代码看结果（journal日志高速缓存了数据）

能否进行消费？完全没得问题。

删也是从高速缓存中删除，过一会，数据库中的数据才会被删除，也即实现了读写分离的思想

生产者生产后，MQ中肯定是所见即所得，然后写进了高速缓存，此时高速缓存正在等候着，如果一直不消费则会写进DB

2.4、ActiveMQ 的 JDBC 消息存储

1、MQ + MySQL 的设置

2、添加 mysql 数据库的驱动包到 lib 文件夹

3、jdbcPersistenceAdapter 的配置

4、数据库连接池配置

5、建仓 SQL 和建表说明

6、代码运行验证

7、数据库情况

8、小总结

9、开发中可能遇到的坑

（先上图镇楼）

2.5、JDBC持久化存储 MQ 的设置步骤详解

2.5.1、jdbcPersistenceAdapter 配置

在 /apache-activemq-5.15.9/conf 路径下修改 activemq.xml 配置文件，按照如下修改：

修改前kahaDB

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

修改后 jdbcPersistenceAdapter

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

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

2.5.2、mysql-ds 的配置（要在之外）

<bean id="mysql-ds" class="org.apache.commons.dbcp2.BasicDataSource" destroy-method="close">
        <property name="driverClassName" value="com.mysql.cj.jdbc.Driver"/>
        <property name="url" value="jdbc:mysql://ip:端口/数据库名?useUnicode=true&amp;characterEncoding=UTF-8&amp;useSSL=false&amp;autoReconnect=true&amp;failOverReadOnly=false&amp;serverTimezone=UTC&amp;relaxAutocommit=true" />
        <property name="username" value="数据库连接名"/>
        <property name="password" value="数据库连接密码"/>
        <property name="maxTotal" value="200"/>
        <property name="poolPreparedStatements" value="true"/>
    </bean>

2.5.3、JDBC配置SQL

1、建一个数据库，名字你喜欢呗

2、如果新建数据库 OK +上述配置 OK + 代码运行 OK ，3 个表会自动生成

3、万一配置出错了的情况，手动建表SQL（如果配置好不需要手动，应急）

2.5.4、ACTIVEMQ_MSGS 表结构说明及建表语句

消息表，缺省表名为 ACTIVEMQ_MSGS，queue 和 topic 都存在里面，结构如下：

ID：自增的数据库主键

CONTAINER：消息的Destination

MSGID_PROD：消息发送者的主键

MSG_SEQ：是发送消息的顺序，MSGID_PROD+MSG_SEQ可以组成JMS的MessagelD

EXPIRATION：消息的过期时间，存储的是从1970-01-01到现在的毫秒数

MSG：消息本体的Java序列化对象的二进制数据

PRIORITY：优先级，从0-9，数值越大优先级越

SET NAMES utf8mb4;
SET FOREIGN_KEY_CHECKS = 0;
-- ----------------------------
-- Table structure for activemq_msgs
-- ----------------------------
DROP TABLE IF EXISTS `activemq_msgs`;
CREATE TABLE `activemq_msgs`  (
  `ID` bigint(20) NOT NULL,
  `CONTAINER` varchar(250) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
  `MSGID_PROD` varchar(250) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT NULL,
  `MSGID_SEQ` bigint(20) NULL DEFAULT NULL,
  `EXPIRATION` bigint(20) NULL DEFAULT NULL,
  `MSG` blob NULL,
  `PRIORITY` bigint(20) NULL DEFAULT NULL,
  `XID` varchar(250) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT NULL,
  PRIMARY KEY (`ID`) USING BTREE,
  INDEX `ACTIVEMQ_MSGS_MIDX`(`MSGID_PROD`, `MSGID_SEQ`) USING BTREE,
  INDEX `ACTIVEMQ_MSGS_CIDX`(`CONTAINER`) USING BTREE,
  INDEX `ACTIVEMQ_MSGS_EIDX`(`EXPIRATION`) USING BTREE,
  INDEX `ACTIVEMQ_MSGS_PIDX`(`PRIORITY`) USING BTREE,
  INDEX `ACTIVEMQ_MSGS_XIDX`(`XID`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci ROW_FORMAT = Dynamic;
SET FOREIGN_KEY_CHECKS = 1;

2.5.5、ACTIVEMQ_ACKS

ACTIVEMQ_ACKS 用于存储订阅关系。如果是持久化 Topic，订阅者和服务器的订阅关系在这个表保存，ACTIVEMQ_ACKS 表存储持久订阅的信息和最后一个持久订阅接收的消息 ID：

CONTAINER：消息的Destination

SUB_DEST：如果是使用Static集群，这个字段会有集群其他系统的信息

CLIENT_ID：每个订阅者都必须有一个唯一的客户端ID用以区分

SUB_NAME：订阅者名称

SELECTOR：选择器，可以选择只消费满足条件的消息。条件可以用自定义属性实现，可支持多属性AND和OR操作

LAST_ACKED_ID：记录消费过的消息的ID。

SET NAMES utf8mb4;
SET FOREIGN_KEY_CHECKS = 0;
-- ----------------------------
-- Table structure for activemq_acks
-- ----------------------------
DROP TABLE IF EXISTS `activemq_acks`;
CREATE TABLE `activemq_acks`  (
  `CONTAINER` varchar(250) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
  `SUB_DEST` varchar(250) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT NULL,
  `CLIENT_ID` varchar(250) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
  `SUB_NAME` varchar(250) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL,
  `SELECTOR` varchar(250) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT NULL,
  `LAST_ACKED_ID` bigint(20) NULL DEFAULT NULL,
  `PRIORITY` bigint(20) NOT NULL DEFAULT 5,
  `XID` varchar(250) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT NULL,
  PRIMARY KEY (`CONTAINER`, `CLIENT_ID`, `SUB_NAME`, `PRIORITY`) USING BTREE,
  INDEX `ACTIVEMQ_ACKS_XIDX`(`XID`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci ROW_FORMAT = Dynamic;
SET FOREIGN_KEY_CHECKS = 1;

2.5.6、ACTIVEMQ_LOCK

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

SET NAMES utf8mb4;
SET FOREIGN_KEY_CHECKS = 0;
-- ----------------------------
-- Table structure for activemq_lock
-- ----------------------------
DROP TABLE IF EXISTS `activemq_lock`;
CREATE TABLE `activemq_lock`  (
  `ID` bigint(20) NOT NULL,
  `TIME` bigint(20) NULL DEFAULT NULL,
  `BROKER_NAME` varchar(250) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT NULL,
  PRIMARY KEY (`ID`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci ROW_FORMAT = Dynamic;
-- ----------------------------
-- Records of activemq_lock
-- ----------------------------
INSERT INTO `activemq_lock` VALUES (1, NULL, NULL);
SET FOREIGN_KEY_CHECKS = 1;

2.5.6、开启持久化模式

jdbc 与 ActiveMQ 消息持久化的时候一定要开启持久化模式：

messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT)

我就不开，怎么地？不怎么地，存不到呗~

2.5.7、代码运行验证

producer

package com.phubing.jdbcAndMySQLAndActiveMQ;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class Producer {
    public static final String ACTIVEMQ_URL =  "tcp://192.168.177.130:61616";
    public static final String TOPIC_NAME = "topic-mysql-01";
    public static void main(String[] args) throws Exception{
        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 < 6; i++) {
            TextMessage message=session.createTextMessage("message--persistent-topic:"+i);
            messageProducer.send(message);
        }
        messageProducer.close();
        session.close();
        connection.close();
    }
}

consumer

package com.phubing.jdbcAndMySQLAndActiveMQ;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class Consumer {
    public static final String ACTIVEMQ_URL =  "tcp://192.168.177.130:61616";
    public static final String TOPIC_NAME = "topic-mysql-01";
    public static void main(String[] args) throws Exception{
        //1创建连接工场使用默认用户名密码，编码不再体现
        ActiveMQConnectionFactory activemoconnectionFactory=new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        //2获得连接并启动
        Connection connection =activemoconnectionFactory.createConnection();
        //2.1需要在连接上设置消费考id，用来识别消费者
        connection.setClientID("phubing-01");
        //3 创建会话，此步骤有两个参数，I第一个是否以事务的方式超交，第二个默认的签收方式
        Session session=connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        //4 创建主题
        Topic topic =session.createTopic(TOPIC_NAME);
        //4.1要创建建opicsubscriber来订阅
        TopicSubscriber topicsubscriber =session.createDurableSubscriber(topic, "mq-jdbc");
        //4.2一定要先运行一次，等于向消息服务中间件注法册这个消费者，然后再运行客户端发送信息，这个时候
        //无论消费者是否在线，都会接收到不在线的话，下次连接的时候，会把没有收过的消息都接收下来。
        connection.start();
        //5 接受消息
        Message message=topicsubscriber.receive();
        while (null != message){
            TextMessage textMessage = (TextMessage)message;
            System.out.println("收到的持久化Tpoic："+textMessage.getText());
            message = topicsubscriber.receive();
        }
        session.close();
        connection.close();
    }
}

分别运行看看 MySQL 中数据库表的变化？

2.5.8、点对点说明以及消费说明

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

如果是queue：

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

如果是topic：

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

数据库运行情况就不列举了，自行测试即可。

2.5.9、开发中可能遇到的坑（呵呵~）

数据库jar包：

记得需要使用到的相关 jar 文件放置到 ActiveMQ 安装路径下的 lib 目录。

mysql-jdbc 驱动的 jar 包和对应的数据库连接池 jar 包

createTablesOnStartup 属性

在 jdbcPersistenceAdapter 标签中设置了 createTablesOnStartup 属性为 true 时在第一次启动 ActiveMQ 时，ActiveMQ 服务节点会自动创建所需要的数据表。启动完成后可以去掉这个属性，或者更改 createTablesOnStartup 属性为 false。

下滑线坑爹

“java.lang.llegalStateException:BeanFactory not initialized or already closed”

这是因为您的操作系统的机器名中有“_”符号。请更改机器名并且重启后即可解决问题。

2.6、持久化机制不是总结的小总结

2.6.1、持久化消息主要是指：

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

2.6.2、持久化机制演化过程：

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

2.6.3、ActiveMQ的消息持久化机制有：

AMQ：基于日志文件

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

JDBC：基于第3方数据库

LevelDB：基于文件的本地数据库储存，从ActiveMQ5.8版本之后又推出了LeveDB的持久化引擎性能高于KahaDB

Replicated LevelDB Store ：从 ActiveMQ 5.9 提供了基于LevelDB和 Zookeeper 的数据复制方式，用于Master-slave方式的首选数据复制方案。

2.6.4、无论使用哪种持久化方式，消息的存储逻辑都是一致的：

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

(未完待续......下篇开始基于LevelDB和 Zookeeper 的数据复制集群 )

ActiveMQ系列：ActiveMQ的持久化机制