Java 消息中间件 - ActiveMQ之消息存储和持久化_java 订阅关系保存在哪里

一、说明

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

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

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

二、有哪些持久化方式？

1、AMQ Message Store(了解)

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

2、LevelDB消息存储(了解)

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

<persistenceAdapter>
    <levelDB directory="activemq-data"/>
</persistenceAdapter>
1
2
3

3、KahaDB(默认)

基于日志文件，从ActiveMQ5.4开始默认的持久化插件。如图，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
        -->
         <!--directory:保存数据的目录;journalMaxFileLength:保存消息的文件大小-->
        <persistenceAdapter>
            <kahaDB directory="${activemq.data}/kahadb" journalMaxFileLength="32MB"/>
        </persistenceAdapter>

1
2
3
4
5
6
7
8
9
10
11
12

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

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

备注：正常情况kahadb文件夹中有5个文件，这个少了 db.free。

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

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

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

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

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

4、JDBC消息存储

1、添加mysql数据库的驱动包到lib文件夹
命令：cp mysql-connector-java-5.1.48.jar /myactiveMQ/apache-activemq-5.15.11/lib/

2、jdbcPersistenceAdapter配置
在 /myactiveMQ/apache-activemq-5.15.11/conf 路径下修改activemq.xml,将原先默认的kahaDB持久化配置替换为 jdbcPersistenceAdapter配置，即：

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

改为

<persistenceAdapter> 
     <jdbcPersistenceAdapter dataSource="#my-ds"/> 
</persistenceAdapter>
1
2
3

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

3、数据库连接池配置
即将连接池配置放到 activemq.xml 文件的 </ broker> 与 < import resource=“jetty.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/activemq?relaxAutoCommit=true"/> 
    <property name="username" value="root"/> 
    <property name="password" value="root"/> 
    <property name="poolPreparedStatements" value="true"/> 
</bean> 
1
2
3
4
5
6
7

说明：
1）这里用的是官网指定的数据库驱动（dbcp2），所以在步骤一只导入了 mysql-connector-java-5.1.48.jar 驱动包，若是公司用其他的数据库驱动，如c3p0等，则还需要导入对应的驱动包；
2）bean中的 id值（my-ds） 要与 上一步配置中的dataSource的值一致，否则activemq无法启动成功。
3）因为我连的是本地数据库，所以连接地址为 127.0.1，其默认端口号是 3306

4、建一个名为activemq的数据库

三张表说明：
1）ACTIVEMQ_MSGS ：用于存储消息，Queue和Topic都存储在这个表中

ID：自增的数据库主键

CONTAINER：消息的Destination

MSGID_PROD：消息发送者客户端的主键

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

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

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

PRIORITY：优先级，从0-9，数值越大优先级越高
1
2
3
4
5
6
7
8
9
10
11
12
13

2）ACTIVEMQ_ACKS：用于存储订阅关系。如果是持久化Topic，订阅者和服务器的订阅关系在这个表保存

CONTAINER：消息的Destination

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

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

SUB_NAME：订阅者名称

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

LAST_ACKED_ID：记录消费过的消息的ID。
1
2
3
4
5
6
7
8
9
10
11

3）ACTIVEMQ_LOCK
在集群环境中才有用，只有一个Broker可以获得消息，称为Master Broker，

其他的只能作为备份等待Master Broker不可用，才可能成为下一个Master Broker。
这个表用于记录哪个Broker是当前的Master Broker。

注意：
1）如果新建库OK + 上述配置OK +代码运行OK(运行activemq服务)，三张表会自动生成。万一情况，手动建表SQL（配置好不需要手动，应急）

2）一定要开启持久化设置，不然无法存储到数据库。即：

messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
1

小总结

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

如果是Topic：
一般是先启动消费订阅然后再生产的情况下会将消息保存到activemq_acks.
而且发布者和消费者都会在数据库中有记录，不会自动删除

1
2
3
4
5
6
7
8

开发遇到的坑
在配置关系型数据库作为ActiveMQ的持久化存储方案时，有些位置需要注意：

数据库jar包
记得需要使用到的相关jar文件放置到ActiveMQ安装路径下的lib目录。mysql-jdbc驱动的jar包和对应的数据库连接池jar包

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

下划线坑爹
“java.lang.lllegalStateException.BeanFactory not initialized or already closed”
这是因为操作系统的机器名中有“_”符号。请更改机器名并重启后即可解决问题
1
2
3
4
5
6
7
8
9

5、JDBC Message store with ActiveMQ Journal

说明：
这种方式克服了JDBC Store的不足，JDBC每次消息过来，都需要去写库和读库。
ActiveMQ Journal，使用高速缓存写入技术，大大提高了性能。
当消费者的消费速度能够及时跟上生产者消息的生产速度时，journal文件能够大大减少需要写入到DB中的消息。

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

配置

三、持久化机制小总结

持久化消息主要是指：
MQ所在的服务器down了消息不会丢失的机制。

ActiveMQ的消息持久化机制有：
AMQ 基于文件日志
KahaDB 基于日志文件，从ActiveMQ5.4开始默认的持久化插件
JDBC 基于第三方数据库
LevelDB 基于文件的本地数据库存储，从ActiveMQ5.8版本之后又推出了LevelDB的持久化引擎性能高于KahaDB
Replicated LevelDB Store 从ActiveMQ5.9提供了基于LevelDB和Zookeeper的数据复制方式，用于Master-slaver方式的首选数据复制方案。

无论使用哪种持久化方式，消息的存储逻辑都是一致的：
就是在发送者将消息发送出去后，消息中心首先将消息存储到本地数据库文件、内存数据库或远程数据库等，然后试图将消息发送给接受者，发送成功则将消息从存储中删除失败则继续尝试。消息中心启动以后首先要检查指定的存储位置，如果有未发送成功的消息，则需要把消息发送出去。