JMS與ActiveMQ消息數據持久化

• 認識JMS
  JMS是Java消息服務（Java message service），是java平臺關於面向消息中間件（MOM）的API。用於兩個應用程序或分佈式系統應用中，消息發送，消息異步處理。即應用程序間通訊經過JMS服務，進行消息轉發。應用程序間消息異步處理可解除應用程序的耦合。
  JMS有消息異步處理、程序間解耦、消息可靠的優點。

• JMS消息模型
  Queue點對點（Point to Point）
  Queue隊列是點對點消費，發送者發送一條消息，只有惟一的一個消費者能對消息進行消費。消息生產者都將消息發送到消息的隊列（Queue）中。隊列的消息能夠是持久的，保證消息服務出現故障仍然可以傳遞消息。
  特色：
  1).每一個消息只有一個消費者（Customer），消息一旦被消費，消息就不在隊列（Queue）了。
  2).消息發送者和接收者沒有時間依賴，消息發送者只管消息發送，無論消息的消費者是否有接收消息。
  3).消息被接收到消息後，會發送消息確認（ACK）通知給消息隊列（Queue）。
  Queue模型圖：
  
  發佈/訂閱（Publish/Subscribe）
  消息發佈者發佈消息，消息經過主題（Topic）傳遞給全部接收者，消息發佈者和訂閱者彼此不相干。主題（Topic）主要用於保存和傳遞消息。
  發佈/訂閱模型中，應用程序有Topic、發佈者（Publish）、訂閱者（Subscribe）組成。
  特色：
  1).每一個消息可有多個消費訂閱者。
  2).發佈者和訂閱者無時間依賴性。某個主題（Topic）的訂閱者，必須先建立一個訂閱者後才能消費發佈者的消息，且爲消費消息，訂閱者必須保持運行狀態。
  3).可持久化訂閱。
  Topic模型圖：
  

• ActiveMQ消息數據持久化
  爲了不機器意外宕機，消息丟失。可進行對消息持久化操做，機器重啓後恢復消息隊列。ActiveMQ消息持久化機制有JDBC、AMQ、KahaDB和LevelDB。消息發送者將消息發送出去後，消息中心將消息數據存儲到本地文件、本地內存或者數據庫等。消息持久化操做後，消息中心啓動後會先檢查指定存儲位置，若有未發送成功的消息，則繼續將消息發送出去。
  1).JDBC持久化方式
  此持久化方式會在數據庫建立3個表：activemq_msgs、activemq_acks、activemq_lock。activemq_msgs用於存儲消息，Queue和Topic都存儲於這個表中。
  配置方式：
  修改安裝目錄下的conf/activemq.xml文件。persistenceAdapter節點中配置。

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

  2).AMQ方式
  性能高於JDBC，消息會按順序追加方式寫入日誌文件中，性能較高。爲了提高性能，會建立消息主鍵索引。缺點是索引文件很大，需佔用大量磁盤空間。若是broker崩潰，重建索引速度很是慢。每一個日誌文件大小有限定（默認32M）。超過此大小，會從新創建一文件。當全部消息消費完成，系統刪除這個文件或進行規定（取決於配置）。
  配置：

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

  索引重建時間長，佔用磁盤空間大，此方式不推薦。
  3).KafaDB方式
  KafaDB持久化是ActiveMQ默認的持久化方式。KafaDB持久化和AMQ同樣都是基於日誌文件，但KafaDB方式恢復時間遠少於AMQ方式，且使用更少的數據文件。優於AMQ方式持久化。
  配置：

  </persistenceAdapter>
  <kahaDB directory="${activemq.data}/activemq-data" journalMaxFileLength="16mb"/>
  </persistenceAdapter>

  directory：指定消息持久化的存儲目錄。
  journalMaxFileLength：指定保存消息日誌文件大小。
  4).LevelDB方式
  ActiveMQ5.6版本後推出的LevelDB方式持久化。不過LevelDB方式性能要高於KafaDB，後面極可能是這個趨勢。
  配置：

  </persistenceAdapter>
  <replicatedLevelDB directory="${activemq.data}/activemq-data" replicas="3" bind="tcp://0.0.0.0:0" zkAddress="127.0.0.1:2181" zkSessionTimeout="2s" hostname="127.0.0.1" zkPath="/activemq/leveldb-stores"/>
  </persistenceAdapter>