ActiveMQ消息数据持久化
2017-05-17 本文已影响1233人
Xiangdong_She
一、 问题提出
场景问题：服务器断电重启，未被消费的消息是否会在重启之后被继续消费？
两种选择：非持久性模式/持久性模式
非持久性模式：服务器断电(关闭)之后，使用非持久性模式时，没有被消费的消息不会继续消费全部丢失；
程序会报一个连接关闭异常停止运行，继续启动服务器运行程序，不会接收任何消息。
持久性模式：服务器断电(关闭)后，使用持久性模式时，没有被消费的消息会继续消费；
程序也会报连接关闭异常，但再次启动服务器和程序后，接收方还能继续原来的消息再次接收。
二、 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>
persistenceAdapter.png
<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>
datasource.png
3. 启动ActiveMQ，分析数据表结构
在启动ActiveMQ之后，会在数据库<
activemq>中建立三张表activemq_acks、activemq_lock、activemq_msgs
- activemq_acks：用于存储订阅关系。如果是持久化Topic，订阅者和服务器的订阅关系在这个表保存。
- CONTAINER：消息的目的地
- SUB_DEST：如果是使用static集群，这个字段会有集群其他系统的信息
- CLIENT_ID：每个订阅者客户端ID
- SUB_NAME：订阅者名称
- SELECTOR：选择器，可以选择只消费满足条件的消息。条件可以用自定义属性实现，可支持多属性and和or操作
- LAST_ACKED_ID：记录消费过的消息的id。
- PRIORITY：优先级
- XID：
- activemq_lock：用于记录哪一个Broker是Master Broker。这张表只有在集群环境中才会用到，在集群中，只能有一个Broker来接收消息，那么这个Broker就是主Broker，其他的作为从Broker，用来备份等待。
- ID：主键
- TIME：时间
- BROKER_NAME：Broker名称
- 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，数值越大优先级越高
