ActiveMq消息持久化

消息持久化

消息持久化是保证消息不丢失的重要方式！！！
ActiveMQ提供了以下三种的消息存储方式：

（1） Memory 消息存储-基于内存的消息存储。
（2） 基于日志消息存储方式，KahaDB是ActiveMQ的默认日志存储方式，它提供了容量的提升和恢复
能力。
（3） 基于JDBC的消息存储方式-数据存储于数据库（例如：MySQL）中。
ActiveMQ持久化机制流程图：

KahaDB存储

消息存储使用一个事务日志和仅仅用一个索引文件来存储它所有的地址：（文件在apache-activemq\data\kahadb下）

1、db-.log（主要存数据）是KahaDB存储消息到预定义大小的数据记录文件，文件命名为db-.log，当数据文件已满时，一个新的文件会随之创建，number数值也会随之递增，它随之消息数量的增多，如每32M一个文件，文件名按照数字进行编号，如sb-1.log……当不再有引用到数据文件中的任何消息时，文件会被删除或归档。

2、db.data文件包含了持久化的BTree索引，索引了消息数据记录中的消息，它是消息的索引文件，本质上是B-Tree（B树），使用B-Tree作为索引指向db-.log里面存储的消息。

3、db.free文件表示当前db.data文件哪些页面是空闲的，文件具体内容是所有空闲页的ID。

4、db.redo文件是用来进行消息恢复，如果KahaDB消息存储在强制退出后启动，用于恢复BTree索引。

5、lock文件表示当前获得KahaDB读写权限的broker。

可以更改的属性：

官网说明：http://activemq.apache.org/kahadb

测试：把mq直接杀死，然后已经消费的消息不在了，未读的还存在

[root@localhost data]# ll
总用量 12
-rw-r--r--. 1 root root 6026 10月 14 11:07 activemq.log
-rw-r--r--. 1 root root    6 10月 13 10:29 activemq.pid
-rw-r--r--. 1 root root    0 10月 13 10:29 audit.log
drwxr-xr-x. 2 root root   64 10月 13 10:29 kahadb
[root@localhost data]# ps -ef |grep activemq
root      22343      1  0 10月13 ?      00:04:59 /usr/bin/java -Xms64M -Xmx1G -Djava.util.logging.config.file=logging.properties -Djava.security.auth.login.config=/root/apache-activemq-5.14.0//conf/login.config -Dcom.sun.management.jmxremote -Djava.awt.headless=true -Djava.io.tmpdir=/root/apache-activemq-5.14.0//tmp -Dactivemq.classpath=/root/apache-activemq-5.14.0//conf:/root/apache-activemq-5.14.0//../lib/: -Dactivemq.home=/root/apache-activemq-5.14.0/ -Dactivemq.base=/root/apache-activemq-5.14.0/ -Dactivemq.conf=/root/apache-activemq-5.14.0//conf -Dactivemq.data=/root/apache-activemq-5.14.0//data -jar /root/apache-activemq-5.14.0//bin/activemq.jar start
root      25880  20668  0 14:09 pts/0    00:00:00 grep --color=auto activemq
[root@localhost data]# kill -9 22343
[root@localhost data]# ll
总用量 12
-rw-r--r--. 1 root root 6026 10月 14 11:07 activemq.log
-rw-r--r--. 1 root root    6 10月 13 10:29 activemq.pid
-rw-r--r--. 1 root root    0 10月 13 10:29 audit.log
drwxr-xr-x. 2 root root   64 10月 13 10:29 kahadb
[root@localhost data]# cd ..
[root@localhost apache-activemq-5.14.0]# ll
总用量 15872
-rwxr-xr-x. 1 root root 16195019 8月   2 2016 activemq-all-5.14.0.jar
drwxr-xr-x. 5 root root      147 10月 13 10:23 bin
drwxr-xr-x. 2 root root     4096 10月 13 10:23 conf
drwxr-xr-x. 3 root root       77 10月 13 10:29 data
drwxr-xr-x. 2 root root       76 10月 13 10:23 docs
drwxr-xr-x. 7 root root       71 8月   2 2016 examples
drwxr-xr-x. 6 root root     4096 10月 13 10:23 lib
-rw-r--r--. 1 root root    40580 8月   2 2016 LICENSE
-rw-r--r--. 1 root root     3334 8月   2 2016 NOTICE
-rw-r--r--. 1 root root     2610 8月   2 2016 README.txt
drwxr-xr-x. 4 root root      136 10月 13 10:29 tmp
drwxr-xr-x. 6 root root       95 10月 13 10:23 webapps
drwxr-xr-x. 3 root root       18 10月 13 10:23 webapps-demo
[root@localhost apache-activemq-5.14.0]# cd bin
[root@localhost bin]# ./activemq start
INFO: Loading '/root/apache-activemq-5.14.0//bin/env'
INFO: Using java '/usr/bin/java'
INFO: Starting - inspect logfiles specified in logging.properties and log4j.properties to get details
INFO: pidfile created : '/root/apache-activemq-5.14.0//data/activemq.pid' (pid '25920')

测试图

JDBC消息存储

1、将原来的kshadb的持久化数据的方式更改为jdbc：

<bean id="mysql-ds" class="com.alibaba.druid.pool.DruidDataSource" destroy-method="close">
  <property name="driverClassName" value="com.mysql.jdbc.Driver"/>
  <property name="url" value="jdbc:mysql://192.168.1.104:3306/db_activemq?relaxAutoCommit=true"/>
  <property name="username" value="root"/>
  <property name="password" value="root"/>
</bean>

       <persistenceAdapter>
            <!--<kahaDB directory="${activemq.data}/kahadb"/>-->
            <jdbcPersistenceAdapter dataSource="#mysql-ds" createTablesOnStartup="true"/>
        </persistenceAdapter>

注意：dataSource指定将要引用的持久化数据库的bean名称，createTablesOnStartup表示是否在启动时创建数据表，默认值为true，这样每次启动都会创建数据表，一般是第一次启动的时候设置为true之后改为false。

自动创建的三张数据表说明：

• 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，数值越大优先级越高
• activemq_acks用于存储订阅关系。如果是持久化Topic，订阅者和服务器的订阅关系在这个表保存：
  主要的数据库字段如下：
  CONTAINER：消息的Destination
  SUB_DEST：如果是使用Static集群，这个字段会有集群其他系统的信息
  CLIENT_ID：每个订阅者都必须有一个唯一的客户端ID用以区分
  SUB_NAME：订阅者名称
  SELECTOR：选择器，可以选择只消费满足条件的消息。条件可以用自定义属性实现，可支持多属性AND和OR操作
  LAST_ACKED_ID：记录消费过的消息的ID。
• activemq_lock在集群环境中才有用，只有一个Broker可以获得消息，称为Master Broker，
  其他的只能作为备份等待Master Broker不可用，才可能成为下一个Master Broker。
  这个表用于记录哪个Broker是当前的Master Broker。

• 如果是queue模式，在没有消费者的情况下会将消息保存到activemq_msgs表中，只要有任意一个消费者已经消费过，相应的消息将会立即被删除。

• 如果是topic模式，一般是先启动消费订阅然后再生产的情况下会将消息保存到activemq_acks表中。