消息冪等（去重）通用解決方案，真頂！

圖片

作者 | JaskeyLam

出品 | http://bgoc1.cn/kQZMV

消息中間件是分佈式系統常用的組件，無論是異步化、解耦、削峯等都有廣泛的應用價值。我們通常會認爲，消息中間件是一個可靠的組件——這裏所謂的可靠是指，只要我把消息成功投遞到了消息中間件，消息就不會丟失，即消息肯定會至少保證消息能被消費者成功消費一次，這是消息中間件最基本的特性之一，也就是我們常說的 “AT LEAST ONCE”，即消息至少會被 “成功消費一遍”。

舉個例子，一個消息 M 發送到了消息中間件，消息投遞到了消費程序 A，A 接受到了消息，然後進行消費，但在消費到一半的時候程序重啓了，這時候這個消息並沒有標記爲消費成功，這個消息還會繼續投遞給這個消費者，直到其消費成功了，消息中間件纔會停止投遞。

然而這種可靠的特性導致，消息可能被多次地投遞。舉個例子，還是剛剛這個例子，程序 A 接受到這個消息 M 並完成消費邏輯之後，正想通知消息中間件 “我已經消費成功了” 的時候，程序就重啓了，那麼對於消息中間件來說，這個消息並沒有成功消費過，所以他還會繼續投遞。這時候對於應用程序 A 來說，看起來就是這個消息明明消費成功了，但是消息中間件還在重複投遞。

這在 RockectMQ 的場景來看，就是同一個 messageId 的消息重複投遞下來了。

基於消息的投遞可靠（消息不丟）是優先級更高的，所以消息不重的任務就會轉移到應用程序自我實現，這也是爲什麼 RocketMQ 的文檔裏強調的，消費邏輯需要自我實現冪等。背後的邏輯其實就是：不丟和不重是矛盾的（在分佈式場景下），但消息重複是有解決方案的，而消息丟失是很麻煩的。

簡單的消息去重解決方案

例如：假設我們業務的消息消費邏輯是：插入某張訂單表的數據，然後更新庫存：

insert into t_order values .....
update t_inv set count = count-1 where good_id = 'good123';

要實現消息的冪等，我們可能會採取這樣的方案：

select * from t_order where order_no = 'order123'

if(order  != null) {

    return ;//消息重複，直接返回

}

這對於很多情況下，的確能起到不錯的效果，但是在併發場景下，還是會有問題。

併發重複消息

假設這個消費的所有代碼加起來需要 1 秒，有重複的消息在這 1 秒內（假設 100 毫秒）內到達（例如生產者快速重發，Broker 重啓等），那麼很可能，上面去重代碼裏面會發現，數據依然是空的（因爲上一條消息還沒消費完，還沒成功更新訂單狀態），

那麼就會穿透掉檢查的擋板，最後導致重複的消息消費邏輯進入到非冪等安全的業務代碼中，從而引發重複消費的問題（如主鍵衝突拋出異常、庫存被重複扣減而沒釋放等）

併發去重的解決方案之一

要解決上面併發場景下的消息冪等問題，一個可取的方案是開啓事務把 select 改成 select for update 語句，把記錄進行鎖定。

select * from t_order where order_no = 'THIS_ORDER_NO' for update  //開啓事務
if(order.status != null) {
    return ;//消息重複，直接返回
}

但這樣消費的邏輯會因爲引入了事務包裹而導致整個消息消費可能變長，併發度下降。

當然還有其他更高級的解決方案，例如更新訂單狀態採取樂觀鎖，更新失敗則消息重新消費之類的。但這需要針對具體業務場景做更復雜和細緻的代碼開發、庫表設計，不在本文討論的範圍。

但無論是 select for update， 還是樂觀鎖這種解決方案，實際上都是基於業務表本身做去重，這無疑增加了業務開發的複雜度， 一個業務系統裏面很大部分的請求處理都是依賴 MQ 的，如果每個消費邏輯本身都需要基於業務本身而做去重 / 冪等的開發的話，這是繁瑣的工作量。本文希望探索出一個通用的消息冪等處理的方法，從而抽象出一定的工具類用以適用各個業務場景。

Exactly Once

在消息中間件裏，有一個投遞語義的概念，而這個語義裏有一個叫”Exactly Once”，即消息肯定會被成功消費，並且只會被消費一次。以下是阿里雲裏對 Exactly Once 的解釋：

Exactly-Once 是指發送到消息系統的消息只能被消費端處理且僅處理一次，即使生產端重試消息發送導致某消息重複投遞，該消息在消費端也只被消費一次。

在我們業務消息冪等處理的領域內，可以認爲業務消息的代碼肯定會被執行，並且只被執行一次，那麼我們可以認爲是 Exactly Once。

但這在分佈式的場景下想找一個通用的方案几乎是不可能的。不過如果是針對基於數據庫事務的消費邏輯，實際上是可行的。

基於關係數據庫事務插入消息表

假設我們業務的消息消費邏輯是：更新 MySQL 數據庫的某張訂單表的狀態：

update t_order set status = 'SUCCESS' where order_no= 'order123';

要實現 Exaclty Once 即這個消息只被消費一次（並且肯定要保證能消費一次），我們可以這樣做：在這個數據庫中增加一個消息消費記錄表，把消息插入到這個表，並且把原來的訂單更新和這個插入的動作放到同一個事務中一起提交，就能保證消息只會被消費一遍了。

  1. 開啓事務

  2. 插入消息表（處理好主鍵衝突的問題）

  3. 更新訂單表（原消費邏輯）

  4. 提交事務

說明：

  1. 這時候如果消息消費成功並且事務提交了，那麼消息表就插入成功了，這時候就算 RocketMQ 還沒有收到消費位點的更新再次投遞，也會插入消息失敗而視爲已經消費過，後續就直接更新消費位點了。這保證我們消費代碼只會執行一次。

  2. 如果事務提交之前服務掛了（例如重啓），對於本地事務並沒有執行所以訂單沒有更新，消息表也沒插入成功；而對於 RocketMQ 服務端來說，消費位點也沒更新，所以消息還會繼續投遞下來，投遞下來發現這個消息插入消息表也是成功的，所以可以繼續消費。這保證了消息不丟失。

事實上，阿里雲 ONS 的 EXACTLY-ONCE 語義的實現上，就是類似這個方案基於數據庫的事務特性實現的。更多詳情可參考：https://help.aliyun.com/document_detail/102777.html

基於這種方式，的確這是有能力拓展到不同的應用場景，因爲他的實現方案與具體業務本身無關——而是依賴一個消息表。

但是這裏有它的侷限性

  1. 消息的消費邏輯必須是依賴於關係型數據庫事務。如果消費的消費過程中還涉及其他數據的修改，例如 Redis 這種不支持事務特性的數據源，則這些數據是不可回滾的。

  2. 數據庫的數據必須是在一個庫，跨庫無法解決

注：業務上，消息表的設計不應該以消息 ID 作爲標識，而應該以業務的業務主鍵作爲標識更爲合理，以應對生產者的重發。阿里雲上的消息去重只是 RocketMQ 的 messageId，在生產者因爲某些原因手動重發（例如上游針對一個交易重複請求了）的場景下起不到去重 / 冪等的效果（因消息 id 不同）。

更復雜的業務場景

如上所述，這種方式 Exactly Once 語義的實現，實際上有很多侷限性，這種侷限性使得這個方案基本不具備廣泛應用的價值。並且由於基於事務，可能導致鎖表時間過長等性能問題。

例如我們以一個比較常見的一個訂單申請的消息來舉例，可能有以下幾步（以下統稱爲步驟 X）：

  1. 檢查庫存（RPC）

  2. 鎖庫存（RPC）

  3. 開啓事務，插入訂單表（MySQL）

  4. 調用某些其他下游服務（RPC）

  5. 更新訂單狀態

  6. commit 事務（MySQL）

這種情況下，我們如果採取消息表 + 本地事務的實現方式，消息消費過程中很多子過程是不支持回滾的，也就是說就算我們加了事務，實際上這背後的操作並不是原子性的。怎麼說呢，就是說有可能第一條小在經歷了第二步鎖庫存的時候，服務重啓了，這時候實際上庫存是已經在另外的服務裏被鎖定了，這並不能被回滾。當然消息還會再次投遞下來，要保證消息能至少消費一遍，換句話說，鎖庫存的這個 RPC 接口本身依舊要支持 “冪等”。

再者，如果在這個比較耗時的長鏈條場景下加入事務的包裹，將大大的降低系統的併發。所以通常情況下，我們處理這種場景的消息去重的方法還是會使用一開始說的業務自己實現去重邏輯的方式，如前面加 select for update，或者使用樂觀鎖。

那我們有沒有方法抽取出一個公共的解決方案，能兼顧去重、通用、高性能呢？

拆解消息執行過程

其中一個思路是把上面的幾步，拆解成幾個不同的子消息，例如：

  1. 庫存系統消費 A：檢查庫存並做鎖庫存，發送消息 B 給訂單服務

  2. 訂單系統消費消息 B：插入訂單表（MySQL），發送消息 C 給自己（下游系統）消費

  3. 下游系統消費消息 C：處理部分邏輯，發送消息 D 給訂單系統

  4. 訂單系統消費消息 D：更新訂單狀態

注：上述步驟需要保證本地事務和消息是一個事務的（至少是最終一致性的），這其中涉及到分佈式事務消息相關的話題，不在本文論述。

可以看到這樣的處理方法會使得每一步的操作都比較原子，而原子則意味着是小事務，小事務則意味着使用消息表 + 事務的方案顯得可行。

然而，這太複雜了！這把一個本來連續的代碼邏輯割裂成多個系統多次消息交互！那還不如業務代碼層面上加鎖實現呢。

更通用的解決方案

上面消息表 + 本地事務的方案之所以有其侷限性和併發的短板，究其根本是因爲它依賴於關係型數據庫的事務，且必須要把事務包裹於整個消息消費的環節。

如果我們能不依賴事務而實現消息的去重，那麼方案就能推廣到更復雜的場景例如：RPC、跨庫等。

例如，我們依舊使用消息表，但是不依賴事務，而是針對消息表增加消費狀態，是否可以解決問題呢？

基於消息冪等表的非事務方案

以上是去事務化後的消息冪等方案的流程，可以看到，此方案是無事務的，而是針對消息表本身做了狀態的區分：消費中、消費完成。只有消費完成的消息纔會被冪等處理掉。而對於已有消費中的消息，後面重複的消息會觸發延遲消費（在 RocketMQ 的場景下即發送到 RETRY TOPIC），之所以觸發延遲消費是爲了控制併發場景下，第二條消息在第一條消息沒完成的過程中，去控制消息不丟（如果直接冪等，那麼會丟失消息（同一個消息 id 的話），因爲上一條消息如果沒有消費完成的時候，第二條消息你已經告訴 broker 成功了，那麼第一條消息這時候失敗 broker 也不會重新投遞了）

上面的流程不再細說，後文有 github 源碼的地址，讀者可以參考源碼的實現，這裏我們回頭看看我們一開始想解決的問題是否解決了：

  1. 消息已經消費成功了，第二條消息將被直接冪等處理掉（消費成功）。

  2. 併發場景下的消息，依舊能滿足不會出現消息重複，即穿透冪等擋板的問題。

  3. 支持上游業務生產者重發的業務重複的消息冪等問題。

關於第一個問題已經很明顯已經解決了，在此就不討論了。

關於第二個問題是如何解決的？主要是依靠插入消息表的這個動作做控制的，假設我們用 MySQL 作爲消息表的存儲媒介（設置消息的唯一 ID 爲主鍵），那麼插入的動作只有一條消息會成功，後面的消息插入會由於主鍵衝突而失敗，走向延遲消費的分支，然後後面延遲消費的時候就會變成上面第一個場景的問題。

關於第三個問題，只要我們設計去重的消息鍵讓其支持業務的主鍵（例如訂單號、請求流水號等），而不僅僅是 messageId 即可。所以也不是問題。

此方案是否有消息丟失的風險？

如果細心的讀者可能會發現這裏實際上是有邏輯漏洞的，問題出在上面聊到的個三問題中的第 2 個問題（併發場景），在併發場景下我們依賴於消息狀態是做併發控制使得第 2 條消息重複的消息會不斷延遲消費（重試）。但如果這時候第 1 條消息也由於一些異常原因（例如機器重啓了、外部異常導致消費失敗）沒有成功消費成功呢？也就是說這時候延遲消費實際上每次下來看到的都是_消費中_的狀態，最後消費就會被視爲消費失敗而被投遞到死信 Topic 中（RocketMQ 默認可以重複消費 16 次）。

有這種顧慮是正確的！對於此，我們解決的方法是，插入的消息表必須要帶一個最長消費過期時間，例如 10 分鐘，意思是如果一個消息處於_消費中_超過 10 分鐘，就需要從消息表中刪除（需要程序自行實現）。所以最後這個消息的流程會是這樣的：

更靈活的消息表存儲媒介

我們這個方案實際上沒有事務的，只需要一個存儲的中心媒介，那麼自然我們可以選擇更靈活的存儲媒介，例如 Redis。使用 Redis 有兩個好處：

  1. 性能上損耗更低

  2. 上面我們講到的超時時間可以直接利用 Redis 本身的 ttl 實現

當然 Redis 存儲的數據可靠性、一致性等方面是不如 MySQL 的，需要用戶自己取捨。

源碼：RocketMQDedupListener

以上方案針對 RocketMQ 的 Java 實現已經開源放到 Github 中，具體的使用文檔可以參考 https://github.com/Jaskey/RocketMQDedupListener ，

以下僅貼一個 Readme 中利用 Redis 去重的使用樣例，用以意業務中如果使用此工具加入消息去重冪等的是多麼簡單：

//利用Redis做冪等表
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("TEST-APP1");
consumer.subscribe("TEST-TOPIC", "*");

String appName = consumer.getConsumerGroup();// 大部分情況下可直接使用consumer group名
StringRedisTemplate stringRedisTemplate = null;// 這裏省略獲取StringRedisTemplate的過程
DedupConfig dedupConfig = DedupConfig.enableDedupConsumeConfig(appName, stringRedisTemplate);
DedupConcurrentListener messageListener = new SampleListener(dedupConfig);

consumer.registerMessageListener(messageListener);
consumer.start();

以上代碼大部分是原始 RocketMQ 的必須代碼，唯一需要修改的僅僅是創建一個DedupConcurrentListener示例，在這個示例中指明你的消費邏輯和去重的業務鍵（默認是 messageId）。

更多使用詳情請參考 Github 上的說明。

這種實現是否一勞永逸？

實現到這裏，似乎方案挺完美的，所有的消息都能快速的接入去重，且與具體業務實現也完全解耦。那麼這樣是否就完美的完成去重的所有任務呢？

很可惜，其實不是的。原因很簡單：因爲要保證消息至少被成功消費一遍，那麼消息就有機會消費到一半的時候失敗觸發消息重試的可能。還是以上面的訂單流程 X：

  1. 檢查庫存（RPC）

  2. 鎖庫存（RPC）

  3. 開啓事務，插入訂單表（MySQL）

  4. 調用某些其他下游服務（RPC）

  5. 更新訂單狀態

  6. commit 事務（MySQL）

當消息消費到步驟 3 的時候，我們假設 MySQL 異常導致失敗了，觸發消息重試。因爲在重試前我們會刪除冪等表的記錄，所以消息重試的時候就會重新進入消費代碼，那麼步驟 1 和步驟 2 就會重新再執行一遍。如果步驟 2 本身不是冪等的，那麼這個業務消息消費依舊沒有做好完整的冪等處理。

本實現方式的價值？

那麼既然這個並不能完整的完成消息冪等，還有什麼價值呢？價值可就大了！雖然這不是解決消息冪等的銀彈（事實上，軟件工程領域裏基本沒有銀彈），但是他能以便捷的手段解決：

  1. 各種由於 Broker、負載均衡等原因導致的消息重投遞的重複問題

  2. 各種上游生產者導致的業務級別消息重複問題

  3. 重複消息併發消費的控制窗口問題，就算重複，重複也不可能同一時間進入消費邏輯

一些其他的消息去重的建議

也就是說，使用這個方法能保證正常的消費邏輯場景下（無異常，無異常退出），消息的冪等工作全部都能解決，無論是業務重複，還是 rocketmq 特性帶來的重複。

事實上，這已經能解決 99% 的消息重複問題了，畢竟異常的場景肯定是少數的。那麼如果希望異常場景下也能處理好冪等的問題，可以做以下工作降低問題率：

  1. 消息消費失敗做好回滾處理。如果消息消費失敗本身是帶回滾機制的，那麼消息重試自然就沒有副作用了。

  2. 消費者做好優雅退出處理。這是爲了儘可能避免消息消費到一半程序退出導致的消息重試。

  3. 一些無法做到冪等的操作，至少要做到終止消費並告警。例如鎖庫存的操作，如果統一的業務流水鎖成功了一次庫存，再觸發鎖庫存，如果做不到冪等的處理，至少要做到消息消費觸發異常（例如主鍵衝突導致消費異常等）

  4. 在 #3 做好的前提下，做好消息的消費監控，發現消息重試不斷失敗的時候，手動做好 #1 的回滾，使得下次重試消費成功。

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/NyyYIZIhimXRk9BOGf85WQ

猜你喜歡