作者：京東物流 劉浩

1. 背景

在分布式系統(tǒng)中，消息隊(duì)列（MQ）是實(shí)現(xiàn)系統(tǒng)解耦、異步通信的重要工具。然而，MQ消費(fèi)時(shí)出現(xiàn)的消息亂序問題，經(jīng)常會對業(yè)務(wù)邏輯的正確執(zhí)行和系統(tǒng)穩(wěn)定性產(chǎn)生不良影響。本文將詳細(xì)探討MQ消息亂序問題的根源，并提供一系列在實(shí)際應(yīng)用中可行的解決方案。

2. MQ消息亂序問題分析

常見的MQ消息亂序問題的根源主要可以歸結(jié)為以下幾點(diǎn)：

2.1 相同topic內(nèi)的消息亂序

1). 并發(fā)消費(fèi)：

在分布式系統(tǒng)中，為了提高消息處理的吞吐量，通常會配置多個(gè)消費(fèi)者實(shí)例來并發(fā)消費(fèi)同一個(gè)隊(duì)列中的消息。然而，由于消費(fèi)者實(shí)例的機(jī)器性能、網(wǎng)絡(luò)延遲以及處理速度的差異，可能導(dǎo)致消息的消費(fèi)順序與發(fā)送順序不一致。

2). 消息分區(qū)：

為了支持更高效的消息存儲和消費(fèi)，MQ系統(tǒng)通常會采用分區(qū)化的設(shè)計(jì)。然而，當(dāng)同一業(yè)務(wù)邏輯的多條消息被分發(fā)到不同的分區(qū)時(shí)，消費(fèi)者在消費(fèi)這些消息時(shí)就可能出現(xiàn)亂序現(xiàn)象。

3). 網(wǎng)絡(luò)延遲與抖動(dòng)：

消息在傳輸過程中可能會受到網(wǎng)絡(luò)延遲和抖動(dòng)的影響，導(dǎo)致消息到達(dá)消費(fèi)者端的時(shí)間順序與發(fā)送順序不一致。

4). 消息重試與故障恢復(fù)：

當(dāng)消費(fèi)者處理消息失敗或出現(xiàn)故障時(shí)，MQ系統(tǒng)通常會進(jìn)行消息重試或故障恢復(fù)操作。如果重試機(jī)制或故障恢復(fù)策略設(shè)計(jì)不當(dāng)，也可能導(dǎo)致消息亂序。

2.2 不同topic的消息亂序

從相對時(shí)間的視角來審視，消息被消費(fèi)的順序并不等同于其被發(fā)送的順序。例如，系統(tǒng)A在12:00時(shí)向TopicA發(fā)送了消息msgA-12:00，而緊接著系統(tǒng)B在12:01時(shí)向TopicB發(fā)送了消息msgB-12:01。當(dāng)系統(tǒng)C同時(shí)訂閱并消費(fèi)這兩個(gè)Topic時(shí)，它無法預(yù)設(shè)msgA-12:00會必然先于msgB-12:01被接收。這是由于消息系統(tǒng)在處理過程中，受到諸如消息分區(qū)策略、各個(gè)Consumer的處理能力以及其諸如網(wǎng)絡(luò)、堆積、重試等他綜合因素的影響，導(dǎo)致無法確保消息遵循嚴(yán)格的先進(jìn)先出原則。

3. 案例分析

3.1 數(shù)據(jù)遷移過程中的mq消費(fèi)亂序場景

在數(shù)據(jù)遷移或同步過程中，尤其是雙寫場景（即數(shù)據(jù)既寫入舊系統(tǒng)，又通過MQ發(fā)送到新系統(tǒng)進(jìn)行異步處理），MQ亂序可能導(dǎo)致嚴(yán)重的數(shù)據(jù)不一致問題。

具體來說，當(dāng)數(shù)據(jù)寫入時(shí)發(fā)送INSERT MQ，數(shù)據(jù)更新時(shí)發(fā)送UPDATE MQ，如果UPDATE MQ先于INSERT MQ到達(dá)目標(biāo)系統(tǒng)，目標(biāo)系統(tǒng)可能會基于一個(gè)不存在的數(shù)據(jù)記錄進(jìn)行更新操作。這會導(dǎo)致以下幾種情況：

數(shù)據(jù)丟失：如果目標(biāo)系統(tǒng)沒有處理UPDATE MQ中提到的數(shù)據(jù)記錄（因?yàn)樵撚涗浬形赐ㄟ^INSERT MQ創(chuàng)建），則更新操作會失敗，可能導(dǎo)致數(shù)據(jù)變更丟失或遺漏。

數(shù)據(jù)覆蓋：在高頻修改的情況下，頻繁更新可能會面臨舊數(shù)據(jù)覆蓋新數(shù)據(jù)的風(fēng)險(xiǎn)，比如UPDATE MQ攜帶的是舊數(shù)據(jù)且先于新數(shù)據(jù)的UPDATE MQ到達(dá)。

3.2 業(yè)務(wù)風(fēng)險(xiǎn)分析

MQ亂序?qū)?shù)據(jù)遷移和同步過程的影響是深遠(yuǎn)的：

數(shù)據(jù)一致性受損：最直接的影響是數(shù)據(jù)一致性受損。目標(biāo)系統(tǒng)中的數(shù)據(jù)可能與源系統(tǒng)不一致，導(dǎo)致業(yè)務(wù)決策基于錯(cuò)誤的數(shù)據(jù)。

用戶體驗(yàn)下降：數(shù)據(jù)不一致可能導(dǎo)致用戶看到錯(cuò)誤的信息或遇到功能故障，從而降低用戶體驗(yàn)。

業(yè)務(wù)中斷：在嚴(yán)重的情況下，數(shù)據(jù)不一致可能導(dǎo)致業(yè)務(wù)中斷或系統(tǒng)故障，影響企業(yè)的運(yùn)營和聲譽(yù)。

4. 解決方案

為了解決這個(gè)問題，可以采取以下措施：

4.1 順序消息

消息順序性保證：雖然Kafka不保證全局消息順序，但可以通過合理的分區(qū)策略和消息鍵來確保同一賬單的消息被發(fā)送到同一個(gè)分區(qū)，從而在一定程度上保證消息的順序性。

比如RocketMQ支持順序消息。但是需要注意這是局部有序，非全局后續(xù)。具體實(shí)現(xiàn)過程：

1.發(fā)送mq消息時(shí)，通過selector將同一個(gè)業(yè)務(wù)主鍵的消息，發(fā)送到同一隊(duì)列中

2.消費(fèi)方使用MessageListenerOrderly消費(fèi)局部有序的消息

該方案需要發(fā)送方和消費(fèi)方同步改造。

4.2 前置檢測

?在消費(fèi)者處理消息之前，進(jìn)行前置條件檢查。例如，可以查詢一個(gè)消息輔助表，確保上一個(gè)消息已經(jīng)被成功消費(fèi)或存入死信隊(duì)列中。這種檢查可以確保消息按照正確的順序被處理。

?另一種方法是，在消息中添加序列號或時(shí)間戳，并在消費(fèi)者端進(jìn)行驗(yàn)證。如果當(dāng)前消息的序列號或時(shí)間戳不符合預(yù)期順序，則暫停處理并等待正確的消息到達(dá)。

4.3 狀態(tài)機(jī)

在消息處理系統(tǒng)中，狀態(tài)機(jī)可以用來定義和處理消息的順序。每個(gè)狀態(tài)代表系統(tǒng)當(dāng)前所處的特定條件或階段，而狀態(tài)之間的轉(zhuǎn)換則是由接收到的消息觸發(fā)的。當(dāng)系統(tǒng)接收到一個(gè)消息時(shí)，它會檢查當(dāng)前的狀態(tài)和消息類型，然后決定是否要轉(zhuǎn)移到另一個(gè)狀態(tài)并執(zhí)行相應(yīng)的動(dòng)作。

對于消息亂序問題，狀態(tài)機(jī)可以通過以下方式解決：

1.定義狀態(tài)轉(zhuǎn)換規(guī)則：首先，需要定義一套明確的狀態(tài)轉(zhuǎn)換規(guī)則。這些規(guī)則應(yīng)該基于業(yè)務(wù)邏輯來確定，以確保消息按照正確的順序被處理。例如，如果系統(tǒng)要求先處理事件A再處理事件B，那么狀態(tài)機(jī)就應(yīng)該在接收到事件A后轉(zhuǎn)移到能夠處理事件B的狀態(tài)。

2.狀態(tài)檢查與消息緩存：當(dāng)系統(tǒng)接收到一個(gè)消息時(shí)，它會檢查當(dāng)前的狀態(tài)是否允許處理該消息。如果當(dāng)前狀態(tài)不允許處理該消息（即消息的順序不正確），則可以將該消息緩存起來，等待狀態(tài)機(jī)轉(zhuǎn)移到正確的狀態(tài)后再進(jìn)行處理。

3.狀態(tài)轉(zhuǎn)移與消息處理：一旦狀態(tài)機(jī)轉(zhuǎn)移到正確的狀態(tài)，它就可以處理緩存中的消息。這可以確保消息按照正確的順序被處理，即使它們最初是以亂序到達(dá)的。

4.4 監(jiān)控與報(bào)警

建立系統(tǒng)的監(jiān)控和報(bào)警機(jī)制，及時(shí)發(fā)現(xiàn)并處理消息錯(cuò)亂等異常情況。

通過采取以上措施，可以大大降低賬單還款系統(tǒng)中消息錯(cuò)亂導(dǎo)致的問題，提高系統(tǒng)的穩(wěn)定性和用戶體驗(yàn)。

5. 小結(jié)

MQ消息亂序是分布式系統(tǒng)的常見難題，影響系統(tǒng)穩(wěn)定性和業(yè)務(wù)一致性。本文深入解析問題根源，探討了順序消息、前置檢查、狀態(tài)機(jī)等實(shí)戰(zhàn)解決方案，為實(shí)際開發(fā)中的問題解決提供有力參考。

文章中難免會有不足之處，希望讀者能給予寶貴的意見和建議。謝謝！

審核編輯 黃宇