[[431068]]

大家好，我是captain，如果覺得下面的文章對你有所幫助，歡迎大家關注！

上一篇說過了如何保證消息不丟失，分別從producer、broker和consumer三個角色來保證消息的不丟失，其實說到底也不可能百分百保證消息不丟失，這種當然是極端的情況下。

這一篇我們要說的話題是消息的堆積處理，其實這個話題還是挺大的，因為消息堆積還是真的很令人頭疼的，當堆積的量很大的時候，這真的是個很暴躁的問題，不過這時候真考驗大家冷靜的處理問題的能力了。

我們一起來分析分析有關問題吧！

• 大量的消息堆積在MQ中幾個小時還沒解決怎么辦呢。

一般這種比較著急的問題，最好的辦法就是臨時擴容，用更快的速度來消費數據。

1、臨時建立一個新的Topic，然后調整queue的數量為原來的10倍或者20倍，根據堆積情況來決定。

2、然后寫一個臨時分發消息的consumer程序，這個程序部署上去消費積壓的消息，消費的就是剛剛新建的Topic，消費之后不做耗時的處理，只需要直接均勻的輪詢將這些消息輪詢的寫入到臨時創建的queue里面即可。

3、然后增加相應倍數的機器來部署真正的consumer消費，注意這里的Topic，然后讓這些consumer去真正的消費這些臨時的queue里面的消息。

不知道大家明白沒有，很簡單的道理，我給大家舉個形象的例子。

一個topic堵住了，新建一個topic去進行分流，臨時將queue資源和consumer資源擴大10倍，將消息平均分配到這些新增的queue資源和consumer資源上，以正常10倍的速度來消費消息，等到這些堆積的消息消費完了，便可以恢復到原來的部署架構。

這種只是用于臨時解決一些異常情況導致的消息堆積的處理，如果消息經常出現堵塞的情況，那該考慮一下徹底增強系統的部署架構了。

• 消息設置了過期時間，過期就丟了怎么辦呢？

在rabbitmq中，可以設置過期時間TTL，和Redis的過期時間一樣，如果消息在queue中積壓超過一定時間就會被rabbitmq清理掉，這個數據就沒了。

這樣可能會造成大量的數據丟失。

這種情況下上面的解決方案就不太合適了，可以采取批量重導的方案來解決，在系統流量比較低的時候，用程序去查詢丟失的這部分數據，然后將消息重新發送到MQ中，把丟失的數據重新補回來。

這也算是一種補償任務吧，補償任務一般是用于對定時跑批的一種補償。

• 分析下RocketMQ中的消息堆積原因

消息的堆積歸根到底就是生產者生產消息的速度和消費者消費的速度不匹配導致的，輸入的和消費的速度不統一。

或許是突然搞了一波促銷，系統業務量暴增，導致生產者發消息暴增，消費速度跟不上。

也有可能是消費方出現失敗的情況，瘋狂重試，也或者就是消費方的消費能力太低了。

RocketMQ是按照隊列進行消息負載的，如果consumer中的一臺機器由于硬件各方面原因導致該機器上的消息隊列不能及時處理，就會造成整個消息隊列的堆積。

RocketMQ分為發布方和訂閱方，雙方都有負載均衡策略，默認都是采用平均分配，producer消息以輪詢方式發送到消息隊列queue中，broker將這些的queue再平均分配到屬于同一個group id的訂閱方集群。

• 如果消費者consumer機器數量和消息隊列相等，則消息隊列平均分配到每一個consumer上
• 如果consumer數量大于消息隊列數量，則超出消息隊列數量的機器沒有可以處理的消息隊列
• 若消息隊列數量不是consumer的整數倍，則部分consumer會承擔跟多的消息隊列的消費任務

如果其中一臺機器處理變慢，可能是機器硬件、系統、遠程 RPC 調用或 Java GC 等原因導致分配至此機器上的 Queue 的消息不能及時處理。

消息隊列 RocketMQ 版的消息負載是按 Queue 為粒度維護，所以，整個 Queue 上的消息都會堆積。

• 那說一下解決思路吧

我們知道了最根本原因是生產和消費速度不匹配導致的，這種問題要是經常出現，就是系統架構導致，這種需要考慮增加消費方的數量了。

如果是搞促銷的這種臨時情況導致的，這種情況下系統應該會比較快的消化掉，堆積時間不會很快，如果搞促銷時間很長，持續高流量時間很長，那沒得辦法，還是得加機器。

經常出現這種消息堆積問題，需要先定位一下消費滿的原因，也也可能是代碼bug，導致多次重試，如果是bug則處理bug，優化下消費的邏輯。

再者就要考慮水平擴容，增加Topic的queue數量和消費者的數量，這兩者增加的時候需要考慮兩邊的平衡，隊列數量一定要增加，不然新增加的消費數量者會導致無消息消費的尷尬場面，一個topic中的一個隊列只會分配給一個消費者。

消費者數量超過隊列數量的時候，超出的部分消費者就無消息可以消費了。

• RocketMQ中消費完的消息去了哪里呢

消息的存儲是一直存在于CommitLog文件中的，大家都知道CommitLog是以文件為單位存在的，而且RocketMQ的設計是只允許順序寫，也就意味著所有消息都是順序的寫入到這個文件中的。

而每個消息的大小又不是定長的，所以這就決定了消息幾乎不可能按照消息為單位進行刪除，邏輯極其復雜。

消息一旦被消費了之后是不會被立即清除的，還是會存在于CommitLog文件中的，那問題來了，消息未刪除，RocketMQ是如何知道哪些消息已經被消費過，哪些還未消費呢。

答案就是客戶端會維護一個消息的offset，客戶端拉取完消息之后，broker會隨著響應體返回一個下一次拉取的位置，消費者會更新自己的下一次的pull的位置。

• CommitLog文件什么時候進行清除

消息存儲到該文件之后，也是會被清理的，但是這個清理只會在下面這些條件中，任一條件成立的時候才會批量的刪除CommitLog消息文件。

• 消息文件過期(默認72小時),且到達清理時點(默認是凌晨4點),刪除過期文件。
• 消息文件過期(默認72小時),且磁盤空間達到了水位線(默認75%),刪除過期文件。
• 磁盤已經達到必須釋放的上限(85%水位線)的時候,則開始批量清理文件(無論是否過期),直到空間充足。

注:若磁盤空間達到危險水位線(默認90%),出于保護自身的目的,broker會拒絕寫入服務。

• 為什么這么設計呢

CommitLog文件默認大小是1GB，在清理的時候屬于大文件操作了，IO壓力也是有的，這樣設計該文件的優點我大概說幾個，當然肯定還有些別的。

只需要保存一份消息文件：一個消息如果需要被多個消費者組消費，消息只需要保存一份即可，消費進度單獨保存，這樣比較容易支撐強大的消息存儲能力。

支持回溯：把消息的消費位置的決定權放在客戶端，只要消息還在，就可以消費，所以也就有了RocketMQ支持的回溯消費。

像看視頻一樣，可以把鏡頭調到前面去，重新看一遍剛剛的視頻

支持消息索引服務：RocketMQ中有一個索引文件，消息只要還存在于CommitLog中，就可以被搜索出來，方便排查問題。