RocketMQ 一行代碼造成大量消息丟失

1、問(wèn)題現(xiàn)象

首先接到項(xiàng)目反饋使用 RocketMQ 會(huì)出現(xiàn)如下錯(cuò)誤：

由于項(xiàng)目組并沒(méi)有對(duì)消息發(fā)送失敗做任何補(bǔ)償，導(dǎo)致丟失消息發(fā)送失敗，故需要對(duì)這個(gè)問(wèn)題進(jìn)行深層次的探討，并加以解決。

2、問(wèn)題分析

首先我們根據(jù)關(guān)鍵字：TIMEOUT_CLEAN_QUEUE 去 RocketMQ 中查詢，去探究在什么時(shí)候會(huì)拋出如上錯(cuò)誤。根據(jù)全文搜索如下圖所示：

Broker 端快速失敗其原理圖如下：

• 消息發(fā)送者向 Broker 發(fā)送消息寫入請(qǐng)求，Broker 端在接收到請(qǐng)求后會(huì)首先放入一個(gè)隊(duì)列中(SendThreadPoolQueue)，默認(rèn)容量為 10000。

• Broker 會(huì)專門使用一個(gè)線程池(SendMessageExecutor)去從隊(duì)列中獲取任務(wù)并執(zhí)行消息寫入請(qǐng)求，為了保證消息的順序處理，該線程池默認(rèn)線程個(gè)數(shù)為1。

如果 Broker 端受到垃圾回收等等因素造成單條寫入數(shù)據(jù)發(fā)生抖動(dòng)，單個(gè) Broker 端積壓的請(qǐng)求太多從而得不到及時(shí)處理，會(huì)極大的造成客戶端消息發(fā)送的時(shí)間延長(zhǎng)。

設(shè)想一下，如果由于 Broker 壓力增大，寫入一條消息需要500ms甚至超過(guò)1s，并且隊(duì)列中積壓了5000條消息，消息發(fā)送端的默認(rèn)超時(shí)時(shí)間為3s，如果按照這樣的速度，這些請(qǐng)求在輪到 Broker 執(zhí)行寫入請(qǐng)求時(shí)，客戶端已經(jīng)將這個(gè)請(qǐng)求超時(shí)了，這樣不僅會(huì)造成大量的無(wú)效處理，還會(huì)導(dǎo)致客戶端發(fā)送超時(shí)。

故 RocketMQ 為了解決該問(wèn)題，引入 Broker 端快速失敗機(jī)制，即開(kāi)啟一個(gè)定時(shí)調(diào)度線程，每隔10毫秒去檢查隊(duì)列中的第一個(gè)排隊(duì)節(jié)點(diǎn)，如果該節(jié)點(diǎn)的排隊(duì)時(shí)間已經(jīng)超過(guò)了 200ms，就會(huì)取消該隊(duì)列中所有已超過(guò) 200ms 的請(qǐng)求，立即向客戶端返回失敗，這樣客戶端能盡快進(jìn)行重試，因?yàn)?Broker 都是集群部署，下次重試可以發(fā)送到其他 Broker 上，這樣能最大程度保證消息發(fā)送在默認(rèn) 3s 的時(shí)間內(nèi)經(jīng)過(guò)重試機(jī)制，能有效避免某一臺(tái) Broker 由于瞬時(shí)壓力大而造成的消息發(fā)送不可用，從而實(shí)現(xiàn)消息發(fā)送的高可用。

從 Broker 端快速失敗機(jī)制引入的初衷來(lái)看，快速失敗后會(huì)發(fā)起重試，除非同一時(shí)刻集群內(nèi)所有的 Broker 都繁忙，不然消息會(huì)發(fā)送成功，用戶是不會(huì)感知這個(gè)錯(cuò)誤的，那為什么用戶感知了呢？難道 TIMEOUT_ CLEAN _ QUEUE 錯(cuò)誤，Broker 不重試？

為了解開(kāi)這個(gè)謎團(tuán)，接下來(lái)會(huì)采用源碼分析的手段去探究真相。接下來(lái)將以消息同步發(fā)送為例揭示其消息發(fā)送處理流程中的核心關(guān)鍵點(diǎn)。

MQ Client 消息發(fā)送端首先會(huì)利用網(wǎng)絡(luò)通道將請(qǐng)求發(fā)送到 Broker，然后接收到請(qǐng)求結(jié)果后并調(diào)用 processSendResponse 方法對(duì)響應(yīng)結(jié)果進(jìn)行解析，如下圖所示：

我們從 proccessSendResponse 方法中可以得知如果 code 為 SYSTEM_BUSY，該方法會(huì)拋出 MQBrokerException，響應(yīng) code 為 SYSTEM_BUSY，其錯(cuò)誤描述為開(kāi)頭部分的錯(cuò)誤信息。

那我們沿著該方法的調(diào)用鏈路，可以找到其直接調(diào)用方：DefaultMQProducerImpl 的 sendKernelImpl，我們重點(diǎn)考慮如果底層方法拋出  MQBrokerException 該方法會(huì)如何處理。

其關(guān)鍵代碼如下圖所示：

sendKernelImpl 方法被 DefaultMQProducerImpl 的 sendDefaultImpl 方法調(diào)用，下面是其核心實(shí)現(xiàn)截圖：

這里非常令人意外的是連 SYSTEM_ERROR 都會(huì)重試，卻沒(méi)有包含 SYSTEM_BUSY，顯然違背了快速失敗的設(shè)計(jì)初衷，故筆者斷定，這是 RocketMQ 的一個(gè)BUG，將 SYSTEM_BUSY 遺漏了，后續(xù)會(huì)提一個(gè) PR，增加一行代碼，將 SYSTEM_BUSY 加上即可。

問(wèn)題分析到這里，該問(wèn)題應(yīng)該就非常明了。

3、解決方案

如果大家在網(wǎng)上搜索 TIMEOUT_CLEAN_QUEUE 的解決方法，大家不約而同提出的解決方案是增加 waitTimeMillsInSendQueue 的值，該值默認(rèn)為 200ms，例如將其設(shè)置為 1000s 等等，以前我是反對(duì)的，因?yàn)槲业恼J(rèn)知里 Broker 會(huì)重試，但現(xiàn)在發(fā)現(xiàn) Broker 不會(huì)重試，所以我現(xiàn)在認(rèn)為該 BUG未解決的情況下適當(dāng)提高該值能有效的緩解。

但這是并不是好的解決方案，我會(huì)在近期向官方提交一個(gè)PR，將這個(gè)問(wèn)題修復(fù)，建議大家在公司盡量對(duì)自己使用的版本進(jìn)行修改，重新打一個(gè)包即可，因?yàn)檫@已經(jīng)違背了 Broker 端快速失敗的設(shè)計(jì)初衷。

但在消息發(fā)送的業(yè)務(wù)方，盡量自己實(shí)現(xiàn)消息的重試機(jī)制，即不依賴 RocketMQ 本身提供的重試機(jī)制，因?yàn)槭苤朴诰W(wǎng)絡(luò)等因素，消息發(fā)送不可能百分之百成功，建議大家在消息發(fā)送時(shí)捕獲一下異常，如果發(fā)送失敗，可以將消息存入數(shù)據(jù)庫(kù)，再結(jié)合定時(shí)任務(wù)對(duì)消息進(jìn)行重試，盡最大程度保證消息不丟失。

特別推薦一個(gè)分享架構(gòu)+算法的優(yōu)質(zhì)內(nèi)容，還沒(méi)關(guān)注的小伙伴，可以長(zhǎng)按關(guān)注一下：
長(zhǎng)按訂閱更多精彩▼
如有收獲，點(diǎn)個(gè)在看，誠(chéng)摯感謝