ActiveMQ消息持久化機制

ActiveMQ的幾種消息持久化機制

為了避免意外當機以後丢失資訊，需要做到重新開機後可以恢複消息隊列，消息系統一般都會采用持久化機制。

ActiveMQ的消息持久化機制有JDBC，AMQ，KahaDB和LevelDB。

1. JDBC持久化方式

使用JDBC持久化方式，資料庫會建立3個表：activemq_msgs，activemq_acks和activemq_lock。

activemq_msgs用于存儲消息，Queue和Topic都存儲在這個表中。

（1）配置方式

配置持久化的方式，都是修改安裝目錄下conf/acticvemq.xml檔案，

首先定義一個mysql-ds的MySQL資料源，然後在persistenceAdapter節點中配置jdbcPersistenceAdapter并且引用剛才定義的資料源。

1 2 3

< persistenceAdapter >  < jdbcPersistenceAdapter dataSource="#mysql-ds" createTablesOnStartup="false" />  </ persistenceAdapter >

dataSource指定持久化資料庫的bean，createTablesOnStartup是否在啟動的時候建立資料表，預設值是true，這樣每次啟動都會去建立資料表了，一般是第一次啟動的時候設定為true，之後改成false。

使用MySQL配置JDBC持久化:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

< beans > < broker brokerName="test-broker" persistent="true" xmlns="http://activemq.apache.org/schema/core"> < persistenceAdapter > < jdbcPersistenceAdapter dataSource="#mysql-ds" createTablesOnStartup="false"/> </ persistenceAdapter > </ broker > < 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="activemq"/> < property name="password" value="activemq"/> < property name="maxActive" value="200"/> < property name="poolPreparedStatements" value="true"/> </ bean > </ beans >

　

（2）資料庫表資訊

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。

2. AMQ方式

性能高于JDBC，寫入消息時，會将消息寫入日志檔案，由于是順序追加寫，性能很高。為了提升性能，建立消息主鍵索引，并且提供緩存機制，進一步提升性能。每個日志檔案的大小都是有限制的（預設32m，可自行配置）。

當超過這個大小，系統會重建立立一個檔案。當所有的消息都消費完成，系統會删除這個檔案或者歸檔（取決于配置）。

主要的缺點是AMQ Message會為每一個Destination建立一個索引，如果使用了大量的Queue，索引檔案的大小會占用很多磁盤空間。

而且由于索引巨大，一旦Broker崩潰，重建索引的速度會非常慢。

配置片段如下：

1 2 3

< persistenceAdapter > < amqPersistenceAdapter directory="${activemq.data}/activemq-data" maxFileLength="32mb"/> </ persistenceAdapter >

雖然AMQ性能略高于下面的Kaha DB方式，但是由于其重建索引時間過長，而且索引檔案占用磁盤空間過大，是以已經不推薦使用。

3. KahaDB方式

KahaDB是從ActiveMQ 5.4開始預設的持久化插件，也是我們項目現在使用的持久化方式。

KahaDb恢複時間遠遠小于其前身AMQ并且使用更少的資料檔案，是以可以完全代替AMQ。

kahaDB的持久化機制同樣是基于日志檔案，索引和緩存。

配置方式：

1 2 3 4 5

< persistenceAdapter > < kahaDB directory="${activemq.data}/activemq-data" journalMaxFileLength="16mb"/> </ persistenceAdapter > directory : 指定持久化消息的存儲目錄 journalMaxFileLength : 指定儲存消息的日志檔案大小，具體根據你的實際應用配置

（1）KahaDB主要特性

1、日志形式存儲消息；

2、消息索引以B-Tree結構存儲，可以快速更新；

3、完全支援JMS事務；

4、支援多種恢複機制；

（2）KahaDB的結構

消息存儲在基于檔案的資料日志中。如果消息發送成功，變标記為可删除的。系統會周期性的清除或者歸檔日志檔案。

消息檔案的位置索引存儲在記憶體中，這樣能快速定位到。定期将記憶體中的消息索引儲存到metadata store中，避免大量消息未發送時，消息索引占用過多記憶體空間。

Data logs：

Data logs用于存儲消息日志，消息的全部内容都在Data logs中。

同AMQ一樣，一個Data logs檔案大小超過規定的最大值，會建立一個檔案。同樣是檔案尾部追加，寫入性能很快。

每個消息在Data logs中有計數引用，是以當一個檔案裡所有的消息都不需要了，系統會自動删除檔案或放入歸檔檔案夾。

Metadata cache ：

緩存用于存放線上消費者的消息。如果消費者已經快速的消費完成，那麼這些消息就不需要再寫入磁盤了。

Btree索引會根據MessageID建立索引，用于快速的查找消息。這個索引同樣維護持久化訂閱者與Destination的關系，以及每個消費者消費消息的指針。

Metadata store 

在db.data檔案中儲存消息日志中消息的中繼資料，也是以B-Tree結構存儲的，定時從Metadata cache更新資料。Metadata store中也會備份一些在消息日志中存在的資訊，這樣可以讓Broker執行個體快速啟動。

即便metadata store檔案被破壞或者誤删除了。broker可以讀取Data logs恢複過來，隻是速度會相對較慢些。

4.LevelDB方式

從ActiveMQ 5.6版本之後，又推出了LevelDB的持久化引擎。

目前預設的持久化方式仍然是KahaDB，不過LevelDB持久化性能高于KahaDB，可能是以後的趨勢。

在ActiveMQ 5.9版本提供了基于LevelDB和Zookeeper的資料複制方式，用于Master-slave方式的首選資料複制方案。

文章轉載至：https://www.cnblogs.com/binyue/p/5371479.html