ActiveMQ消息持久化

Bertha 。 2023-05-28 06:58 18阅读 0赞

### 前言 ###

在前面我我们讲JMS规范的时候有简单的说过activeMQ的消息持久化，演示了如何来设置使消息能够持久化存储。

本次呢，我们将来深入了解activeMQ的消息持久化，看看底层的实现机制。

### 是什么？ ###

官方文档：[http://activemq.apache.org/persistence][http_activemq.apache.org_persistence]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70]  
此页介绍了activeMQ各个版本的持久化特性。

消息持久化就是为了避免意外宕机以后丢失信息，需要做到重启后可以恢复消息队列，消息系统一般都会采用持久化机制。ActiveMQ的消息持久化机制有`AMQ，KahaDB，JDBC和LevelDB`，无论使用哪种持久化方式，消息的存储逻辑都是一致的。就是在发送者将消息发送出去后，消息中心首先将消息存储到本地数据文件、内存数据库或者远程数据库等。再试图将消息发给接收者，成功则将消息从存储中删除，失败则继续尝试尝试发送。消息中心启动以后，要先检查指定的存储位置是否有未成功发送的消息，如果有，则会先把存储位置中的消息发出去。

简单理解就是：ActiveMQ宕机了，消息不会丢失的机制。

### 有哪些？ ###

*  AMQ Mesage Store：它是一种文件存储形式，它具有写入速度快和容易恢复的特点。消息存储再一个个文件中文件的默认大小为32M，当一个文件中的消息已经全部被消费，那么这个文件将被标识为可删除，在下一个清除阶段，这个文件被删除。适用于ActiveMQ5.3之前的版本。是以前的默认消息存储，现在不用了
 *  KahaDB：KahaDB是目前默认的存储方式，可用于任何场景，提高了性能和恢复能力。消息存储使用一个事务日志和仅仅用一个索引文件来存储它所有的地址。KahaDB是一个专门针对消息持久化的解决方案，它对典型的消息使用模型进行了优化。数据被追加到data logs中。当不再需要log文件中的数据的时候，log文件会被丢弃。
 *  JDBC：基础数据库的消息存储
 *  JDBC Journa：是一种日志+数据库的持久化机制，当生产者发送了消息，首先会放到日志中，等过一段时间，如果此消息还未被消费掉，才会将此消息入库。就相当于数据库前加了层redis，它解决了传统JDBC频繁读写所带来性能问题。
 *  LevelDB：这种文件系统是从ActiveMQ5.8之后引进的，它和KahaDB非常相似，也是基于文件的本地数据库存储形式，但是它提供比KahaDB更快的持久性。但它不使用自定义B-Tree实现来索引独写日志，而是使用基于LevelDB的索引。

### 怎么用？ ###

#### KahaDB ####

官方介绍：[http://activemq.apache.org/kahadb][http_activemq.apache.org_kahadb]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 1]  
`activemq.xml`配置：

还可以加一些其它参数，具体的在此处就不在演示了，可参考官方文档。

如果你的activeMQ是5.4版本或以上版本的，默认就是使用kahaDB方式。

消息的持久化的文件放在activeMQ的安装目录/data/kahadb下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 2]  
kahadb目录文件介绍：kahadb目录下一共有4种文件和一个lock。

*  db-number.log（db-1.log）：KahaDB存储消息到预定大小的数据纪录文件中，文件名为db-number.log。当数据文件已满时，一个新的文件会随之创建，number数值也会随之递增，它随着消息数量的增多，如没32M一个文件，文件名按照数字进行编号，如db-1.log，db-2.log······。当不再有引用到数据文件中的任何消息时，文件会被删除或者归档。
 *  db.data：该文件包含了持久化的BTree索引，索引了消息数据记录中的消息，它是消息的索引文件，本质上是B-Tree（B树），使用B-Tree作为索引指向db-number。log里面存储消息。
 *  db.free：当问当前db.data文件里面哪些页面是空闲的，文件具体内容是所有空闲页的ID
 *  db.redo：用来进行消息恢复，如果KahaDB消息存储再强制退出后启动，用于恢复BTree索引。
 *  lock：文件锁，表示当前kahadb独写权限的broker。

#### JDBC ####

首先我们需要有一个能够与activeMQ服务通信的数据库服务，并新建一个名称为activemq的数据库（也可以是其它名称）。

数据库可以是mysql也可以是Oracle，只不过是配置不同，以下示例将是在mysql上进行的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 3]  
`activemq.xml`配置：

<persistenceAdapter>  
          <jdbcPersistenceAdapter dataSource="#mysql-ds"> 
    </persistenceAdapter>
    
    <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://数据库地址/activemq?relaxAutoCommit=true"/> 
        <property name="username" value="数据库账号"/> 
        <property name="password" value="数据库密码"/> 
        <property name="poolPreparedStatements" value="true"/> 
     </bean>

根据自己的环境和个性化需要，可修改数据库的连接池、驱动、数据库地址、数据库账号、密码等配置。

注意数据库连接配置要放到`</broker>`标签和`<import>`标签之间：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 4]  
接着添加数据库的驱动包到activeMQ的lib目录下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 5]  
默认是的dbcp数据库连接池，如果要换成其他数据库连接池，也需要将该连接池jar包，放到lib目录下。

配置好后，启动activeMQ服务，会自动创建表，可以在`jdbcPersistenceAdapter`标签中增加`createTableOnStartup="false"`配置，设置为不自动创建，默认是自动创建的，一般情况下我们会在activeMQ首次启动表创建好之后将配置改为不自动创建。

启动好之后就可以看到创建好的三张表：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 6]  
activemq\_acks：存储订阅关系，如果是持久化topic，订阅者和服务器的订阅关系会在这个表保存，字段如下：

<table> 
 <thead> 
  <tr> 
   <th>字段名</th> 
   <th>数据类型</th> 
   <th align="left">描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>CONTAINER</td> 
   <td>varchar(250)</td> 
   <td align="left">消息的Destination</td> 
  </tr> 
  <tr> 
   <td>SUB_DEST</td> 
   <td>varchar(250)</td> 
   <td align="left">如果是使用static集群，这个字段会存储其它集群的信息</td> 
  </tr> 
  <tr> 
   <td>CLIENT_ID</td> 
   <td>varchar(250)</td> 
   <td align="left">自己定义的订阅者的唯一id</td> 
  </tr> 
  <tr> 
   <td>SUB_NAME</td> 
   <td>varchar(250)</td> 
   <td align="left">订阅者的名称</td> 
  </tr> 
  <tr> 
   <td>SELECTOR</td> 
   <td>varchar(250)</td> 
   <td align="left">选择器，可以选择只消费满足条件的消息。条件可以用自定义属性实现，可支持多属性AND和OR操作</td> 
  </tr> 
  <tr> 
   <td>LAST_ACKED_ID</td> 
   <td>bigint(20)</td> 
   <td align="left">最后一次确认ID,这个字段存的该该订阅者最后一次收到的消息的ID</td> 
  </tr> 
  <tr> 
   <td>PRIORITY</td> 
   <td>bigint(20)</td> 
   <td align="left">消息优先级</td> 
  </tr> 
 </tbody> 
</table>

![在这里插入图片描述][20200328161623833.png]

activemq\_lock：在集群环境下使用，用于记录哪个broker是当前的master

activemq\_msgs：消息表，queue和topic消息都存在此表里面

<table> 
 <thead> 
  <tr> 
   <th>字段名</th> 
   <th>数据类型</th> 
   <th align="left">描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>ID</td> 
   <td>bigint(20)</td> 
   <td align="left">主键</td> 
  </tr> 
  <tr> 
   <td>CONTAINER</td> 
   <td>varchar(250)</td> 
   <td align="left">消息的Destination</td> 
  </tr> 
  <tr> 
   <td>MSGID_PROD</td> 
   <td>varchar(250)</td> 
   <td align="left">消息id，由activeMQ生成</td> 
  </tr> 
  <tr> 
   <td>MSGID_SEQ</td> 
   <td>bigint(20)</td> 
   <td align="left">消息顺序</td> 
  </tr> 
  <tr> 
   <td>EXPIRATION</td> 
   <td>bigint(20)</td> 
   <td align="left">消息的过期时间</td> 
  </tr> 
  <tr> 
   <td>MSG</td> 
   <td>longblob</td> 
   <td align="left">消息的二进制数据</td> 
  </tr> 
  <tr> 
   <td>PRIORITY</td> 
   <td>bigint(20)</td> 
   <td align="left">消息优先级</td> 
  </tr> 
 </tbody> 
</table>

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 7]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 8]  
发送queue消息：

*  当DeliveryMode设置为NON\_PERSISTENCE时，消息被保存在内存中
 *  当DeliveryMode设置为PERSISTENCE时，消息保存在activemq\_msgs表中
 *  消息一旦被Consumer消费，就会从activemq\_msgs表中删除

发送topic消息：

*  先启动消费订阅者然后再生产的情况下会将持久订阅者永久保存到activemq\_acks
 *  生产者发送的topic消息永久保存在activemq\_msgs
 *  在activemq\_acks表中的订阅者有一个last\_ack\_id对应了activemq\_msgs中的id字段，这样就知道订阅者最后收到的消息是哪一条。
 *  持久化topic的消息不管是否被消费，是否有消费者，产生的数据永远都存在，且activemq\_msgs种只存储一条，消费者消费完后，消息还会保留

#### JDBC Journa ####

JDBC Journa是为了解决JDBC Store的不足，JDBC的方式，每次消息过来，都要对数据库进行读写。ActiveMQ Journal，使用高速缓存写入技术，大大提高了性能。当消费者的速度能够及时跟上生产者消息的生产速度时，journal文件能够大大减少需要写入到DB中的消息。

举个例子：生产者生产了1000条消息，这1000条消息会保存到journal文件，如果消费者的消费速度很快的情况下，在journal文件还没有同步到DB之前，消费者已经消费了90%的以上消息，那么这个时候只需要同步剩余的10%的消息到DB。如果消费者的速度很慢，这个时候journal文件可以使消息以批量方式写到DB。  
为了高性能，这种方式使用日志文件存储+数据库存储。先将消息持久到日志文件，等待一段时间再将未消费的消息持久到数据库。该方式要比JDBC性能要高。

`activemq.xml`配置：

[http_activemq.apache.org_persistence]: http://activemq.apache.org/persistence
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70]: /images/20230528/8c527fc6bd2c41209ff43825dabbec12.png
[http_activemq.apache.org_kahadb]: http://activemq.apache.org/kahadb
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 1]: /images/20230528/25ce97386187497581f4e69f73a3b9f0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 2]: /images/20230528/5508b7d8701944a68e3dd6ba5b3ab937.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 3]: /images/20230528/2c49a5d55fd642d0b94de5d89e058876.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 4]: /images/20230528/f9c7b7145e3d4cce85fb70e07c42213d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 5]: /images/20230528/762830aac6864e3ab5587434762be084.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 6]: /images/20230528/3d6dba3e5b81456a8a77be5c1013a1f9.png
[20200328161623833.png]: /images/20230528/490ea53151d94e279fda438d17222156.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 7]: /images/20230528/cac6fb35153640daba473217f42511d7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NoZW5fY2hhbmd5aW5n_size_16_color_FFFFFF_t_70 8]: /images/20230528/24b3d0f1cbb841719433d02c8223f9e4.png