ActiveMQ中如何實現消息存儲

這篇文章將為大家詳細講解有關ActiveMQ中如何實現消息存儲，小編覺得挺實用的，因此分享給大家做個參考，希望大家閱讀完這篇文章后可以有所收獲。

一、隊列和topic

• 概述
  ActiveMQ不僅支持persistent和non-persistent兩種方式，還支持消息的恢復（recovery）方式

• PTP
  Queue的存儲方式很簡單，就是一個FIFO（先進先出）的Queue
  

• PUB/SUB
  對于持久化訂閱主題，每一個消費者將獲得一個消息的復制
  

• 有效的消息存儲
    ActiveMQ提供了一個插件式的消息存儲，類似于消息的多點傳播，主要實現了如下幾種：
  1：AMQ消息存儲一基于文件的存儲方式，是以前的默認消息存儲
  2：KahaDB消息存儲一提供了容量的提升和恢復能力，是現在的默認存儲方式
  3：JDBC消息存儲一消息基于JDBC存儲的
  4：Memory消息存儲一基于內存的消息存儲

二、KahaDB

• KahaDB Message Store概述
      KahaDB是目前默認的存儲方式,可用于任何場景，提高了性能和恢復能力。消息存儲使用一個事務日志和僅僅用一個索引文件來存儲它所有的地址。
      KahaDB是一個專門針對消息持久化的解決方案，它對典型的消息使用模式進行了優化。在Kaha中，數據被追加到datalogs中。當不再需要log文件中的數的時候，log文件會被丟棄。

• KahaDB基本配置例子

  <persistenceAdapter>
  <kahaDB directory="${actlvemq.data}/kahadb"/>
  </persistenceAdapter>

  可用的屬性有：
  1：director:KahaDB存放的路徑，默認值activemq-data
  2：indexWriteBatchSize:批量寫入磁盤的索引page數量，默認值1000
  3：indexCacheSize：內存中緩存索引page的數量，默認值10000
  4：enableIndexWriteAsync：是否異步寫出索引，默認false
  5：journalMaxFi1eLength：設置每個消息data log的大小，默認是32M
  6：enab1eJournalDiskSyncs：設置是否保證每個沒有事務的內容，被同步寫入磁盤，JMS持久化的時候需要，默認為true
  7：cleanupInterval：在檢查到不再使用的消后，在具體刪除消息前的時間，默認30000
  8：checkpointInterval：checkpoint的間隔時間，默認5000
  9：ignoreMissingJournalfiles：是否忽略丟失的消息日志文件，默認false
  10：checkForCourruptJournalFiles:在啟動的時候，將會驗證消息文件是否損壞，默認為false
  11：checksumJournalFiles：是否為每個消息日志文件提供checksum，默認false
  12：archiveDataLogs：是否移動文件到特定的路徑，而不是刪除它們，默認false
  13：directoryArchive：定義消息已經被消費后，移動data log到的路徑，默認為null
  14：databaseLockedWaitDelay：獲得數據庫鎖的等待時間，默認10000
  15：maxAsyncJobs：設置最大的可以存儲的異步消息隊列，默認10000，可以和concurrent MessageProducers設置成一樣的值
  16：concurrentStoreAndDispatchTransactions：是否分發消息到客戶端，同時事務存儲消息，默認true
  17：concurrentStoreAndDispatchTopics：是否分發Topic消息到客戶端，同時進行存儲，默認true
  18：concurrentStoreAndDispatchQueues：是否分發queue消息到客戶端，同時進行存儲，默認true

• 在Java中內嵌使用Broker，使用KahaDB的例子

  public class EmbeddedBrokerUsingKahaDBStoreExample {
  public BrokerService createEmbeddedBroker()throws Exception{
      BrokerService broker = new BrokerService();
      File dataFileDir = new File("target/amq-in-action/kahadb");
      KahaDBStore kaha = new KahaDBStore();
      kaha.setDirectory(dataFileDir);
      kaha.setJournalMaxFileLength(1024*1000);
      kaha.setIndexWriteBatchSize(100);
      kaha.setEnableIndexWriteAsync(true);
  
      broker.setPersistenceAdapter(kaha);
      broker.addConnector("tcp://localhost:61616");
      broker.start();
      return broker;
  }
  }

三、AMQ

• AMQ Message Store概述
      AMQ Message Store 是ActiveMQ5.0缺省的持久化存儲，它是一個基于文件、事務存儲設計為快速消息存儲的一個結構，該結構是以流的形式來進行消息交互的。
      這種方式中，Messages被保存到data logs中，同時被reference store進行索引以提高存取速度。Data logs由一些單獨的data log文件組成，缺省的文件大小是32M，如果某個消息的大小超過了data log文件的大小，那么可以修改配置以增加data log文件的大小。如果某個data log文件中所有的消息都被成功消費了，那么這個data log文件將會被標記，以便在下一輪的清理中被刪除或者歸檔。

• AMQ Message Store配置示例

  <broker brokerName="broker" persistent="true" useShutdownHook="false">
  <persistenceAdapter>
      <amqPersistenceAdapter driectory="${activemq.base}/data"    maxFileLength="32mb"/>
  </persistenceAdapter>
  </broker>

  四、 JDBC

• 使用JDBC來持久化消息（此步驟不需要手工跑腳本）
  ActiveMQ支持使用JDBC來持久化消息，預定義的表如下：
  1：消息表，缺省表明為ACTIVEMQ_MSGS，quue和topic都存在里面，結構如下：
  
  2：ACTIVEMQ_ACKS表存儲持久訂閱的信息和最后一個持久訂閱接收的消息ID，結構如下：
  
  3：鎖定表，缺省表明為ACTIVEMQ_LOCK，用來確保在某一時刻，只能有一個Act broker實例來訪問數據庫，結構如下：
  

• 使用JDBC來持久化消息的配置示例
  

注意：
（1）數據庫需要字符集設置為latin1。
（2）需要把mysql-connector-java.jar包放入lib中。
（3）啟動成功之后會出現三張表。

• 示例
  一、（queue模式）：
  
  
  運行發送者（數據庫中有三條未接收的消息）：
  
  運行接收者（消息成功接收的同時，數據庫中的消息也會被刪除）

二、persistence模式
消息接收者接收完成之后，數據庫中的消息不會被刪除。
如圖所示：

• JDBC Message Store with ActiveMQ Journal（日志）
  這種方式克服了JDBC Store的不足，使用快速的緩存寫入技術，大大提高了性能。配置示例如下：

  <beans>
  <broker brokerName="test-broker" xmlns="http://activemq.apache.org/schema/core">
      <persistenceFactory>
          <journalPersistenceAdapterFactory 
              journalLogFiles="4"
              journalLogFileSize="32768"
              useJournal="true"
              useQuickJournal="true"
              dataSource="#derby-ds"
              dataDirectory="activemq-data"/>         
      </persistenceFactory>
  </broker>
  </beans>

• JDBC Store和JDBC Message Store with ActiveMQ Journal的區別
  1：jdbc with journal的性能優于jdbc
  2：jdbc用于master/slave模式的數據庫分享
  3：jdbc with journal不能用于master/slave模式
  4：一般情況下（非集群狀態下），推薦使用jdbc with journal

五、 MMS

• Memory Message Store
      內存消息存儲主要是存儲所有的持久化的消息在內存中。這里沒有動態的緩存存在，所以必須要注意設置broker所在的JVM和內存限制。

• Memory Message Store 配置示例

  <beans>
  <broker brokerName="test-broker" persistent="false" xmlns="http://activemq.apache.org/schema/core">
      <transportConnectors uri="tcp://localhost:61616"/>
      </transportConnectors>
  </broker>
  </beans>

• 在Java中內嵌使用Broker，使用Memory的例子

  public void createEmbeddedBroker()throws Exception{
  BrokerService broker = new BrokerService();
  broker.setPersistent(false);
  broker.addConnector("tcp://localhost:61616");
  broker.start();
  }

關于“ActiveMQ中如何實現消息存儲”這篇文章就分享到這里了，希望以上內容可以對大家有一定的幫助，使各位可以學到更多知識，如果覺得文章不錯，請把它分享出去讓更多的人看到。