ActiveMQ 消息存储

Posted soprano

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了ActiveMQ 消息存储相关的知识,希望对你有一定的参考价值。

深入浅出 JMS(四) - ActiveMQ 消息存储

 一、消息的存储方式

ActiveMQ 支持 JMS 规范中的持久化消息与非持久化消息

  • 持久化消息通常用于不管是否消费者在线,它们都会保证消息会被消费者消费。当消息被确认消费后,会从存储中删除
  • 非持久化消息通常用于发送通知以及实时数据,通常要求性能优先,消息可靠性并不是必须的情况
  • MQ 支持可插拔式的消息存储,如:内存、文件和关系数据库等方式

Queue 消息模型在 ActiveMQ 的存储:

  采用存储采用先进先出(FIFO),一个消息只能被一个消费者消费,当消息被确认消费之后才会被删除。

Topic消息模型(针对持久订阅):

  每个订阅者获取的消息实际是消息的一个副本,只有一个消息副本会被存储,MQ 提供了一个指针来指向消息存储并且分发消息副本到订阅者,消息直到所有的持久化订阅者都被接收才能被删除。

持久化存储方式:

  • KahaDB 消息存储
  • AMQ 消息存储
  • JDBC 消息存储
  • 内存消息存储

二、KahaDB 存储方式

  KahaDB 是从 ActiveMQ 5.4 开始默认的持久化插件。KahaDb 恢复时间远远小于其前身 AMQ 并且使用更少的数据文件,所以可以完全代替 AMQ,kahaDB 的持久化机制同样是基于日志文件,索引和缓存。

(一)KahaDB 主要特性:

  • 日志形式存储消息;
  • 消息索引以 B-Tree 结构存储,可以快速更新;
  • 完全支持 JMS 事务;
  • 支持多种恢复机制;

(二)适用场景:

高吞吐量的应用程序
存储大数据量的消息

(三)配置方式 conf/activemq.xml:

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

技术图片

(四)KahaDB 存储原理

当有活动消费者时,用于临时存储,消息会被发送给消费着,同时被安排将被存储,如果消息及时被确认,就不需要写入到磁盘。写入到磁盘中的数据消息,在后续的消息活动中,如果消息发送成功,变标记为可删除的。系统会周期性的清除或者归档日志文件。

(1) KahaDB 内部结构

技术图片

  • Data logs:消息日志包含了消息日志和一些命令
  • Cache:当有活动消费者时,用于临时存储,消息会被发送给消费着,同时被安排将被存储,如果消息及时被确认,这不需要写入到磁盘
  • Btree indexes(消息索引):用于引用消息日志(message id),它存储在内存中,这样能快速定位到。MQ会定期将内存中的消息索引保存到 metadata store 中,避免大量消息未发送时,消息索引占用过多内存空间。
  • Redo log:用于在非正常关机情况下维护索引完整性。

(2) 目录结构

技术图片

  • Db log files:用于存储消息(默认大小32M),当 log 日志满了,会创建一个新的,当 log 日志中的消息都被删除,该日志文件会被删除或者归档。
  • Archive directory:当 datalog 不在被 kahadb 需要会被归档(通过 archiveDataLogs 属性控制)。
  • Db.data:存放 Btree indexs。
  • Db.redo:存放 redo file,用于恢复 Btree indexs。

三、AMQ 消息存储

写入消息时,会将消息写入日志文件,由于是顺序追加写,性能很高。为了提升性能,创建消息主键索引,并且提供缓存机制,进一步提升性能。每个日志文件的大小都是有限制的(默认32m,可自行配置)。当超过这个大小,系统会重新建立一个文件。当所有的消息都消费完成,系统会删除这个文件或者归档(取决于配置)。主要的缺点是 AMQ Message 会为每一个 Destination 创建一个索引,如果使用了大量的 Queue,索引文件的大小会占用很多磁盘空间。而且由于索引巨大,一旦 Broker 崩溃,重建索引的速度会非常慢。

特点:类似 KahaDB,也包含了事务日志,每个 destination 都包含一个 index 文件,AMQ 适用于高吞吐量的应用场景,但是不适合多个队列的场景。

配置方式 conf/activemq.xml:

<!--AMQ directory:数据存储路径 syncOnWrite:是否同步写入  maxFileLength:日志文件大小 -->
<persistenceAdapter>
    <amqPersistenceAdapter
        directory="${activemq.data}/AMQdb"
        syncOnWrite="true"
        maxFileLength="10mb" />
</persistenceAdapter>

(1) AMQ内部结构

技术图片

  • Data logs:消息日志包含了消息日志
  • Cache:用于消息的快速检索
  • Reference store indexes:用于引用 datalogs 中的消息,通过 message ID 关联

(2) 目录结构

技术图片

  • Lock:保证同一时间只有一个 borker 访问文件目录
  • temp-storag:用于存储非持久化消息(当不在被存储在内存中),如等待慢消费者处理消息
  • Kr-store:用于存储引用消息日志数据
  • journal directory:包含了消息文件、消息日志和消息控制信息
  • Archive:归档的数据日志

四、JDBC存储

支持通过 JDBC 将消息存储到关系数据库,性能上不如文件存储,能通过关系型数据库查询到消息的信息。

MQ 支持的数据库:Apache Derby、mysql、PostgreSQL、Oracle、SQLServer、Sybase、Informix、MaxDB。

存储表结构:

表 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,数值越大优先级越高

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

字段说明
CONTAINER 消息的Destination
SUB_DEST 如果是使用Static集群,这个字段会有集群其他系统的信息
CLIENT_ID 每个订阅者都必须有一个唯一的客户端ID用以区分
SUB_NAME 订阅者名称
SELECTOR 选择器,可以选择只消费满足条件的消息。条件可以用自定义属性实现,可支持多属性AND和OR操作
LAST_ACKED_ID 记录消费过的消息的ID

表 3:ACTIVEMQ_LOCK(消息锁,保证同一时间只能有一个broker访问这些表结构):

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

配置方式:

1、配置数据源 conf/acticvemq.xml 文件:

<!-- 配置数据源-->
<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://localhost:3306/activemq?relaxAutoCommit=true"/>
    <property name="username" value="root"/>
    <property name="password" value="111111"/>
    <property name="maxActive" value="200"/>
    <property name="poolPreparedStatements" value="true"/>
</bean>

2、配置 broke 中的 persistenceAdapter

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

<!-- JDBC配置 -->
<persistenceAdapter>
    <jdbcPersistenceAdapter dataSource="#mysql-ds"  createTablesOnStartup="false"/>
</persistenceAdapter> 

ps:数据库 activemq 需要手动创建。

五、内存消息存储

内存消息存储,会将所有的持久化消息存储在内存中,必须注意JVM使用情况以及内存限制,适用于一些能快速消费的数据量不大的小消息,当MQ关闭或者宕机,未被消费的内存消息会被清空。

配置方式 设置 broker属性值 persistent="false":

<broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="${activemq.data}" persistent="false"/>

每天用心记录一点点。内容也许不重要,但习惯很重要!

以上是关于ActiveMQ 消息存储的主要内容,如果未能解决你的问题,请参考以下文章

学习ActiveMQ:activemq的持久化

ActiveMQ的学习(ActiveMQ的持久化)

ActiveMQ 消息存储

ActiveMQ 消息存储持久化

ActiveMQ消息存储持久化

在 ActiveMQ 中存储和处理消息