消息隊列全面了解(二)


消息隊列都應用到了哪些實際的應用場景中?

一、再談消息隊列的應用場景

1、異步處理:例如短信通知、終端狀態推送、App推送、用戶注冊等

2、數據同部:業務數據推送同步

3、重試補償:記賬失敗重試

4、系統解耦:通訊上下行、終端異常監控、分布式事件中心

5、流量削峰:秒殺場景下的下單處理

6、發布訂閱:HSF的服務狀態變化通知、分布式事件中心

7、高並發緩沖:日志服務、監控上報

但是,我們對消息隊列的底層技術和原理還是不了解,那么我們馬上開始吧。

二、消息隊列的一些基本概念和簡單原理

1、Broker

Broker的概念來自於Apache ActiveMQ,通俗的講就是MQ的服務器。

2、消息的生產者、消費者

消息生產者Producer:發送消息到消息隊列。

消息消費者Consumer:從消息隊列接收消息。

 

 3、點對點消息隊列模型

消息生產者向一個特定的隊列發送消息,消息消費者從該隊列中接收消息;消息的生產者和消費者可以不同時處於運行狀態。每個成功處理的消息都由消息消費者簽收確認(Acknowledge)。如圖:

 

 4、發布訂閱消息模型-Topic

發布訂閱消息模型中,支持向一個特定的主題Topic發布消息,0個或多個訂閱者接收來自這個主題的消息。這種情模型下,發布者和訂閱者閉此不知道對方。實際操作過程中,必須先訂閱,再發送消息,而后接收訂閱的消息,這個順序必須保證。

 

 

5、消息的順序性保證

基於Queue消息模型,利用FIFO先進先出的特性,可以保證消息的順序性。

6、消息的ACK確認機制

即消息的Acknowledge確認機制

為了確保消息不丟失,消息隊列提供了消息Acknowledge機制,即ACK機制,當Consumer確認消息已經被消費者處理,發送一個ACK給消息隊列,此時消息隊列便可以刪除這個 消息了。如果Consumer宕機/關閉,沒有發送ACK,消息隊列講人為這個消息沒有被處理,會將這個消息發送給其他的Consumer重新消息處理。

7、消息的持久化

消息的持久化,對於一些關鍵的核心業務來說是非常重要的,啟用消息持久化后,消息隊列宕機重啟后,消息而已從持久化存儲恢復,消息不丟失,可以繼續消費處理。

8、消息的同部和異步收發

同部:消息的收發支持同步收發的方式。

同時還有另一種同步方式:同步收發場景下,消息生產者和消費者雙向應答模式,例如:張三寫封信松道郵局中轉站,然后李四從中轉站獲得信,然后在寫一份回執信,放到中轉站,然后張三去取,當然張三寫信的時候就得寫明回信地址。

消息的接收如果以同步的方式(Pull)進行接收,如哦隊列中為空,此時接收處於同步阻塞狀態。會一直等待,直到消息的到達。

異步:消費的收發同樣支持異步方式:異步發送 消息,不需要等待消息隊列的接收確認;異步接收消息,以Push的方式觸發消息消費者接收消息。

9、消息的事務支持

消息的收發處理支持事務,例如:在任務中心場景中,一次處理可能處理涉及多個消息的接收、處理,這處於同一個事務范圍內,如果一個消息處理失敗,事務回滾,消息重新回到隊列中。

三、我們對消息隊列的實際使用

我在實際的項目中,使用過兩種消息隊列組件:

RabbitMQ:高可用、高可靠消息應用場景,例如記賬失敗重試、通知服務,消息不允許丟

Kafka:高性能消息應用場景,例如日志、監控、消息允許丟失。

在此之上,我們封裝了消息應用中心,日志服務等核心組件和服務,那么,消息應用中心和日志都用到了消息隊列什么技術?干活來了……

1、消息應用中心

消息應用中心(任務中心)使用了消息隊列的異步處理、數據同步、重試補償、系統解耦、流量削峰等特性。其中:消息應用中心(任務中心),支持RabbitMQ和Kafka兩種消息通道,支持在任務元數據層面設置

任務:就是一個包含了任務執行上下文的消息,同時代表了異步處理

任務發送者(ITaskSender)發送任務:消息的生產者將任務消息發送到消息隊列

任務類型:消息隊列名稱,例如:HaKeepAcco***Queue,充電補償記賬隊列

消息隊列:任務的臨時存儲

任務中心:任務計中處理,消息消費者

任務處理完成:消息Ack確認

任務的多級重試:多個重試消息隊列,HaSysTaskStore2Queue

2、日志組件

日志組件,使用了消息隊列的高並發緩沖和發布訂閱特性。其中:日志組件使用Kafka作為消息通道,因為Kafka的性能號,吞吐量大,可以容忍偶爾的消息數據丟失,日志組件使用發布訂閱的消息模型,日志組件包含日志服務SDK和日志HSF服務,二者都是消息的生產者Producer,日志類型:消息的Topic主題。日志處理器:消息的消費者、Topic的訂閱、日志數據處理(Hbase\ES\其他)

3、RPC服務狀態變化通知

RPC服務狀態變化通知,使用了消息隊列的發布訂閱特性,其中:RPC服務狀態變化通知,使用RobbitMQ消息隊列技術,使用發布訂閱的消息模型,Topic:RPCServiceState。RPCService.Proxy:RPC服務狀態變化消息的訂閱者。RPC服務注冊、發布:消息的生產者,發送RPC服務狀態變化消息。

四、消息隊列使用的最佳實踐

1、RabbitMQ的鏈接,底層都是Socket鏈接,長連接 or 短鏈接?

RabbitMQ在創建每個鏈接的同時,會自動創建一個監控線程來定時(默認60s)偵測鏈接的狀態,如果鏈接斷開,觸發ConnectionShutdown事件。

用長連接,還是用短連接?

發送端:建議使用短連接,用完即釋放,避免長連接帶來的端口占用,因為發送端無處不在,發送操作短而急促。

接收端:建議使用長連接,時刻接收處理消息,因為消息的接收消費比較集中,接收操作久而彌堅。

2、網絡是有抖動的,連接的斷開是正常的,如何應對?

發送端:發送失敗重試

接收端:注冊ConnectionShutdown事件同時捕獲消息接收異常,重新建立連接,接收消費消息。

3、RabbitMQ Exchange(Topic)模式下帶來的消息隊列數量激增

只是創建了一個Exchange(Topic),為什么會增加這么多Queue。以為,每個Topic的訂閱都是綁定一個Queue用作消息的消費。

4、需求的演變,消息結構的改變,如何平滑過渡?

消息是byte[]數組,我們將復雜對象消息二進制序列化。接收到消息后,我們將二進制數組反序列化為實體類。當我們的實體類消息體的結構發生變化后,因為受到二進制學歷惡化處理的影響,導致無法反序列化。

解決方案:

消息體預留一些string類型的擴展字段。

消息隊列版本化,支持多個版本的消息體。

5、Kafka Conusmer Group

同一個Topic的一條消息只能被同一個Group內的一個Consumer消費,多個Consumer Group可同時消費同一個消息。

 6、消息的擠壓

消息的擠壓產生的原因:消息接收消費的速率低,發送的速度>接收的速度。

消息積壓后得影響:

消息大量積壓后,當新得消費者連接上MQ並開始接收消息時,發送速率會大幅降低。消息隊列集群得壓力增加,大量得消息要持久化存儲和同步。

如何減少消息擠壓:快速消費消息,同時保持消息體不要過大。

這次的MQ相關知識先分享到這里。


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM