MQ（消息隊列）的使用場景以及常見的MQ

MQ（消息隊列）的使用場景以及常見的MQ

 

一、什么是消息中間件？

消息中間件通也被稱為消息隊列服務器，是當今分布式應用架構中經常采用的技術，程序員之間一般叫作消息隊列或者 MQ。

消息隊列屬于面向消息的中間件，使用它可以使我們的系統之間進行解耦，尤其是現在微服務架構，分布式架構中，使用消息隊列可以使我們的分布式應用之間通過發送和接收消息來進行數據交互。

消息中間件常見的角色就是 Producer（生產者）、Consumer（消費者） 、broker（消息中間件實例），消息隊列通過消息的“發送-簽收”機制來確保生產者和消費者之間數據傳輸的可靠性。

二、消息隊列發展史

世界上第一個現代消息隊列軟件 The Information Bus(TIB)，它采用了發布訂閱的模式進行數據通信，TIB 這個東西出來之后，解決了不同程序之間的通信的難題，得到了很多工程師的認可；之后 TIB 的發展引起了 IBM 的注意，于是 IBM 開發了自己的消息隊列軟件“IBM WebSphere MQ”，后來微軟發布了自己的（MSMQ），至此，各大廠商都在開發自己的消息隊列軟件。各大 IT 廠商為了保證自己的利益，建立了各種技術壁壘，源源不斷地進行商業收費，由于巨大的成本，那時消息中間件只有銀行、電信等大型企業才能使用的起。

各大廠商之間的協議不通，導致了開發成本的增加，開發人員既要懂 IBM 的 MQ ，又要懂微軟的 MQ，對開發人員來說充滿了壓力，2001 年 Sun 公司發布了 JMS （ Java Message Service）技術，目的是為了在 Java 平臺統一消息中間件的規范，只需要針對 JMS API 編程，不需要關注使用了什么樣的消息中間件。2004 年 AMQP （高級消息隊列協議）誕生了，真正促進了消息隊列的繁榮發展，任何人都可以針對 AMQP 的標準進行編碼，并和實現了 AMQP 的中間件供應商進行通信。RabbitMQ 就是基于 AMQP 協議實現的。

隨著 IT 技術的蓬勃發展，以及互聯網開源軟件的發展，加上分布式應用的迅猛發展，使得消息中間件一飛沖天，開源的消息中間件層出不窮，現在比較流行的消息中間件有 ActiveMQ、RabbitMQ、Kafak、阿里的 RocketMQ 等，在互聯網應用中消息隊列的應用基本上是個標配。

 

三、JMS消息服務
    講消息隊列就不得不提JMS 。JMS（JAVA Message Service，java消息服務）API是一個消息服務的標準/規范，JMS  只是定義了Java訪問消息中間件的接口，其實就是在包javax.jms中，你會發現這個包下除了異常定義，其他都是interface。JMS只給出接口，然后由具體的中間件去實現，比如ActiveMQ就是實現了JMS的一種Provider，還有阿里巴巴的RocketMQ。這些消息中間件都符合JMS規范。允許應用程序組件基于JavaEE平臺創建、發送、接收和讀取消息。它使分布式通信耦合度更低，消息服務更加可靠以及異步性。

說起規范，自然要定義一些術語：

• Provider/MessageProvider：生產者
• Consumer/MessageConsumer：消費者
• PTP：Point To Point，點對點通信消息模型
• Pub/Sub：Publish/Subscribe，發布訂閱消息模型
• Queue：隊列，目標類型之一，和PTP結合
• Topic：主題，目標類型之一，和Pub/Sub結合
• ConnectionFactory：連接工廠，JMS用它創建連接
• Connnection：JMS Client到JMS Provider的連接
• Destination：消息目的地，由Session創建
• Session：會話，由Connection創建，實質上就是發送、接受消息的一個線程，因此生產者、消費者都是Session創建的

 

消息模型
在JMS標準中，有兩種消息模型P2P（Point to Point）,Publish/Subscribe(Pub/Sub)。

P2P模式

P2P模式包含三個角色：消息隊列（Queue），發送者(Sender)，接收者(Receiver)。每個消息都被發送到一個特定的隊列，接收者從隊列中獲取消息。隊列保留著消息，直到他們被消費或超時。

P2P的特點

• 每個消息只有一個消費者（Consumer）(即一旦被消費，消息就不再在消息隊列中)
• 發送者和接收者之間在時間上沒有依賴性，也就是說當發送者發送了消息之后，不管接收者有沒有正在運行，它不會影響到消息被發送到隊列
• 接收者在成功接收消息之后需向隊列應答成功 
• 如果希望發送的每個消息都會被成功處理的話，那么需要P2P模式。

Pub/Sub模式

包含三個角色主題（Topic），發布者（Publisher），訂閱者（Subscriber） 多個發布者將消息發送到Topic，系統將這些消息傳遞給多個訂閱者。

Pub/Sub的特點

• 每個消息可以有多個消費者
• 發布者和訂閱者之間有時間上的依賴性。針對某個主題（Topic）的訂閱者，它必須創建一個訂閱者之后，才能消費發布者的消息
• 為了消費消息，訂閱者必須保持運行的狀態
• 為了緩和這樣嚴格的時間相關性，JMS允許訂閱者創建一個可持久化的訂閱。這樣，即使訂閱者沒有被激活（運行），它也能接收到發布者的消息。
• 如果希望發送的消息可以不被做任何處理、或者只被一個消息者處理、或者可以被多個消費者處理的話，那么可以采用Pub/Sub模型。

消息消費
在JMS中，消息的產生和消費都是異步的。對于消費來說，JMS的消息者可以通過兩種方式來消費消息。
（1）同步
訂閱者或接收者通過receive方法來接收消息，receive方法在接收到消息之前（或超時之前）將一直阻塞；

（2）異步
訂閱者或接收者可以注冊為一個消息監聽器。當消息到達之后，系統自動調用監聽器的onMessage方法。

JMS 消息結構組成

消息（Message）包括消息頭和消息體以及其它的擴展屬性，消息頭中包含代理路由和管理消息所需的信息，消息體用來保存需要傳輸的數據，屬性由客戶端程序定義。

（1）消息頭（Header）

消息頭包含消息的識別信息和路由信息，每個消息頭字段都有相應的 getter 和 setter 方法，所有消息類型的這部分格式都是一樣的。

消息頭包含一些標準的屬性如下：

屬性名稱	說明
JMSCorrelationID	關聯的消息 ID，這個通常用在需要回傳消息的時候
JMSDeliveryMode	消息的發送模式，分為持久化和非持久化模式
JMSDestination	消息發送的目的地，是一個 Topic 或 Queue
JMSExpiration	消息的過期時間
JMSMessageID	消息 ID，需要以 ID:開頭
JMSPriority	優先級分為 10 個級別， 范圍從 0（最低）到 9（最高） ，默認級別是 4
JMSRedelivered	消息是否重復發送過
JMSReplyTo	消息回復的目的地，是一個 Topic 或 Queue
JMSTimestamp	消息發送時的時間

（2）消息體（Body）

消息體就是我們需要傳遞的消息內容，JMS 定義了五種不同的消息格式，以及調用的消息類型 ：

消息類型	說明
TextMessage	java.lang.String 對象，比如：xml 或者 json 格式
MapMessage	鍵值對的集合，鍵是 String 對象，值類型可以是 Java 任何基本類型
BytesMessage	字節流的消息
StreamMessage	Java 中原始的輸入輸出流
ObjectMessage	Java 中的可序列化對象

（3）屬性（Properties）

如果需要除消息頭字段以外的值，那么可以使用消息屬性，有三種類型的屬性：分為自定義屬性，JMS 定義的屬性和提供者特定的屬性 。

• 自定義屬性

可以使用任何 Java 類型來自由定義任何屬性，如下：

發送者可以使用 set 方法設置屬性，消費者可以使用 get 方法獲取發送者設置屬性的值。

• JMS 定義的屬性

JMS 規范為 JMS 定義的屬性，保留 JMSX 屬性名稱前綴，這些屬性是可選的：

屬性名稱	說明
JMSXAppID	標識發送消息的應用程序
JMSXConsumerTXID	使用此消息的事務的事務標識符
JMSXDeliveryCount	消息重新傳遞的次數
JMSXGroupID	此消息所屬的消息組
JMSXGroupSeq	消息組中此消息的序列號
JMSXProducerTXID	生成此消息的事務的事務標識符
JMSXRcvTimestamp	JMS provider 將消息傳遞給消費者的時間
JMSXState	用于定義 provider 特定的狀態
JMSXUserID	標識發送消息的用戶

只有 JMSXGroupID 和 JMSXGroupSeq 需要所有 JMS 提供者的支持，也就是我們經常說的消息分組。

 

 

四、消息隊列應用場景
以下介紹消息隊列在實際應用中常用的使用場景。應用解耦，異步處理，流量削鋒和消息通訊四個場景。實現高性能，高可用，可伸縮和最終一致性架構。目前使用較多的消息隊列有ActiveMQ，RabbitMQ，ZeroMQ，Kafka，MetaMQ，RocketMQ

異步處理
場景說明：用戶注冊后，需要發注冊郵件和注冊短信。傳統的做法有兩種 1.串行的方式； 2.并行方式
a、串行方式：將注冊信息寫入數據庫成功后，發送注冊郵件，再發送注冊短信。以上三個任務全部完成后，返回給客戶端。

 

b、并行方式：將注冊信息寫入數據庫成功后，發送注冊郵件的同時，發送注冊短信。以上三個任務完成后，返回給客戶端。與串行的差別是，并行的方式可以提高處理的時間

 

假設三個業務節點每個使用50毫秒鐘，不考慮網絡等其他開銷，則串行方式的時間是150毫秒，并行的時間可能是100毫秒。
因為CPU在單位時間內處理的請求數是一定的，假設CPU1秒內吞吐量是100次。則串行方式1秒內CPU可處理的請求量是7次（1000/150）。并行方式處理的請求量是10次（1000/100）
小結：如以上案例描述，傳統的方式系統的性能（并發量，吞吐量，響應時間）會有瓶頸。如何解決這個問題呢？

 

引入消息隊列，將不是必須的業務邏輯，異步處理。改造后的架構如下：

按照以上約定，用戶的響應時間相當于是注冊信息寫入數據庫的時間，也就是50毫秒。注冊郵件，發送短信寫入消息隊列后，直接返回，因此寫入消息隊列的速度很快，基本可以忽略，因此用戶的響應時間可能是50毫秒。因此架構改變后，系統的吞吐量提高到每秒20 QPS。比串行提高了3倍，比并行提高了兩倍。

 

應用解耦
場景說明：用戶下單后，訂單系統需要通知庫存系統。傳統的做法是，訂單系統調用庫存系統的接口。如下圖：

傳統模式的缺點：假如庫存系統無法訪問，則訂單減庫存將失敗，從而導致訂單失敗，訂單系統與庫存系統耦合

 

如何解決以上問題呢？引入應用消息隊列后的方案，如下圖：

訂單系統：  用戶下單后，訂單系統完成持久化處理，將消息寫入消息隊列，返回用戶訂單下單成功
庫存系統：  訂閱下單的消息，采用拉/推的方式，獲取下單信息，庫存系統根據下單信息，進行庫存操作
假如：在下單時庫存系統不能正常使用。也不影響正常下單，因為下單后，訂單系統寫入消息隊列就不再關心其他的后續操作了。實現訂單系統與庫存系統的應用解耦

 

流量削鋒
流量削鋒也是消息隊列中的常用場景，一般在秒殺或團搶活動中使用廣泛。
應用場景：秒殺活動，一般會因為流量過大，導致流量暴增，應用掛掉。為解決這個問題，一般需要在應用前端加入消息隊列。
a、可以控制活動的人數
b、可以緩解短時間內高流量壓垮應用

用戶的請求，服務器接收后，首先寫入消息隊列。假如消息隊列長度超過最大數量，則直接拋棄用戶請求或跳轉到錯誤頁面。
秒殺業務根據消息隊列中的請求信息，再做后續處理

 

日志處理
日志處理是指將消息隊列用在日志處理中，比如Kafka的應用，解決大量日志傳輸的問題。架構簡化如下

日志采集客戶端，負責日志數據采集，定時寫受寫入Kafka隊列
Kafka消息隊列，負責日志數據的接收，存儲和轉發
日志處理應用：訂閱并消費kafka隊列中的日志數據 

 

消息通訊
消息通訊是指，消息隊列一般都內置了高效的通信機制，因此也可以用在純的消息通訊。比如實現點對點消息隊列，或者聊天室等
點對點通訊：

客戶端A和客戶端B使用同一隊列，進行消息通訊。

 

聊天室通訊：

客戶端A，客戶端B，客戶端N訂閱同一主題，進行消息發布和接收。實現類似聊天室效果。

以上實際是消息隊列的兩種消息模式，點對點或發布訂閱模式。模型為示意圖，供參考。

 

五、消息中間件示例 
電商系統

消息隊列采用高可用，可持久化的消息中間件。比如Active MQ，Rabbit MQ，Rocket Mq。
（1）應用將主干邏輯處理完成后，寫入消息隊列。消息發送是否成功可以開啟消息的確認模式。（消息隊列返回消息接收成功狀態后，應用再返回，這樣保障消息的完整性）
（2）擴展流程（發短信，配送處理）訂閱隊列消息。采用推或拉的方式獲取消息并處理。
（3）消息將應用解耦的同時，帶來了數據一致性問題，可以采用最終一致性方式解決。比如主數據寫入數據庫，擴展應用根據消息隊列，并結合數據庫方式實現基于消息隊列的后續處理。

 

日志收集系統

分為Zookeeper注冊中心，日志收集客戶端，Kafka集群和Storm集群（OtherApp）四部分組成。
Zookeeper注冊中心，提出負載均衡和地址查找服務
日志收集客戶端，用于采集應用系統的日志，并將數據推送到kafka隊列
Kafka集群：接收，路由，存儲，轉發等消息處理
Storm集群：與OtherApp處于同一級別，采用拉的方式消費隊列中的數據

 

六、常用消息隊列

一般商用的容器，比如WebLogic，JBoss，都支持JMS標準，開發上很方便。但免費的比如Tomcat，Jetty等則需要使用第三方的消息中間件。本部分內容介紹常用的消息中間件（Active MQ,Rabbit MQ，Zero MQ,Kafka）以及他們的特點。

 ActiveMQ
ActiveMQ 是Apache出品，最流行的，能力強勁的開源消息總線。ActiveMQ 是一個完全支持JMS1.1和J2EE 1.4規范的 JMS Provider實現，盡管JMS規范出臺已經是很久的事情了，但是JMS在當今的J2EE應用中間仍然扮演著特殊的地位。

ActiveMQ特性如下：

• ⒈ 多種語言和協議編寫客戶端。語言: Java,C,C++,C#,Ruby,Perl,Python,PHP。應用協議： OpenWire,Stomp REST,WS Notification,XMPP,AMQP
• ⒉ 完全支持JMS1.1和J2EE 1.4規范 （持久化，XA消息，事務)
• ⒊ 對Spring的支持，ActiveMQ可以很容易內嵌到使用Spring的系統里面去，而且也支持Spring2.0的特性
• ⒋ 通過了常見J2EE服務器（如 Geronimo,JBoss 4,GlassFish,WebLogic)的測試，其中通過JCA 1.5 resource adaptors的配置，可以讓ActiveMQ可以自動的部署到任何兼容J2EE 1.4 商業服務器上
• ⒌ 支持多種傳送協議：in-VM,TCP,SSL,NIO,UDP,JGroups,JXTA
• ⒍ 支持通過JDBC和journal提供高速的消息持久化
• ⒎ 從設計上保證了高性能的集群，客戶端-服務器，點對點
• ⒏ 支持Ajax
• ⒐ 支持與Axis的整合
• ⒑ 可以很容易得調用內嵌JMS provider，進行測試

 Kafka
     Kafka是一種高吞吐量的分布式發布訂閱消息系統，它可以處理消費者規模的網站中的所有動作流數據。 這種動作（網頁瀏覽，搜索和其他用戶的行動）是在現代網絡上的許多社會功能的一個關鍵因素。 這些數據通常是由于吞吐量的要求而通過處理日志和日志聚合來解決。 對于像Hadoop的一樣的日志數據和離線分析系統，但又要求實時處理的限制，這是一個可行的解決方案。Kafka的目的是通過Hadoop的并行加載機制來統一線上和離線的消息處理，也是為了通過集群機來提供實時的消費。

Kafka是一種高吞吐量的分布式發布訂閱消息系統，有如下特性：
通過O(1)的磁盤數據結構提供消息的持久化，這種結構對于即使數以TB的消息存儲也能夠保持長時間的穩定性能。（文件追加的方式寫入數據，過期的數據定期刪除）
高吞吐量：即使是非常普通的硬件Kafka也可以支持每秒數百萬的消息
支持通過Kafka服務器和消費機集群來分區消息
支持Hadoop并行數據加載
Kafka相關概念
Broker
Kafka集群包含一個或多個服務器，這種服務器被稱為broker[5]
Topic
每條發布到Kafka集群的消息都有一個類別，這個類別被稱為Topic。（物理上不同Topic的消息分開存儲，邏輯上一個Topic的消息雖然保存于一個或多個broker上但用戶只需指定消息的Topic即可生產或消費數據而不必關心數據存于何處）
Partition
Parition是物理上的概念，每個Topic包含一個或多個Partition.
Producer
負責發布消息到Kafka broker
Consumer
消息消費者，向Kafka broker讀取消息的客戶端。
Consumer Group
每個Consumer屬于一個特定的Consumer Group（可為每個Consumer指定group name，若不指定group name則屬于默認的group）。
一般應用在大數據日志處理或對實時性（少量延遲），可靠性（少量丟數據）要求稍低的場景使用。