作者：京東物流 劉浩

1. 背景

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

2. MQ消息亂序問題分析

常見的MQ消息亂序問題的根源主要可以歸結為以下幾點：

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

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

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

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

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

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

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

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

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

2.2 不同topic的消息亂序

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

3. 案例分析

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

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

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

數(shù)據(jù)丟失：如果目標系統(tǒng)沒有處理UPDATE MQ中提到的數(shù)據(jù)記錄（因為該記錄尚未通過INSERT MQ創(chuàng)建），則更新操作會失敗，可能導致數(shù)據(jù)變更丟失或遺漏。

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

3.2 業(yè)務風險分析

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

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

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

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

4. 解決方案

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

4.1 順序消息

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

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

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

2.消費方使用MessageListenerOrderly消費局部有序的消息

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

4.2 前置檢測

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

?另一種方法是，在消息中添加序列號或時間戳，并在消費者端進行驗證。如果當前消息的序列號或時間戳不符合預期順序，則暫停處理并等待正確的消息到達。

4.3 狀態(tài)機

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

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

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

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

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

4.4 監(jiān)控與報警

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

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

5. 小結

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

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

審核編輯 黃宇