哎，被這個叫做at least once的玩意坑麻了。

你好呀，我是歪歪。

前幾天遇到一個生產(chǎn)問題，同一個數(shù)據(jù)在數(shù)據(jù)庫里面被插入了兩次，導(dǎo)致后續(xù)處理出現(xiàn)了一些問題。

當(dāng)時我們首先檢討了自己，沒有做好冪等校驗。甚至還發(fā)現(xiàn)了一個低級錯誤：對應(yīng)的表，針對訂單號，這個業(yè)務(wù)上具有唯一屬性的字段，連唯一索引都沒有加。如果加了唯一索引，也不至于出現(xiàn)落庫兩次的情況。

然后拿著數(shù)據(jù)去問上游系統(tǒng)，為什么會出現(xiàn)同一個訂單發(fā)起了兩次的這種異常場景。

上游系統(tǒng)一聽到我們的描述，立馬就站出來解釋：不可能，在沒有人工介入的情況下，同一個單子，我們絕對不可能發(fā)送兩次。在開發(fā)的過程中，我們還特意注意了這個場景。

但是我是不相信他們的“鬼話”，我更覺得這就是他們的一個 BUG。

既然爭執(zhí)不下了，那就拿事實說話。

日志就是事實。

于是我們一起查詢了日志，最后的結(jié)果就更加奇怪了。

調(diào)用方確實只有一次調(diào)用日志。

但是我們接收方卻收到了兩次請求。

通過圖片也能看出來，我們之間是通過 MQ 異步交互的。

所以，自然而然的就把目光放到 MQ 上。

我們使用的 MQ 是一個叫做 SofaMQ 的玩意，比較冷門，但是有螞蟻金服背書。

在官方文檔的“常見問題”部分，有這樣的描述：

這句話你仔細品一下：可以保證消息不丟失，但是無法保證消息不重復(fù)。

言外之意是不是就是在說：為了保證消息不丟失，在我拿不準你到底有沒有消費成功的情況下，我有可能針對的同一個消息再次發(fā)送。

再次發(fā)送，那不就是一個消息會被消費多次嗎？

不就是我們遇到的這個問題嗎？

然后我突然想起了一個曾經(jīng)學(xué)過的東西：at least once。

在 MQTT 協(xié)議中，給出了三種傳遞消息時能夠提供的服務(wù)質(zhì)量標(biāo)準，這三種服務(wù)質(zhì)量從低到高依次是：

• At most once: 至多一次。消息在傳遞時，最多會被送達一次。換一個說法就是，沒什么消息可靠性保證，允許丟消息。一般都是一些對消息可靠性要求不太高的監(jiān)控場景使用，比如每分鐘上報一次機房溫度數(shù)據(jù)，可以接受數(shù)據(jù)少量丟失。
• At least once：至少一次。消息在傳遞時，至少會被送達一次。也就是說，不允許丟消息，但是允許有少量重復(fù)消息出現(xiàn)。
• Exactly once：恰好一次。消息在傳遞時，只會被送達一次，不允許丟失也不允許重復(fù)，這個是最高的等級

同時，在“消息冪等”部分，也特別進行了強調(diào)：

https://help.aliyun.com/document_detail/146983.html

為了防止消息重復(fù)消費導(dǎo)致業(yè)務(wù)處理異常，有必要根據(jù)業(yè)務(wù)上的唯一 Key 對消息做冪等處理。

雖然我用的這個玩意是一個冷門的 MQ ，但是這個問題和具體使用的哪個 MQ 關(guān)系不大，常見的 RabbitM、RocketMQ、Kafka 都有類似的問題，都需要消費端做好冪等處理。

本文就基于這個問題，來討論一下，在“消息可能重復(fù)消費”這個場景下，有沒有啥好的解決方案。

舉個例子

前面說了，要處理消息重復(fù)消費的場景，最核心的邏輯是需要實現(xiàn)冪等機制。

冪等，這個概念大家應(yīng)該是比較清晰了。

舉個具體的例子。

比如在支付場景下，消費者消費扣款消息，對一筆訂單執(zhí)行扣款操作，扣款金額為 100 元。

如果因各種原因?qū)е驴劭钕�息重�?fù)投遞，比如簡單的一個場景，消費者接受到“扣款金額為 100 元”這個信息，完成消費，還沒來得及告訴 MQ，“老哥，這個消息我已經(jīng)收到了”，就重啟了。

站在 MQ 的角度，沒有收到回執(zhí)，就代表這個消息并沒有消費成功，基于“必須保證消息不丟失的指導(dǎo)思想”，它就會繼續(xù)投遞。

所以消費者會重復(fù)消費這個扣款消息。

但是，最終的業(yè)務(wù)結(jié)果是只扣款一次，扣費 100 元，且用戶的扣款記錄中對應(yīng)的訂單只有一條扣款流水，不會多次扣除費用。

那么這次扣款操作是符合要求的，整個消費過程實現(xiàn)了消費冪等。

在要求冪等的場景中，我們要找到一個抓手。

比如在這個案例里面，扣款一般來說都會對應(yīng)一個業(yè)務(wù)上的唯一流水號，這個業(yè)務(wù)上的唯一流水號，就是抓手，我們可以基于這個流水號來做冪等。

最常規(guī)的方案就是在這個字段上加唯一索引，然后出現(xiàn)重復(fù)投遞時，落庫的時候會拋出主鍵沖突的異常。

不要覺得重復(fù)投遞是一個小概率事件，就不上心了。我們敲代碼的，不就是要多考慮這些正常流程之外的“小概率事件”嗎，只寫正常流程，誰都會寫。

根據(jù)官方的說法，消息重復(fù)會發(fā)生在這些場景中：

• 發(fā)送時消息重復(fù)。當(dāng)一條消息已被成功發(fā)送到服務(wù)端并完成持久化，此時出現(xiàn)了網(wǎng)絡(luò)閃斷或者客戶端宕機，導(dǎo)致服務(wù)端對客戶端應(yīng)答失敗。 如果此時生產(chǎn)者意識到消息發(fā)送失敗并嘗試再次發(fā)送消息，消費者后續(xù)會收到兩條內(nèi)容相同并且 Message ID 也相同的消息。
• 投遞時消息重復(fù)消息消費的場景下，消息已投遞到消費者并完成業(yè)務(wù)處理，當(dāng)客戶端給服務(wù)端反饋應(yīng)答的時候網(wǎng)絡(luò)閃斷。為了保證消息至少被消費一次，消息隊列的服務(wù)端將在網(wǎng)絡(luò)恢復(fù)后再次嘗試投遞之前已被處理過的消息，消費者后續(xù)會收到兩條內(nèi)容相同并且 Message ID 也會收到相同的消息。
• 負載均衡時消息重復(fù)（包括但不限于網(wǎng)絡(luò)抖動、Broker 重啟以及消費者應(yīng)用重啟）。當(dāng)消息隊列的 Broker 或客戶端重啟、擴容或縮容時，會觸發(fā) Rebalance，此時消費者可能會收到重復(fù)消息。

搞搞具體方案

還是順著前面扣款的例子說。

收到消息之后，我們第一步一般來說是保存信息到數(shù)據(jù)庫。

save(扣款信息);

現(xiàn)在我們要做冪等，已經(jīng)找到了扣款唯一流水號這個抓手，那我們的代碼應(yīng)該怎么寫呢？

扣款信息?=?select(扣款唯一流水號);
if(扣款信息?==?null){
????save(扣款信息);
}

先查詢，再判斷，最后保存。

這個方案，在一般的情況下，能達到冪等的效果。

但是，由于是三步，在并發(fā)場景下，立馬就扛不住了。

而且，消息重復(fù)投遞的場景，本來就是在極短的時間內(nèi)產(chǎn)生的兩條信息。

所以，上面這個方案會出現(xiàn)什么場景呢？

兩個請求，在 select(扣款唯一流水號) 的時候都沒有查詢到數(shù)據(jù)，擊穿了校驗邏輯，然后兩個請求就都會去落庫。

這個時候怎么辦呢？

很簡單，前面說了，在扣款唯一流水號上加唯一索引，即使兩個請求都去落庫，但是由于有唯一索引，一定只會落一筆數(shù)據(jù)到數(shù)據(jù)庫。

另外一個怎么辦？

拋出唯一索引沖突的異常，在程序里面通過捕獲這個異常來控制流程上的后續(xù)運轉(zhuǎn)。

這個方案，很常見，很常用，實話實說我們用的就是這個方案。

但是既然已經(jīng)有唯一索引了，那是不是前面的 select 都顯得沒啥卵用了？

我們要從辯證的角度去看待這個問題。

所以，是，也不是。

是的原因是因為前面這一層 select 相當(dāng)于過濾層，能在一些非并發(fā)的場景下讓程序不拋出唯一索引沖突的異常，顯得更加優(yōu)雅。

不是的原因是因為優(yōu)雅的程度還不夠高，畢竟是通過“異�！眮砜刂屏顺绦虻淖呦�。

有沒有不拋出異常的方案呢？

也有，也很簡單，上鎖就行了：

扣款信息 = select(扣款唯一流水號);//select *** for update

這樣確實能保證不拋出唯一索引沖突的異常，但是關(guān)鍵是一旦涉及到上鎖，性能就拉胯了，為了解決這個偶發(fā)的問題，犧牲了接口的性能，這個路線就走的有點遠了。

所以，上鎖也不夠優(yōu)雅。

什么是真正的優(yōu)雅？

我也不知道，但是我試圖去思考一個相對優(yōu)雅的方案。

思考一波

首先，我覺得上面的方案，不管是唯一索引，還是上鎖，不夠優(yōu)雅的原因是因為，它們都是在基于業(yè)務(wù)表搞事情。

業(yè)務(wù)表干得事兒，應(yīng)該就是業(yè)務(wù)上的事兒。

那我問你：消息重復(fù)投遞，需要保持冪等，這個屬于業(yè)務(wù)上的事兒嗎？

我認為是不屬于的，這是屬于技術(shù)上的事情，任何業(yè)務(wù)都是可能遇到的。只不過，在前面的方案里面，我們想借用業(yè)務(wù)表的能力，來幫我們做一個它可以做，但是本來不該它做的事情。

首先，我們必須要在這一點上達成一致，不然后面的論述就不能展開了。

如果你不這樣認為，那么你可以不用往下看。

我想到的方案是什么呢？

我相信你聽過這樣一句話：計算機領(lǐng)域中的所有問題都可以通過增加一個中間層來解決。

所以，我也想著抽一層。

我還是需要數(shù)據(jù)庫通過唯一索引來幫我保證只有一條數(shù)據(jù)被成功落庫，所以我想著抽一個專門的表出來，比如叫做消息消費記錄表。

只要數(shù)據(jù)插入了這個表，就代表消息被消費了。后續(xù)即使重發(fā)，也不會插入成功。

那么怎么來保證這個機制呢？

前面提到的抓手又可以用上了：業(yè)務(wù)唯一流水號。

這個消息消費記錄表里面最重要的一個字段，可以叫做“消息唯一標(biāo)識”，并且作為唯一索引。

這里的這個“消息唯一標(biāo)識”就是對應(yīng)業(yè)務(wù)唯一流水號。

如果你要基于這個表來實現(xiàn)消息冪等，那么你必須具備這樣的一個業(yè)務(wù)唯一流水號，當(dāng)重復(fù)的時候，還是會拋出主鍵沖突異常。

我知道著聽起來就像是脫褲子放屁。

但是，你想想，這個表是完全脫離于業(yè)務(wù)的存在。

在前面的解決方案中，你要問別人，你有沒有一張業(yè)務(wù)表來做這個事情。

在現(xiàn)在的方案中，你會給別人說，我這里有一個解決方案，你只需要執(zhí)行我給你的 SQL，生成一張消息消費記錄表就行。

這張表是完全獨立于業(yè)務(wù)的存在，它只是為了解決消息重復(fù)投遞這個共性問題。

從你問別人要，到別人按照你說的做，就這么輕輕的抽一小層，攻守易形了啊，朋友。

它是一種通用的解決方案，一種策略，甚至可以叫做一個框架。

現(xiàn)在，我們可以給它取一個新的名字。

比如：一種基于數(shù)據(jù)庫唯一索引實現(xiàn)消息冪等的解決方案。

或者：一種分布式系統(tǒng)中數(shù)據(jù)唯一性的保障策略。

再或者：一個由數(shù)據(jù)庫約束驅(qū)動的消息冪等保護性框架。

好，現(xiàn)在我們有了這么一個“高大上”的通用解決方案了。

到底怎么用呢？

名字很厲害，但是用起來其實也就那么回事兒。

回到前面轉(zhuǎn)賬的例子，很簡單：

if(保存數(shù)據(jù)到消息消費記錄表){//出現(xiàn)主鍵沖突就返回false
????save(扣款信息);????
}

這樣，消息防重，由消息消費記錄表來保證。

業(yè)務(wù)表，不感知“消息是否重復(fù)”的場景。

看起來似乎是優(yōu)雅了那么一點點。

但是，同時帶來了另外一個問題。

又回到了之前“先校驗，再保持”的非原子性的邏輯。

我們想想一個極端場景，如果保存數(shù)據(jù)到消息消費記錄表成功，還沒來得及 save(扣款信息) ，服務(wù)重啟了，怎么辦？

其實換句話說，這兩個信息需要保持一致性。

所以可以加入事務(wù)嘛，把這兩步綁定到一起：

開啟事務(wù)；
if(保存數(shù)據(jù)到消息消費記錄表){//出現(xiàn)主鍵沖突就返回false
????save(扣款信息);????
}
提交事務(wù)；

這樣，如果保存數(shù)據(jù)到消息消費記錄表成功，還沒來得及 save(扣款信息) ，服務(wù)重啟，事務(wù)回滾，消息消費記錄表就不會真的插入成功。

而 MQ 沒有收到這個消息的回執(zhí)，也會再次進行投遞。

由于消息消費記錄表里沒有這個數(shù)據(jù)，所以會再次進行消費。

在上面的這個過程中，MQ 再次投遞，是為了 at least once。

而我們引入了消息消費記錄表，通過唯一索引來保證不重復(fù)消費，這個玩意加上 at least once，在業(yè)界有另外一個叫法： exactly-only。

現(xiàn)在，我們通過引入事務(wù)來解決了“非原子性”的問題，但是又帶來另外一個問題：事務(wù)。

一般來說，大家都是能不使用事務(wù)的地方就盡量不使用事務(wù)，通過最終一致性來保證數(shù)據(jù)的完整性。

那現(xiàn)在有沒有不基于事務(wù)的解決方案呢？

我想到的是可以在消息消費記錄表里面再引入一個“狀態(tài)字段”，這個字段有三個取值：未消費、消費中、消費完成。

通過維護狀態(tài)的流轉(zhuǎn)，來代替事務(wù)的邏輯。

這個思路來源于我實習(xí)的時候，給老師做外包項目。當(dāng)時我是真的不知道 Spring 的事務(wù)怎么用，但是我知道結(jié)合當(dāng)時我開發(fā)的業(yè)務(wù)場景，一個數(shù)據(jù)的狀態(tài)很重要，處理之前把數(shù)據(jù)的狀態(tài)修改了，但是如果出了異常，應(yīng)該把狀態(tài)給它恢復(fù)回去。

于是我手動寫了這樣的一坨代碼，四處散落在我寫的模塊里面。

后來一個師兄看了我的代碼，提出了應(yīng)該用事務(wù)來保證這樣的邏輯，并給我做了演示，我才去了解了事務(wù)相關(guān)的東西。

但是有一說一，我后來也思考了，在我那個特定的業(yè)務(wù)場景下，通過狀態(tài)的流轉(zhuǎn)，確實是可以代替事務(wù)的存在的。

好，回到我們現(xiàn)在的這個場景中。

一個消息過來的時候，首先根據(jù)唯一消息標(biāo)識獲取對應(yīng)的數(shù)據(jù)。

如果沒有獲取到，就初始化為“未消費”狀態(tài)落庫，然后去執(zhí)行具體的業(yè)務(wù)邏輯。在業(yè)務(wù)邏輯執(zhí)行之前，把狀態(tài)修改為“消費中”，然后在執(zhí)行完成之后，把狀態(tài)修改為“消費完成”。

如果這個消息被重新投遞了，那么根據(jù)唯一消息標(biāo)識就能獲取到對應(yīng)的數(shù)據(jù)，接著檢查這個消息的狀態(tài)。如果是“消費完成”，直接就丟掉。

但是上面的描述只是描述了最簡單的場景，一些復(fù)雜場景下狀態(tài)的流轉(zhuǎn)和判斷應(yīng)該怎么做，我確實還沒想好。

所以就當(dāng)是個課后習(xí)題吧，你去推一推，看看用狀態(tài)流轉(zhuǎn)代替事務(wù)的方式是否能成功落地。

學(xué)會了記得回來教我。