阅读 631

ActiveMQ 消息数据持久化

一、 问题提出

场景问题:服务器断电重启,未被消费的消息是否会在重启之后被继续消费?
两种选择:非持久性模式/持久性模式
非持久性模式: 服务器断电(关闭)之后,使用非持久性模式时,没有被消费的消息不会继续消费全部丢失;
程序会报一个连接关闭异常停止运行,继续启动服务器运行程序,不会接收任何消息。
持久性模式: 服务器断电(关闭)后,使用持久性模式时,没有被消费的消息会继续消费;
程序也会报连接关闭异常,但再次启动服务器和程序后,接收方还能继续原来的消息再次接收。

二、 ActiveMQ的几种存储模式选择
  • AMQ消息存储 (默认的消息存储)
    • 基于文件存储的消息数据库,不依赖于第三方数据库
  • KahaDB 消息存储(提供容量的提升和恢复能力)
    • 消息存储机制
  • JDBC 消息存储(消息基于JDBC存储)
    • 使用Mysql或Oracle数据库存储消息
  • Memory 消息存储(基于内容的消息存储)
    • ActiveMQ支持将消息保存到内存中,将Broker的“prsistent” 属性设置为“false”。
三、本文主要记录将ActiveMQ消息持久化到Mysql数据库
1. 将以下Jar包添加到到%ACTIVEMQ_HOME%\lib下面
  • commons-dbcp-1.4.jar
  • commons-pool-1.6.jar
  • mysql-connector-java-5.1.30.jar
2. 修改ActiveMQ的conf目录下的activemq.xml文件,修改数据持久化的方式

在添加如下配置时,需要注意添加位置

<broker>
<persistenceAdapter>
<!--createTablesOnStartup="true"表示在第一次启动时自动建立表结构,再次启动时,应该改为false-->
<jdbcPersistenceAdapter dataSource="#mysql-ds" createTablesOnStartup="true"/>
</persistenceAdapter>
</broker>复制代码

<beans>
<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/activemq?relaxAutoCommit=true"/>
        <property name="username" value="root"/>
        <property name="password" value="root"/>
        <property name="maxActive" value="200"/>
        <property name="poolPreparedStatements" value="true"/>
    </bean>
 </beans>复制代码

3. 启动ActiveMQ,分析数据表结构

在启动ActiveMQ之后,会在数据库<activemq>中建立三张表activemq_acksactivemq_lockactivemq_msgs

  1. activemq_acks:用于存储订阅关系。如果是持久化Topic,订阅者和服务器的订阅关系在这个表保存。
    • CONTAINER:消息的目的地
    • SUB_DEST:如果是使用static集群,这个字段会有集群其他系统的信息
    • CLIENT_ID:每个订阅者客户端ID
    • SUB_NAME:订阅者名称
    • SELECTOR:选择器,可以选择只消费满足条件的消息。条件可以用自定义属性实现,可支持多属性and和or操作
    • LAST_ACKED_ID:记录消费过的消息的id。
    • PRIORITY:优先级
    • XID:
  2. activemq_lock:用于记录哪一个Broker是Master Broker。这张表只有在集群环境中才会用到,在集群中,只能有一个Broker来接收消息,那么这个Broker就是主Broker,其他的作为从Broker,用来备份等待。
    • ID:主键
    • TIME:时间
    • BROKER_NAME:Broker名称
  3. activemq_msgs:用于存储消息,Topic和Queue消息都会保存在这张表中
    • ID:自增主键
    • CONTAINER:容器名称
    • MSGID_PROD:消息发送者客户端的主键
    • MSGID_SEQ:发送消息的顺序,msgid_prod+msg_seq可以组成jms的messageid
    • EXPIRATION:消息的过期时间,存储的是从1970-01-01到现在的毫秒数
    • MSG:消息本体的java序列化对象的二进制数据
    • PRIORITY:优先级,从0-9,数值越大优先级越高
参考:
  1. ActiveMQ 数据持久化
  2. ActiveMQ服务器数据迁移的一点心得
  3. activemq-queue-Browser
  4. ActiveMQ简介
关注下面的标签,发现更多相似文章
评论