ActiveMQ 消息數據持久化

1、 問題提出

場景問題：服務器斷電重啓，未被消費的消息是否會在重啓以後被繼續消費？
兩種選擇：非持久性模式/持久性模式
非持久性模式： 服務器斷電(關閉)以後，使用非持久性模式時，沒有被消費的消息不會繼續消費所有丟失；
程序會報一個鏈接關閉異常中止運行，繼續啓動服務器運行程序，不會接收任何消息。
持久性模式： 服務器斷電(關閉)後，使用持久性模式時，沒有被消費的消息會繼續消費；
程序也會報鏈接關閉異常，但再次啓動服務器和程序後，接收方還能繼續原來的消息再次接收。

2、 ActiveMQ的幾種存儲模式選擇

• AMQ消息存儲 (默認的消息存儲)
  • 基於文件存儲的消息數據庫，不依賴於第三方數據庫
• KahaDB 消息存儲(提供容量的提高和恢復能力)
  • 消息存儲機制
• JDBC 消息存儲(消息基於JDBC存儲)
  • 使用Mysql或Oracle數據庫存儲消息
• Memory 消息存儲(基於內容的消息存儲)
  • ActiveMQ支持將消息保存到內存中，將Broker的「prsistent」 屬性設置爲「false」。

3、本文主要記錄將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_acks、activemq_lock、activemq_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簡介