一、消息積壓的原因
消息積壓的直接原因,一定是系統中某個部分出現了性能問題,來不及處理上游發送的消息,才會導致消息積壓。
二、優化性能來避免消息積壓
在使用消息隊列的系統中,對於性能的優化,主要體現在生產者和消費者兩部分的業務邏輯中。對於消息隊列本身的性能,作為使用者不需要太關注。主要原因是對於絕大多數使用消息隊列的業務來說,消息隊列本身的處理能力要遠大於業務系統的處理能力。所以對於消息隊列的性能優化,更關注的是在消息的收發兩端,我們的業務代碼怎么和消息隊列配合,達到一個最佳性能。
1、發送端性能優化
發送端業務代碼的處理性能,實際上和消息隊列的關系不大,因為一般發送端都是先執行自己的業務邏輯,最后再發送消息。如果說,你的代碼發送消息的性能上不去,你需要優先檢查一下,是不是發消息之前的業務邏輯耗時太多導致的。
對於發送消息的業務邏輯,只需要注意設置合適的並發和批量大小,就可以達到很好的發送性能。
Producer 發送消息的過程,Producer 發消息給 Broker,Broker 收到消息后返回確認響應,這是一次完整的交互。假設這一次交互的平均時延是 1ms,我們把這 1ms 的時間分解開,它包括了下面這些步驟的耗時:
- 發送端准備數據、序列化消息、構造請求等邏輯的時間,也就是發送端在發送網絡請求之前的耗時;
- 發送消息和返回響應在網絡傳輸中的耗時;
- Broker 處理消息的時延。
如果是單線程發送,每次只發送 1 條消息,那么每秒只能發送 1000ms / 1ms * 1 條 /ms = 1000 條 消息,這種情況下並不能發揮出消息隊列的全部實力。無論是增加每次發送消息的批量大小,還是增加並發,都能成倍地提升發送性能。
至於到底是選擇批量發送還是增加並發,主要取決於發送端程序的業務性質。簡單來說,只要能夠滿足你的性能要求,怎么實現方便就怎么實現。
- 比如說,你的消息發送端是一個微服務,主要接受 RPC 請求處理在線業務。很自然的,微服務在處理每次請求的時候,就在當前線程直接發送消息就可以了,因為所有 RPC 框架都是多線程支持多並發的,自然也就實現了並行發送消息。並且在線業務比較在意的是請求響應時延,選擇批量發送必然會影響 RPC 服務的時延。這種情況,比較明智的方式就是通過並發來提升發送性能。
- 如果你的系統是一個離線分析系統,離線系統在性能上的需求是它不關心時延,更注重整個系統的吞吐量。發送端的數據都是來自於數據庫,這種情況就更適合批量發送,你可以批量從數據庫讀取數據,然后批量來發送消息,同樣用少量的並發就可以獲得非常高的吞吐量。
2、消費端性能優化
使用消息隊列時,大部分的性能問題都出現在消費端,如果消費的速度跟不上發送端生產消息的速度,就會造成消息積壓。要是消費速度一直比生產速度慢,時間長了,整個系統就會出現問題。要么消息隊列的存儲填滿無法提供服務,要么消息丟失。
所以在設計系統時,一定要保證消費端的消費性能要高於生產端的發送性能,這樣系統才能健康的持續運行。
消費端的性能優化除了優化消費業務邏輯外,也可以通過水平擴展,增加消費端的並發數來提升總體的消費性能。注意:在擴容Consumer的實例數量的同時,必須同步擴容主題中的分區(也叫隊列)數量,確保Consumer的實例數和分區數量是相等的。如果Consumer的實例數量超過分區數量,這樣的擴容實際上是沒有效果的,原因是對於消費者,在每個分區上實際上只能支持單線程消費。
很多消費程序,他們是這樣來解決消費慢的問題的:

它收消息處理的業務邏輯可能比較慢,也很難再優化了,為了避免消息積壓,在收到消息的 OnMessage 方法中,不處理任何業務邏輯,把這個消息放到一個內存隊列里面就返回了。然后它可以啟動很多的業務線程,這些業務線程里面是真正處理消息的業務邏輯,這些線程從內存隊列里取消息處理,這樣它就解決了單個 Consumer 不能並行消費的問題。
但是,這是一個非常常見的錯誤方法! 為什么錯誤?因為會丟消息。如果收消息的節點發生宕機,在內存隊列中還沒來及處理的這些消息就會丟失。在onMessage方法結束后,如果沒有拋異常,就自動ACK了。而這個時候,消息只是在內存隊列中,並沒有被真正處理完。
三、消息積壓的緊急處理
如果日常系統正常運轉的時候,沒有積壓或者只有少量積壓很快就消費掉了,但是某一個時刻,突然就開始積壓消息並且積壓持續上漲。這種情況下需要你在短時間內找到消息積壓的原因,迅速解決問題才不至於影響業務。
排查消息積壓原因的方法:能導致積壓突然增加,最粗粒度的原因,只有兩種:要么是發送變快了,要么是消費變慢了。
-
大部分消息隊列都內置了監控的功能,只要通過監控數據,很容易確定是哪種原因。如果是單位時間發送的消息增多,比如說是趕上大促或者搶購,短時間內不太可能優化消費端的代碼來提升消費性能,唯一的方法是通過擴容消費端的實例數來提升總體的消費能力。如果短時間內沒有足夠的服務器資源進行擴容,沒辦法的辦法是,將系統降級,通過關閉一些不重要的業務,減少發送方發送的數據量,最低限度讓系統還能正常運轉,服務一些重要業務。
- 還有一種不太常見的情況,你通過監控發現,無論是發送消息的速度還是消費消息的速度和原來都沒什么變化,這時候你需要檢查一下你的消費端,是不是消費失敗導致的一條消息反復消費這種情況比較多,這種情況也會拖慢整個系統的消費速度。
- 如果監控到消費變慢了,你需要檢查你的消費實例,分析一下是什么原因導致消費變慢。優先檢查一下日志是否有大量的消費錯誤,如果沒有錯誤的話,可以通過打印堆棧信息,看一下你的消費線程是不是卡在什么地方不動了,比如觸發了死鎖或者卡在等待某些資源上了。
