[[431068]]

大家好，我是captain，如果覺得下面的文章對(duì)你有所幫助，歡迎大家關(guān)注！

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

這一篇我們要說的話題是消息的堆積處理，其實(shí)這個(gè)話題還是挺大的，因?yàn)橄⒍逊e還是真的很令人頭疼的，當(dāng)堆積的量很大的時(shí)候，這真的是個(gè)很暴躁的問題，不過這時(shí)候真考驗(yàn)大家冷靜的處理問題的能力了。

我們一起來分析分析有關(guān)問題吧！

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

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

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

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

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

不知道大家明白沒有，很簡(jiǎn)單的道理，我給大家舉個(gè)形象的例子。

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

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

• 消息設(shè)置了過期時(shí)間，過期就丟了怎么辦呢？

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

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

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

這也算是一種補(bǔ)償任務(wù)吧，補(bǔ)償任務(wù)一般是用于對(duì)定時(shí)跑批的一種補(bǔ)償。

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

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

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

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

RocketMQ是按照隊(duì)列進(jìn)行消息負(fù)載的，如果consumer中的一臺(tái)機(jī)器由于硬件各方面原因?qū)е略摍C(jī)器上的消息隊(duì)列不能及時(shí)處理，就會(huì)造成整個(gè)消息隊(duì)列的堆積。

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

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

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

消息隊(duì)列 RocketMQ 版的消息負(fù)載是按 Queue 為粒度維護(hù)，所以，整個(gè) Queue 上的消息都會(huì)堆積。

• 那說一下解決思路吧

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

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

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

再者就要考慮水平擴(kuò)容，增加Topic的queue數(shù)量和消費(fèi)者的數(shù)量，這兩者增加的時(shí)候需要考慮兩邊的平衡，隊(duì)列數(shù)量一定要增加，不然新增加的消費(fèi)數(shù)量者會(huì)導(dǎo)致無消息消費(fèi)的尷尬場(chǎng)面，一個(gè)topic中的一個(gè)隊(duì)列只會(huì)分配給一個(gè)消費(fèi)者。

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

• RocketMQ中消費(fèi)完的消息去了哪里呢

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

而每個(gè)消息的大小又不是定長(zhǎng)的，所以這就決定了消息幾乎不可能按照消息為單位進(jìn)行刪除，邏輯極其復(fù)雜。

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

答案就是客戶端會(huì)維護(hù)一個(gè)消息的offset，客戶端拉取完消息之后，broker會(huì)隨著響應(yīng)體返回一個(gè)下一次拉取的位置，消費(fèi)者會(huì)更新自己的下一次的pull的位置。

• CommitLog文件什么時(shí)候進(jìn)行清除

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

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

注:若磁盤空間達(dá)到危險(xiǎn)水位線(默認(rèn)90%),出于保護(hù)自身的目的,broker會(huì)拒絕寫入服務(wù)。

• 為什么這么設(shè)計(jì)呢

CommitLog文件默認(rèn)大小是1GB，在清理的時(shí)候?qū)儆诖笪募僮髁?，IO壓力也是有的，這樣設(shè)計(jì)該文件的優(yōu)點(diǎn)我大概說幾個(gè)，當(dāng)然肯定還有些別的。

只需要保存一份消息文件：一個(gè)消息如果需要被多個(gè)消費(fèi)者組消費(fèi)，消息只需要保存一份即可，消費(fèi)進(jìn)度單獨(dú)保存，這樣比較容易支撐強(qiáng)大的消息存儲(chǔ)能力。

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

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

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