ActiveMQ 原理分析—消息持久化篇

消息持久化策略

背景
当消息发送者(provider)发送消息后消费者(consumer)没启动、故障, 或者消息中心在发送者发送消息后宕机了。ActiveMQ是如何保证消息不丢失，消费者能够正常的消费已发送到消息中心的消息。
原理
消息持久性的原理很简单，就是在发送消息出去后，消息中心首先将消息存储在本地文件、内存或者远程数据库，然后消费者监听消息进行消费。消费成功后会通知消息中心把消息从存储中删除，失败则继续尝试。

持久化消息和非持久化消息的存储原理

ActiveMQ是按照非持久化消息存储在内存中、持久化消息存储在文件中的机制来设计的。
具体它是通过配置文件进行配置的。在ActiveMQ安装目录config文件夹下有个activemq.xml文件内容如下。

<systemUsage><systemUsage><memoryUsage>//标记设置整个ActiveMQ节点的“可用内存限制”。这个值不能超过ActiveMQ本身设置的最大内存大小。其中的//percentOfJvmHeap属性表示百分比。占用70%的JVM堆内存<memoryUsage percentOfJvmHeap="70" /></memoryUsage><storeUsage>//该标记设置整个ActiveMQ节点，用于存储“持久化消息”的“可用磁盘空间”。该子标记的limit属性必须要进行设置<storeUsage limit="100 gb"/></storeUsage><tempUsage>//一旦ActiveMQ服务节点存储的消息达到了memoryUsage的限制，非持久化消息就会被转储到 temp store区域，虽然//我们说过非持久化消息不进行持久化存储，但是ActiveMQ为了防止“数据洪峰”出现时非持久化消息大量堆积致使内存耗//尽的情况出现，还是会将非持久化消息写入到磁盘的临时区域——temp store。这个子标记就是为了设置这个temp//store区域的“可用磁盘空间限制<tempUsage limit="50 gb"/></tempUsage></systemUsage>
</systemUsage>

SystemUsage配置设置了一些系统内存和硬盘容量。从上面的配置信息可以得出：当非持久化消息堆积到一定程度的时候(到我们设置的阈值)ActiveMQ会将内存中的非持久化消息写入到临时文件。但是非持久化消息在临时文件遇到系统宕机不会进行消息恢复。

ActiveMQ主要提供以下几种消息持久化方式

KahaDB存储（默认存储方式）
JDBC存储
Memory存储
LevelDB存储
JDBC With ActiveMQ Journal
KahaDB存储
KahaDB是目前默认的存储方式,消息存储使用一个事务日志和一个索引（B-tree）文件来存储它所有的地址。KahaDB是一个专门针对消息持久化的解决方案,它对典型的消息使用模式进行了优化。在Kaha中,数据被append到data logs中。当不再需要log文件中的数据的时候,log文件会被丢弃。
配置方式如下：

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

KahaDB存储的文件介绍

db.data 它是消息的索引文件，采用B-Tree（B树）进行存储，使用B-Tree作为索引指向db-*.log里面存储的消息。
db.redo 用来进行消息恢复。
db-*.log 存储消息内容。新的数据以APPEND的方式追加到日志文件末尾。属于顺序写入，因此消息存储是比较快的。默认是32M，达到阀值会自动递增。会新增个文件采用db-阿拉伯数字进行文件命名.
lock文件锁，解决集群环境下leader 写的问题。

JDBC存储
在配置文件配置以下内容：

<persistenceAdapter><!-- createTablesOnStartup  是否在启动的时候创建数据表 --><jdbcPersistenceAdapter dataSource="# MySQL-DS " createTablesOnStartup="true" />
</persistenceAdapter>
<!--配置数据库连接信息-->
<bean id="Mysql-DS" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close"><property name="driverClassName" value="com.mysql.jdbc.Driver"/><property name="url" value="jdbc:mysql://192.168.11.156:3306/activemq?relaxAutoCommit=true"/><property name="username" value="root"/><property name="password" value="root"/>
</bean>

配置文件完以后需要往 ${ActiveMQ_HOME}/lib 文件夹中添加相应 jar 包：

配置好后在启动ActiveMQ的时候会创建三张表 activemq_msgs，activemq_acks和activemq_lock。

ACTIVEMQ_MSGS ：消息表，queue和topic都存在这个表中。
ACTIVEMQ_ACKS ：存储持久订阅的信息和最后一个持久订阅接收的消息ID 。
ACTIVEMQ_LOCKS：锁表，用来确保某一时刻，只能有一个ActiveMQ broker实例来访问数据库。

Memory 内存存储
配置如下：

     <broker brokerName="my-broker"  persistent="false"   xmlns="http://activemq.apache.org/schema/core"><transportConnectors><transportConnector uri="tcp://192.168.1.21:61635"/></transportConnectors></broker>

基于内存的消息存储，内存消息存储主要是存储所有的持久化的消息在内存中。persistent=”false”,表示不设置持久化存储，直接存储到内存中。

JDBC Message store with ActiveMQ Journal
配置如下：

<persistenceFactory><journalPersistenceAdapterFactory dataSource="#Mysql-DS" dataDirectory="activemq-data"/>
</persistenceFactory>

这种方式优化了jdbc 存储性能问题，JDBC每次消息过来，都需要去写库和读库。而ActiveMQ Journal 采用了内存缓存技术，会先将消息放入缓存中，然后再将内存中的消息批量同步到数据库。它是这样进行优化的：在这期间如果消费者把消息消费掉后会立即删除 Journal内存中的消息，这样就不需要同步到数据库中。减少了写库的资源浪费。