【ActiveMQ】之ActiveMQ 的持久化机制

Bertha 。 2023-02-28 01:56 10阅读 0赞

## MQ 的高可用实现为： ##

*  MQ 自带的 **事务、持久、签收**
 *  第三方可持久化机制

### 为什么要持久化呢？ ###

--------------------

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

ActiveMQ 的消息持久化机制有 **AMQ、KahaDB、JDBC 和 LevelDB**，无论使用哪一种持久化方式，消息的存储逻辑都是一致的：

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

下面简单介绍一下 ActiveMQ 的主要持久化机制

### AMQ Message Store： ###

--------------------

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

#### KahaDB： ####

基于日志文件，从ActiveMQ5.4开始默认的持久化插件，我们从配置文件 activemq.xml 中可以看到：

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

从中我们可以看到默认使用的是 kahaDB 并且存储的数据文件在 activemq 跟目录下的 data/kahadb 目录中，我们进去看一下这个目录：

-rw-r--r-- 1 root 197121 33554432 7月  18 17:59 db-1.log
    -rw-r--r-- 1 root 197121   147456 7月  18 23:23 db.data
    -rw-r--r-- 1 root 197121    49240 7月  18 23:23 db.redo
    -rw-r--r-- 1 root 197121       22 7月  18 23:23 db.free
    -rw-r--r-- 1 root 197121        8 7月  18 17:57 lock

其中：

*  db-<number>.log 为 kahaDB 存储消息的数据记录文件，文件大小限定32M，大于32M就会递增下一个数据文件，比如：log-2.log、log-3.log 以此类推。当不再引用数据文件中的任何消息时，文件就会被删除或归档。
 *  db.data 文件包含了持久化的 BTree 索引，是消息的索引文件，本质上是 B-Tree 树，使用 B-Tree 作为索引指向 db-<number>.log 里面存储的消息。
 *  db.free 记录 db.data 文件中哪些页面是空闲的，后面建索引则优先从空闲页中创建。
 *  db.redo 用来进行消息恢复
 *  lock 文件为文件锁，表示当前获得 kahaDB 读写权限的 broker

需要记住的就是 db-<number>.log 文件相等于 mysql 中的 MYD 数据文件，而 db.data 相当于 mysql 中的 MYI 索引文件。

#### LevelDB： ####

从 ActiveMQ 5.8 之后引进的，和 KahaDB 非常相似，也是基于文件的本地数据存储形式，但是它提供比 KahaDB 更快的持久性。

但它不适用自定义 B-Tree 实现来索引预写日志，而是使用基于 LevelDB 的索引。要使用 LevelDB 需要修改配置文件为：

#### JDBC ####

JDBC 消息存储是我们需要重点掌握的，包括后面的 JDBC Message Store with ActiveMQ Journal，一个比较更加快速的 JDBC 消息存储机制。

JDBC 存储必然是要涉及到数据库，我们以 MySQL 数据库为例，想要操作 mysql 数据库，必然要用到 mysql 数据的驱动包，所以我们需要把驱动包 mysql-connector-java-5.1.44.jar 放到 ActiveMQ 根目录下的 lib 目录下。

然后修改配置文件，把默认的 kahaDB 修改为 JDBC ：

这里有两个参数需要留意：**createTablesOnStartup** 表示是否启动的赌场自创建数据表，默认值是 true，这样每次启动都会去创建数据表，一般是第一次启动的时候设置为 true，之后改为 false。

**dataSource** 参数指定要引用的持久化数据库的 bean 名称，这里我们引用的名称为 mysql-ds，所以我们需要配置一个ID为mysql-ds的 bean，根据官网的例子我们在 <\\broker> 标签和 <import resource=“jetty.xml”/> 标签之间添加：

这是给出的实例，当然里面的参数需要自己进行设置，比如 url、username、password 等等。

这里需要注意的是，例子中使用的是 dbcp2 数据源连接池，因为它是本身自带的，如果想要使用其它例如 c3p0、druid 之类的，就不止需要导入 mysql 驱动包了，还需要导入对应的第三方驱动包了，这点事需要注意的。

从我配置的 bean 的 url 可以看出，我们需要连接的是 activemq 这个数据库，所以，我们需要去数据库中创建这个数据库：

CREATE DATABASE `activemq`;

创建好数据库并配置好配置文件之后，当我们重启 ActiveMQ 之后就会自动为我们创建三张表（第一次创建好后记得把参数 createTablesOnStartup 设置为 false，否则下次重启会创建覆盖原来的表）：

**activemq\_msgs**： 消息表，queue 和 topic 都存在里面。表结构如下

DESC activemq_msgs;
    
    Field       Type          Null    Key     Default  Extra   
    ---------- ------------ ------ ------ ------- --------
    ID          bigint(20)    NO      PRI     (NULL)           
    CONTAINER   varchar(250)  NO      MUL     (NULL)   # 消息的 destination
    MSGID_PROD  varchar(250)  YES     MUL     (NULL)   # 消息发送者的主键
    MSGID_SEQ   bigint(20)    YES             (NULL)   # 发送消息的顺序
    EXPIRATION  bigint(20)    YES     MUL     (NULL)   # 消息的过期时间
    MSG         longblob      YES             (NULL)   # 消息本体的 Java 序列化对象的二进制数据
    PRIORITY    bigint(20)    YES     MUL     (NULL)           
    XID         varchar(250)  YES     MUL     (NULL)

**activemq\_acks**：存储持久订阅的信息和最后一个持久订阅接收的消息ID，表结构如下：

DESC activemq_acks;
    
    Field          Type          Null    Key     Default  Extra   
    ------------- ------------ ------ ------ ------- --------
    CONTAINER      varchar(250)  NO      PRI     (NULL)   # 消息的 destination
    SUB_DEST       varchar(250)  YES             (NULL)   # 记录集群的系统信息
    CLIENT_ID      varchar(250)  NO      PRI     (NULL)   # 订阅者的唯一客户端ID
    SUB_NAME       varchar(250)  NO      PRI     (NULL)   # 订阅者名称
    SELECTOR       varchar(250)  YES             (NULL)   # 选择器，可以选择之消费满足条件的消息
    LAST_ACKED_ID  bigint(20)    YES             (NULL)   # 记录消费国的消息的ID
    PRIORITY       bigint(20)    NO      PRI     5                
    XID            varchar(250)  YES     MUL     (NULL)

**activemq\_lock**：该表用于记录那个 broker 是当前的 master broker，在集群环境中才有用，只有一个 broker 可以获得消息，成为 master borker，其它的 broker 只能作为备份等待 master broker 不可用，才可能成为下一个 master broker。表的结构如下：

DESC activemq_lock;
    
    Field        Type          Null    Key     Default  Extra   
    ----------- ------------ ------ ------ ------- --------
    ID           bigint(20)    NO      PRI     (NULL)           
    TIME         bigint(20)    YES             (NULL)           
    BROKER_NAME  varchar(250)  YES             (NULL)   # 当前的 master broker

### 代码实战 ###

--------------------

下面我先以 queue 为例，想要消息保存到数据库生效，必须在创建消息生产者的时候开启持久化模式（消费者无需改写代码）：

MessageProducer messageProducer = session.createProducer(queue);
    messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);

这样，当生产者发送消息成功后就可以在数据库中的 activemq\_msgs 表看到相应的数据了。而当消费者消费消息之后，数据库的相应记录就会被删除。

ID  CONTAINER         MSGID_PROD                                      MSGID_SEQ  EXPIRATION
    ------ ---------------- ---------------------------------------------- --------- ----------
    	1  queue://my-queue ID:DESKTOP-5VJT1GH-63743-1595138598000-1:1:1:1 1 0
    	2  queue://my-queue ID:DESKTOP-5VJT1GH-63743-1595138598000-1:1:1:1 2 0
    	3  queue://my-queue ID:DESKTOP-5VJT1GH-63743-1595138598000-1:1:1:1 3 0

**注意，如果没有开启持久化（setDeliveryMode(DeliveryMode.PERSISTENT)）则不会把消息存储到数据库中。**

而对于 topic 而言，数据的存储机制有点不同，它不仅会把消息存储到 activemq\_msgs 表中，而且还会把所有订阅者都记录到 activemq\_acks 表中，同时这些**数据是不会随着消息被消费而删除的**，也就是说会一直保存下去。

activemq_msgs 表
    
    	ID  CONTAINER         MSGID_PROD                                      MSGID_SEQ  EXPIRATION
    ------ ---------------- ---------------------------------------------- --------- ----------
         1  topic://my-topic ID:DESKTOP-5VJT1GH-63696-1595138305695-1:1:1:1 1 0
         2  topic://my-topic ID:DESKTOP-5VJT1GH-63696-1595138305695-1:1:1:1 2 0
         3  topic://my-topic ID:DESKTOP-5VJT1GH-63696-1595138305695-1:1:1:1 3 0
    
    activemq_acks 表
    
    CONTAINER         SUB_DEST          CLIENT_ID  SUB_NAME   SELECTOR  LAST_ACKED_ID  PRIORITY  XID     
    ---------------- ---------------- --------- --------- -------- ------------- -------- --------
    topic://my-topic topic://my-topic wang remark... (NULL) 18 0 (NULL)

#### JDBC Message Store with ActiveMQ Journal 带高速缓存的 JDBC ####

这种方式克服了 JDBC 存储的不足：JDBC 每次消息过来，都要去读写数据库，比较消耗性能，所以引进了 ActiveMQ Journal，使用高速缓存写入技术，大大提高了性能，

要使用 ActiveMQ Journa 的话，需要修改配置文件，把持久化工厂修改为如下：

运行原理为：当消费者的消费速断能够及时更上生产者消息的生产速度时，journal 文件能够大大减少需要写入到 DB 中的消息，比如：当消费者的消费速度很快的情况下，在 journal 文件还没有同步到 DB之前，消费者已经消费了 90% 的以上消息时，此时只需同步剩余的 10% 的消息到 DB 中。如果消费者消费速度很慢，这个时候 journal 文件可以使消息以批量方式写到 DB 中。

**关于 ActiveMQ 的更多持久化机制可以参考官网的文档：[http://activemq.apache.org/persistence][http_activemq.apache.org_persistence]**

[http_activemq.apache.org_persistence]: http://activemq.apache.org/persistence