ActiveMQ消息存储持久化

转https://www.cnblogs.com/xinhuaxuan/p/6128380.html

https://blog.csdn.net/lr131425/article/details/68064914

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

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

ActiveMQ提供了插件式的消息存储，主要有有如下几种:

1.AMQ消息存储-基于文件的存储方式，是以前的默认消息存储

2.KahaDB消息存储-提供了容量的提升和恢复能力，是现在的默认存储方式

3.JDBC消息存储-消息基于JDBC存储的

4.Memory消息存储-基于内存的消息存储

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

其配置位于activemq.xml文件中

可用的属性有:
1. director: KahaDB存放的路径，默认值activemq-data
2. indexWriteBatchSize: 批量写入磁盘的索引page数量,默认值为1000
3. indexCacheSize: 内存中缓存索引page的数量,默认值10000
4. enableIndexWriteAsync: 是否异步写出索引,默认false
5. journalMaxFileLength: 设置每个消息data log的大小，默认是32MB
6. enableJournalDiskSyncs: 设置是否保证每个没有事务的内容，被同步写入磁盘，JMS持久化的时候需要，默认为true
7. cleanupInterval: 在检查到不再使用的消息后,在具体删除消息前的时间,默认30000
8. checkpointInterval: checkpoint的间隔时间，默认是5000
9. ignoreMissingJournalfiles: 是否忽略丢失的消息日志文件，默认false
10. checkForCorruptJournalFiles: 在启动的时候，将会验证消息文件是否损坏，默认false
11. checksumJournalFiles: 是否为每个消息日志文件提供checksum,默认false
12. archiveDataLogs: 是否移动文件到特定的路径，而不是删除它们，默认false
13. directoryArchive: 定义消息已经被消费过后，移动data log到的路径，默认null
14. databaseLockedWaitDelay: 获得数据库锁的等待时间(used by shared master/slave),默认10000
15. maxAsyncJobs: 设置最大的可以存储的异步消息队列，默认值10000，可以和concurrent MessageProducers设置成一样的值。
16. concurrentStoreAndDispatchTransactions:是否分发消息到客户端，同时事务存储消息，默认true
17. concurrentStoreAndDispatchTopics: 是否分发Topic消息到客户端，同时进行存储，默认true
18. concurrentStoreAndDispatchQueues: 是否分发queue消息到客户端，同时进行存储，默认true

(2)AMQ Message Store概述

AMQ Message Store是ActiveMQ5.0缺省的持久化存储，它是一个基于文件、事务存储设计为快速消息存储的一个结构，该结构是以流的形式来进行消息交互的。
　　这种方式中，Messages被保存到data logs中，同时被reference store进行索引以提高存取速度。Data logs由一些简单的data log文件组成，缺省的文件大小是32M，如果某个消息的大小超过了data log文件的大小，那么可以修改配置以增加data log文件的大小。如果某个data log文件中所有的消息都被成功消费了，那么这个data log文件将会被标记，以便在下一轮的清理中被删除或者归档。
　　AMQ Message Store配置示例：

(3)使用JDBC来持久化消息

ActiveMQ支持使用JDBC来持久化消息，预定义的表如下：
1.消息表,缺省表明为ACTIVEMQ_MSGS, queue和topic都存储在里面，结构如下:

2.ACTIVEMQ_ACKS表存储持久订阅的信息和最后一个持久订阅接收的消息ID,结构如下:

3.锁定表，缺省表名为ACTIVEMQ_LOCK，用来确保在某一时刻，只能有一个ActiveMQ broker实例来访问数据库，结构如下：

使用JDBC来持久化消息的配置示例:
在activemq.xml中的配置如下:

 1 <broker brokerName="test-broker" persistent=true xmlns="http://activemq.apache.org/schema/core">2 　　<persistenceAdapter>3 　　　　<jdbcPersistenceAdapter dataSource="#mysql_ds">4 　　</persistenceAdapter>5 </broker>6 <bean name="mysql_ds" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">7 　　<property name="driverClassName">8 　　　　<value>org.gjt.mm.mysql.Driver</value>9 　　</property>
10 　　<property name="url">
11 　　　　<value>jdbc:mysql://192.168.1.102:3306/test?useUnicode=true&amp;characterEncoding=UTF-8</value>
12 　　</property>
13 　　<property name="username">
14 　　　　<value>root</value>
15 　　</property>
16 　　<property name="password">
17 　　　　<value>123123</value>
18 　　</property>
19 </bean>

JDBC Message Store with ActiveMQ Journal
这种方式克服了JDBC Store的不足，使用快速的缓存写入技术，大大提高了性能。
配置如下：

1 <beans>
2 　　<broker brokerName="test-broker" xmlns="http://activemq.apache.org/schema/core">
3 　　　　<persistenceFactory>
4 　　　　<journalPersistenceAdapterFactory journalLogFiles="4" journalLogFileSize="32768"
5 　　　　　　useJournal="true" useQuickJournal="true" dataSource="#mysql_ds" dataDirectory="activemq-data">
6 　　　　</persistenceFactory>
7 　　</broker>
8 </beans>

JDBC Store和JDBC Message Store with ActiveMQ Journal的区别：
　　1. JDBC with journal的性能优于jdbc
　　2. JDBC用于master/slave模式的数据库分享
　　3. JDBC with journal不能用于master/slave模式
　　4. 一般情况下，推荐使用jdbc with journal 
(4)Memory Message Store
　　内存消息存储主要是存储所有的持久化的消息在内存中。这里没有动态的缓存存在，所以你必须注意设置你的broker所在的JVM和内存限制。

1 <beans>
2 　　<broker brokerName="test_broker" persistent="false" xmlns="http://activemq.apache.org/schema/core">
3 　　　　<transportConnectors>
4 　　　　<transportConnector uri="tcp://localhost:61635">
5 　　　　</transportConnectors>
6 </beans>