分佈式消息通訊-ActiveMQ-JMS-筆記

消息中間件的初步認識

• 什麼是消息中間件？

  • 消息中間件是指利用高效可靠的消息傳遞機制進行平臺無關的數據交流，並基於數據通訊來進行分佈式系統的集成。

  • 經過提供消息傳遞和消息排隊模型，能夠在分佈式架構下擴展進程之間的通訊。

• 消息中間件能作什麼？

  • 消息中間件主要解決的就是分佈式系統之間消息傳遞的問題，它可以屏蔽各類平臺以及協議之間的特性，實現應用程序之間的協同。

咱們從註冊這個服務能夠看到，每個子操做都是相對獨立的，

• 同時，基於領域劃分之後，發送激活郵件、發送營銷短信、贈送積分及紅包都屬於不一樣的子域。
• 因此咱們能夠對這些子操做進行來實現異步化執行，相似於多線程並行處理的概念。
• 如何實現異步化呢？
  • 用多線程能實現嗎？多線程固然能夠實現，只是，消息的持久化、消息的重發這些條件，多線程並不能知足。
  • 因此須要藉助一些開源中間件來解決。
• 而分佈式消息隊列就是一個很是好的解決辦法，引入分佈式消息隊列之後，架構圖就變成這樣了（下圖是異步消息隊列的場景）。
  • 經過引入分佈式隊列，就可以大大提高程序的處理效率，而且還解決了各個模塊之間的耦合問題
  • 這個是分佈式消息隊列的第一個解決場景【異步處理】
•  

咱們再來展開一種場景，

• 經過分佈式消息隊列來實現流量整形，
• 好比在電商平臺的秒殺場景下，流量會很是大。
• 經過消息隊列的方式能夠很好的緩解高流量的問題

• 用戶提交過來的請求，先寫入到消息隊列。
  • 消息隊列是有長度的，若是消息隊列長度超過指定長度，直接拋棄
• 秒殺的具體核心處理業務，
  • 接收消息隊列中消息進行處理，這裏的消息處理能力取決於消費端自己的吞吐量

固然，消息中間件還有更多應用場景，

• 好比在弱一致性事務模型中，能夠採用分佈式消息隊列的實現最大能力通知方式來實現數據的最終一致性等等

ActiveMQ 簡介

• ActiveMQ 是徹底基於 JMS 規範實現的一個消息中間件產品。
• 是 Apache 開源基金會研發的消息中間件。
• ActiveMQ 主要應用在分佈式系統架構中，幫助構建高可用、高性能、可伸縮的企業級面向消息服務的系統

ActiveMQ 特性

• 多語言和協議編寫客戶端 語言：java/C/C++/C#/Ruby/Perl/Python/PHP
  • 應 用 協 議 : openwire/stomp/REST/ws/notification/XMPP/AMQP
• 徹底支持 jms1.1 和 J2ee1.4 規範
• 對 spring 的支持，ActiveMQ 能夠很容易內嵌到 spring 模塊中

從 JMS 規範來了解 ActiveMQ

JMS 定義

• Java 消息服務（Java Message Service）是 java 平臺中關於面向消息中間件的 API，
• 用於在兩個應用程序之間，或者分佈式系統中發送消息，進行異步通訊。
• JMS 是一個與具體平臺無關的 API，
• 絕大多數 MOM（Message Oriented Middleware）（面向消息中間件）提供商都對 JMS 提供了支持。
• ActiveMQ 就是其中一個實現

什麼是 MOM

• MOM 是面向消息的中間件，使用消息傳送提供者來協調消息傳送操做。
• MOM 須要提供 API 和管理工具。
• 客戶端使用 api 調用，把消息發送到由提供者管理的目的地。
• 在發送消息以後，客戶端會繼續執行其餘工做，而且在接收方收到這個消息確認以前，提供者一直保留該消息。

MOM 的特色

• 消息異步接收，發送者不須要等待消息接受者響應
• 消息可靠接收，確保消息在中間件可靠保存。
• 只有接收方收到後才刪除消息

Java 消息傳送服務規範最初的開發目的是爲了使 Java 應用程序可以訪問現有 MOM 系統。

• 引入該規範以後，它已被許多現有的 MOM 供應商採用而且已經憑藉自身的功能實現爲異步消息傳送系統。

JMS 規範

• JMS 規範的目的
  • 是爲了使得 Java 應用程序可以訪問現有 MOM (消息中間件)系統，
  • 造成一套統一的標準規範，解決不一樣消息中間件之間的協做問題。

在建立 JMS 規範時，設計者但願可以結合現有的消息傳送的精髓，

• 好比說：
  • 不一樣的消息傳送模式或域，例如點對點消息傳送和發佈/訂閱消息傳送
  • 提供於接收同步和異步消息的工具
  • 對可靠消息傳送的支持
  • 常見消息格式，例如流、文本和字節

JMS 的體系結構

經過 JMS 規範結合 ActiveMQ 實現消息發送案例

• 案例總結這個案例的架構圖以下

細化 JMS 的基本功能

消息傳遞域

• JMS 規範中定義了兩種消息傳遞域：
  • 點對點（point-topoint ） 消 息 傳 遞 域
  •  發 布 / 訂 閱 消 息 傳 遞 域(publish/subscribe)
• 簡單理解就是：
  • 有點相似於咱們經過 qq 聊天的時候，在羣裏面發消息和給其中一個同窗私聊消息。
    • 在羣裏發消息，全部羣成員都能收到消息。
    • 私聊消息只能被私聊的學員能收到消息
• 點對點消息傳遞域
  • 每一個消息只能有一個消費者
  • 消息的生產者和消費者之間沒有時間上的相關性。
  • 不管消費者在生產者發送消息的時候是否處於運行狀態，均可以提取消息

• 發佈訂閱消息傳遞域
  • 每一個消息能夠有多個消費者
  • 生產者和消費者之間有時間上的相關性。
    • 訂閱一個主題的消費者只能消費自它訂閱以後發佈的消息。
    • JMS 規範容許客戶建立持久訂閱，這在必定程度上下降了時間上的相關性要求。
    • 持久訂閱容許消費者消費它在未處於激活狀態時發送的消息

消息結構組成

• JMS 消息由這幾部分組成：消息頭、屬性、消息體

消息頭

• 消息頭(Header) - 消息頭包含消息的識別信息和路由信息，消息頭包含一些標準的屬性如：
  • JMSDestination    消息發送的目的地，queue 或者 topic)
  • JMSDeliveryMode  傳送模式。持久模式和非持久模式
  • JMSPriority     消息優先級
    • 優先級分爲 10 個級別，從 0(最低)到 9(最高). 若是不設定優先級，默認級別是 4。
    • 須要注意的是，JMS provider 並不必定保證按照優先級的順序提交消息
  • JMSMessageID       惟一識別每一個消息的標識

屬性

• 按類型能夠分爲：
  • 應用設置的屬性，標準屬性和消息中間件定義的屬性 

• 應用程序設置和添加的屬性，好比

  • Message.setStringProperty(「key」,」value」);

  • 經過下面的代碼能夠得到自定義屬性:

    • 在發送端，定義消息屬性:   

      •  message.setStringProperty("Mic","Hello World");

    • 在接收端接收數據:   

      • Enumeration enumeration=message.getPropertyNames(); 
        while(enumeration.hasMoreElements()){     
             String name=enumeration.nextElement().toString(); 
            
             System.out.println("name:"+name+":"+messag
             e.getStringProperty(name)); 
             System.out.println(); 
        }

         

         

          

         

• JMS 定義的屬性:

  • 使用「JMSX」做爲屬性名的前綴，經過下面這段代碼能夠返回全部鏈接支持的 JMSX 屬性的名字

• JMS provider 特定的屬性

消息體

• 就是咱們須要傳遞的消息內容，
• JMS API 定義了 5 中消息體格式，
• 可使用不一樣形式發送接收數據，並能夠兼容現有的消息格式，其中包括 :

TextMessage	java.lang.String 對象，如 xml 文件內容
MapMessage	名/值對的集合，名是 String 對象，值類型能夠是 Java 任何基本類型
BytesMessage	字節流
StreamMessage	Java 中的輸入輸出流
ObjectMessage	Java 中的可序列化對象
Message	沒有消息體，只有消息頭和屬性。

• 絕大部分的時候，咱們只須要基於消息體進行構造

持久訂閱

• 持久訂閱的概念，也很容易理解，好比仍是以 QQ 爲例，咱們把 QQ 退出了，可是下次登陸的時候，仍然能收到離線的消息。

持久訂閱就是這樣一個道理，持久訂閱有兩個特色：

1. 持久訂閱者和非持久訂閱者針對的 Domain 是 Pub/Sub，而不是 P2P
2. 當 Broker 發送消息給訂閱者時，
   • 若是訂閱者處於 未激活狀態狀態：持久訂閱者能夠收到消息，而非持久訂閱者則收不到消息。
   • 固然這種方式也有必定的影響：
     • 當持久訂閱者處於 未激活狀態時，Broker 須要爲持久訂閱者保存消息；
     • 若是持久訂閱者訂閱的消息太多則會溢出。

消費端改動

connection=connectionFactory.createConnection(); 
connection.setClientID("Mic-001"); connection.start(); 
Session session=connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE); 
Topic destination=session.createTopic("myTopic");
MessageConsumer consumer=session.createDurableSubscriber(des tination,"Mic-001"); 
TextMessage message=(TextMessage)consumer.receive(); 
System.out.println(message.getText());

• 修改三處地方，而後先啓動消費端去註冊一個持久訂閱。
• 持久訂閱時，客戶端向 JMS 服務器註冊一個本身身份的 ID，
  • 當這個客戶端處於離線時，JMS Provider 會爲這個 ID 保存全部發送到主題的消息，
  • 當客戶再次鏈接到 JMS Provider 時，會根據本身的 ID 獲得全部當本身處於離線時發送到主題的消息。
• 這個身份 ID，在代碼中的體現就是 connection 的 ClientID，
  • 這個其實很好理解，你要想收到朋友發送的 qq 消息，前提就是你得先註冊個 QQ 號，並且還要有臺能上網的設備，電腦或手機。
  • 設備就至關因而 clientId 是惟一的；
  • qq 號至關因而訂閱者的名稱，在同一臺設備上，不能用同一個 qq 號掛 2 個客戶端。
  • 鏈接的 clientId 必須是惟一的，訂閱者的名稱在同一個鏈接內必須惟一。
  • 這樣才能惟一的肯定鏈接和訂閱者。

activeMQ 控制檯的截圖

• 設置持久訂閱之後，在控制檯能看到下圖的變化

JMS 消息的可靠性機制

• 理論上來講，咱們須要保證消息中間件上的消息，只有被消費者確認過之後纔會被簽收，
  • 至關於咱們寄一個快遞出去，收件人沒有收到快遞，就認爲這個包裹仍是屬於待簽收狀態，這樣才能保證包裹可以安全達到收件人手裏。
  • 消息中間件也是同樣。
• 消息的消費一般包含 3 個階段：客戶接收消息、客戶處理消息、消息被確認
• 首先，來簡單瞭解 JMS 的事務性會話和非事務性會話的概念
  • ​​​​​​​JMS Session 接口提供了 commit 和 rollback 方法。
  • 事務提交意味着生產的全部消息被髮送，消費的全部消息被確認；
  • 事務回滾意味着生產的全部消息被銷燬，消費的全部消息被恢復並從新提交，除非它們已通過期。
  • 事務性的會話老是牽涉到事務處理中，commit 或 rollback 方法一旦被調用，一個事務就結束了，而另外一個事務被開始。
  • 關閉事務性會話將回滾其中的事務

在事務型會話中

• 在事務狀態下進行發送操做，消息並未真正投遞到中間件，
• 而只有進行 session.commit 操做以後，消息纔會發送到中間件，再轉發到適當的消費者進行處理。
• 若是是調用 rollback 操做，則代表，當前事務期間內所發送的消息都取消掉。
• 經過在建立 session 的時候使用 true or false 來決定當前的會話是事務性仍是非事務性

• connection.createSession(Boolean.TRUE,Session.AUTO_ACKNOWLEDGE);

   在事務性會話中，消息的確認是自動進行，也就是經過 session.commit()之後，消息會自動確認。

• 必須保證發送端和接收端都是事務性會話

在非事務型會話中

• 消 息 何 時 被 確 認 取 決 於 創 建 會 話 時 的 應 答 模 式(acknowledgement mode).
• 有三個可選項
  • Session.AUTO_ACKNOWLEDGE
    • 當 客 戶 成 功 的 從 receive 方 法 返 回 的 時 候 ，
    • 或 者 從MessageListenner.onMessage 方法成功返回的時候，會話自動確認客戶收到消息。
  • Session.CLIENT_ACKNOWLEDGE
    • ​​​​​​​客戶經過調用消息的 acknowledge 方法確認消息。
    • CLIENT_ACKNOWLEDGE 特性
      • ​​​​​​​在這種模式中，確認是在會話層上進行，
      • 確認一個被消費的消息將自動確認全部已被會話消費的消息。
      • 列如，若是一個消息消費者消費了 10 個消息，而後確認了第 5 個消息，那麼 0~5 的消息都會被確認 
  • Session.DUPS_ACKNOWLEDGE
    • ​​​​​​​消息延遲確認。
    • 指定消息提供者在消息接收者沒有確認發送時從新發送消息，這種模式不在意接受者收到重複的消息。

消息的持久化存儲

• 消息的持久化存儲也是保證可靠性最重要的機制之一，
• 也就是消息發送到 Broker 上之後，若是 broker 出現故障宕機了，那麼存儲在 broker 上的消息不該該丟失。
• 能夠經過下面的代碼來設置消息發送端的持久化和非持久化特性

MessageProducer producer=session.createProducer(destination); 
producer.setDeliveryMode(DeliveryMode.PERSISTENT);

• 對於非持久的消息，JMS provider 不會將它存到文件/數據庫等穩定的存儲介質中。
  • 也就是說非持久消息駐留在內存中，若是 jms provider 宕機，那麼內存中的非持久消息會丟失
• 對於持久消息，消息提供者會使用存儲-轉發機制，
  • 先將消息存儲到穩定介質中，等消息發送成功後再刪除。
  • 若是 jms provider 掛掉了，那麼這些未送達的消息不會丟失；
  • jms provider 恢復正常後，會從新讀取這些消息，並傳送給對應的消費者。