[[431068]]

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

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

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

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

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

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

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

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

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

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

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

這種只是用于臨時解決一些異常情況導致的消息堆積的處理，如果消息經(jīng)常出現(xiàn)堵塞的情況，那該考慮一下徹底增強系統(tǒng)的部署架構(gòu)了。

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

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

這樣可能會造成大量的數(shù)據(jù)丟失。

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

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

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

消息的堆積歸根到底就是生產(chǎn)者生產(chǎn)消息的速度和消費者消費的速度不匹配導致的，輸入的和消費的速度不統(tǒng)一。

或許是突然搞了一波促銷，系統(tǒng)業(yè)務量暴增，導致生產(chǎn)者發(fā)消息暴增，消費速度跟不上。

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

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

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

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

如果其中一臺機器處理變慢，可能是機器硬件、系統(tǒng)、遠程 RPC 調(diào)用或 Java GC 等原因?qū)е路峙渲链藱C器上的 Queue 的消息不能及時處理。

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

• 那說一下解決思路吧

我們知道了最根本原因是生產(chǎn)和消費速度不匹配導致的，這種問題要是經(jīng)常出現(xiàn)，就是系統(tǒng)架構(gòu)導致，這種需要考慮增加消費方的數(shù)量了。

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

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

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

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

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

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

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

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

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

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

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

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

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

• 為什么這么設計呢

CommitLog文件默認大小是1GB，在清理的時候?qū)儆诖笪募僮髁耍琁O壓力也是有的，這樣設計該文件的優(yōu)點我大概說幾個，當然肯定還有些別的。

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

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

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

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