8 ActiveMQ的持久化

1. 为了避免服务器意外宕机后导致消息队列数据丢失，需要引入持久化机制保证服务器重启后恢复消息队列数据使ActiveMQ达到高可用性.
2. ActiveMQ的消息持久化机制有JDBC，AMQ，KahaDB和LevelDB，无论使用哪种持久化方式，消息的存储逻辑都是一致的.
   就是在发送者将消息发送出去后，消息中心首先将消息存储到本地数据文件、内存数据库或者远程数据库等。再试图将消息发给接收者，成功则将消息从存储中删除，失败则继续尝试尝试发送。
3. 消息中心启动以后，要先检查指定的存储位置是否有未成功发送的消息，如果有，则会先把存储位置中的消息发出去.
4. 持久化方式:
   (1)AMQ（了解）
   AMQ 消息存储是一种基于文件存储形式，它具有写入速度快和容易恢复的特点。消息存储在一个个文件中文件的默认大小为32M，当一个文件中的消息已经全部被消费，那么这个文件将被标识为可删除，在下一个清除阶段，这个文件被删除。AMQ适用于ActiveMQ5.3之前的版本
   (2)KahaDB（重点）
   KahaDB消息存储是基于日志文件的存储方式，它是5.4版本之后默认存储方式。
   在ActiveMQ安装目录的conf/activemq.xml文件配置了ActiveMQ的默认持久化方式。
   
   其中，directory属性值配置了KahaDB持久化方式日志所在目录，即data/kahadb.
   
   KahaDB可用于任何场景，提高了性能和恢复能力。消息存储使用一个事务日志和仅仅用一个索引文件来存储它所有的地址。
   事务日志用于保存持久化数据，相当于新华字典内容；索引文件作为索引指向事务日志，相当于新华字典目录。
   KahaDB是一个专门针对消息持久化的解决方案，它对典型的消息使用模型进行了优化。
   数据被追加到data logs中。当不再需要log文件中的数据的时候，log文件会被丢弃。
   db-number.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
5. LevelDB（了解）
   LevelDB文件系统是从ActiveMQ5.8之后引进的，它和KahaDB很相似，也是基于文件的本地数据库存储形式，但是它提供比KahaDB更快的持久性，但它不再使用自定义B-Tree实现来索引预写日志，而是使用基于LevelDB的索引。相对于KahaDB它具有如下更好的优点：
   快速更新（无需进行随机磁盘更新）
   并发读取
   使用硬链接快速索引快照

为什么LevelDB比KahaDB有着更好的性能，为什么现在使用KahaDB使用广泛呢？按照官网的说法是：LevelDB存储已被弃用，不再支持或推荐使用，Replicated LevelDB（可复制的LevelDB）有望取代LevelDB。推荐的存储方式是KahaDB。参考官网地址：http://activemq.apache.org/leveldb-store

如何配置LevelDB?在ActiveMQ安装目录的conf/activemq.xml找到persistenceAdapter地方将原来默认的kahaDB进行如下修改：

 <persistenceAdapter>
 		<!--<kahaDB directory="${activemq.data}/kahadb"/> -->
        <levelDB directory="${activemq.data}/leveldb"/>
 </persistenceAdapter>

1
2
3
4
5

4. JDBC（重点）
   JDBC持久化方式顾名思义就是将数据持久化到数据库中（如mysql），实现步骤如下：
   (1)添加mysql驱动到ActiveMQ安装目录的lib文件夹中:
   
   (2)配置ActiveMQ.xml:
   在ActiveMQ安装目录的conf/activemq.xml中找到persistenceAdapter标签替换有以下内容:

<persistenceAdapter>
        <!--<kahaDB directory="${activemq.data}/kahadb"/> -->
        <jdbcPersistenceAdapter dataSource="#mysql-ds" createTablesOnStartup="true" />
</persistenceAdapter>

1
2
3
4
5

dataSource属性值指定将要引用的持久化数据库的bean名称（后面会配置一个名为mysql-ds的bean）；createTablesOnStartup属性值是否在启动的时候创建数据库表，默认是true，这样每次启动都会去创建表了，一般是第一次启动的时候设置为true，然后再去改成false。
(3)数据库连接池配置:
配置连接池之前先在数据库中建立一个数据库，比如我建立数据库名称为：activemq_test

<bean id="mysql-ds" class="org.apache.commons.dbcp2.BasicDataSource" destroy-method="close">
		<!-- 数据库驱动名称，此处是mysql驱动名称-->
        <property name="driverClassName" value="com.mysql.jdbc.Driver"/>
        <!-- 连接数据库url，ip换成自己数据库所在的机器ip，数据库名为新建立数据库的名称-->
        <property name="url" value="jdbc:mysql://your ip:3306/your db's name?relaxAutoCommit=true&serverTimezone=GMT"/>
        <!-- 连接数据库用户名-->
        <property name="username" value="your username"/>
        <!-- 连接数据库密码-->
        <property name="password" value="your password"/>
        <property name="poolPreparedStatements" value="true"/>
</bean>

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

ActiveMQ默认使用DBCP连接池，并且自带了DBCP连接池相关jar包，如果想要换成C3P0等连接池，需要自行引入相关jar包。其中bean的id属性值一定要和上面的dataSource属性值一样。
1

(4)启动ActiveMQ
运行命令./activemq start启动服务，不出意外的话，启动成功后，将会在配置的数据库中生成相应的三张表:

表创建后记得修改activemq.xml的jdbcPersistenceAdapter标签的createTablesOnStartup属性值改为false
1

ACTIVEMQ_MSGS：消息表，Queue和Topic都存在里面:

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

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

<!-- <persistenceAdapter> -->
<!--<kahaDB directory="${activemq.data}/kahadb"/> -->
<!--<jdbcPersistenceAdapter dataSource="#mysql-ds" createTablesOnStartup="true" /> -->
<!-- </persistenceAdapter> -->

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

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

总结:
从最初的AMQ Message Store方案到ActiveMQ V4版本退出的High Performance Journal（高性能事务支持）附件，并且同步推出了关于关系型数据库的存储方案。ActiveMQ5.3版本又推出了对KahaDB的支持（5.4版本后被作为默认的持久化方案），后来ActiveMQ 5.8版本开始支持LevelDB，到现在5.9提供了标准的Zookeeper+LevelDB集群化方案。

ActiveMQ消息持久化机制有：

AMQ 基于日志文件
KahaDB 基于日志文件，从ActiveMQ5.4开始默认使用
JDBC 基于第三方数据库
Replicated LevelDB Store 从5.9开始提供了LevelDB和Zookeeper的数据复制方法，用于Master-slave方式的首选数据复制方案。