如何能在實(shí)戰(zhàn)中完成分布式事務(wù)
背景
在一年前我寫過一篇關(guān)于分布式事務(wù)的文章：
再有人問你分布式事務(wù)，把這篇扔給他，在這篇文章中我詳細(xì)介紹了分布式事務(wù)是什么，實(shí)現(xiàn)分布式事務(wù)有哪些常用的方案，但是其中的東西很多是偏于理論，很多讀者對(duì)其真正在實(shí)戰(zhàn)上的使用可能還是有點(diǎn)差距。所以在前幾次文章的更新中，我介紹了很多關(guān)于Seata（一款由阿里開源的分布式事務(wù)框架）的內(nèi)容，如果大家對(duì)Seata不是很熟悉的可以閱讀下面的內(nèi)容：
Seata已經(jīng)為我們提供了兩種實(shí)現(xiàn)分布式模式：
AT：自動(dòng)模式，通過我們記錄運(yùn)行sql的undolog，來完成事務(wù)失敗時(shí)的自動(dòng)重做。
TCC：TCC模式，這種模式彌補(bǔ)我們AT模式只能支持ACID數(shù)據(jù)庫(kù)的場(chǎng)景。
大多數(shù)時(shí)候Seata已經(jīng)足夠了，但是很多時(shí)候不同場(chǎng)景下我們沒辦法選擇Seata這類TCC框架：
改造困難，目前Seata支持的通信框架不多只有Dubbo和Spring-Cloud-Alibaba，如果使用的是其他框架，或者直接是簡(jiǎn)單的HTTP，甚至有些公司可能目前系統(tǒng)中都沒有支持Trace。
維護(hù)成本高，Seata需要一個(gè)單獨(dú)的集群去維護(hù)，一般在公司都需要分配一定的資源（人員資源，機(jī)器資源）去管理維護(hù)Seata，很多時(shí)候不可能為了幾個(gè)分布式事務(wù)去花費(fèi)這么大的成本，當(dāng)然這一塊的話未來可以上云解決。
而我最近在做一些分布式事務(wù)的事的時(shí)候也遇到了這些問題，由于一般使用分布式事務(wù)是業(yè)務(wù)方，你需要驅(qū)動(dòng)做RPC組件的同事支持，并且我們并不是純金融服務(wù)的公司，搭建一套類似Seata的分布式事務(wù)中間件也是比較耗費(fèi)資源。
之前介紹的方案大多數(shù)都比較籠統(tǒng)，俗話說授人以魚不如授人以漁，所以接下來我將會(huì)一步一步的教大家如何不用框架，而是我們自己去編碼去實(shí)現(xiàn)分布式事務(wù)。
問題
為了更好的講解如何在實(shí)戰(zhàn)中完成分布式事務(wù)，這里直接舉一個(gè)大家都熟悉的例子：用戶下單的時(shí)候，可以選擇三種資產(chǎn)，分別是儲(chǔ)值余額，積分，券，這個(gè)場(chǎng)景幾乎在每個(gè)應(yīng)用都能看見，而這個(gè)場(chǎng)景在我們的后端可以映射為4個(gè)服務(wù)，如下圖所示：
在這個(gè)場(chǎng)景下大多數(shù)人的代碼基本會(huì)按照下面的寫,在訂單服務(wù)中有如下步驟，這里為了簡(jiǎn)單沒有設(shè)置過多的訂單狀態(tài)：
Step 1：創(chuàng)建訂單狀態(tài)為初始化，并檢查用戶所有資源是否足夠
Step 2：支付儲(chǔ)值余額
Step 3：支付券
Step 4：支付金幣
Step 5：更新訂單狀態(tài)為已完成
差不多這里就是簡(jiǎn)簡(jiǎn)單單4行，有很多人會(huì)把這5步直接放進(jìn)事務(wù)之中，也就是加上@Transactional注解，但其實(shí)加上這個(gè)注解不僅沒有起到事務(wù)作用，而且還讓我們的事務(wù)變成了長(zhǎng)事務(wù)，我們這里的Step2-4都是RPC遠(yuǎn)程調(diào)用，一旦某個(gè)RPC出現(xiàn)了Timeout，那么我們的數(shù)據(jù)庫(kù)連接會(huì)被長(zhǎng)期持有不被釋放，有可能導(dǎo)致我們系統(tǒng)雪崩。
既然這里加上事務(wù)沒有用，我們可以看看會(huì)出現(xiàn)什么問題，如果Step2支付成功，Step3失敗，那么就會(huì)導(dǎo)致數(shù)據(jù)不一致。其實(shí)很多人就會(huì)有僥幸心理，默認(rèn)我們的Step 2-4會(huì)成功，如果出現(xiàn)問題我們?nèi)斯ば迯?fù)就是了。人工修復(fù)的成本太高，你就想如果你在外面旅游突然叫你修復(fù)數(shù)據(jù)，那你是不是會(huì)氣得吐血？所以我們這里一步一步的教大家如何逐漸的把這段業(yè)務(wù)邏輯優(yōu)化成能保證我們數(shù)據(jù)一致的。
方法
一般來說任何一個(gè)分布式事務(wù)框架都離不開三個(gè)關(guān)鍵字：重做記錄，重試機(jī)制，冪等。而在我們的業(yè)務(wù)中同樣也離不開這三個(gè)關(guān)鍵字。
重做記錄
我們想想我們mysql的事務(wù)回滾是依靠什么的？依靠的是undolog，我們的undolog保存了事務(wù)發(fā)生之前的數(shù)據(jù)的一個(gè)版本，那么我們發(fā)生回滾的時(shí)候直接利用這個(gè)版本的數(shù)據(jù)回滾即可。這里我們首先需要添加我們的重做記錄，我們沒必要叫undolog，我們?cè)俑鱾€(gè)資源服務(wù)中需要添加一個(gè)事務(wù)記錄表：
CREATE?TABLE?`transaction_record`?(
??`orderId`?int(11)?unsigned?NOT?NULL?AUTO_INCREMENT,
??`op_total`?int(11)?NOT?NULL?COMMENT?'本次操作資源操作數(shù)量',
??`status`?int(11)?NOT?NULL?COMMENT?'1:代表支付成功?2:代表支付取消',
??`resource_id`?int(11)?NOT?NULL?COMMENT?'本次操作資源的Id',
??`user_id`?int(11)?NOT?NULL?COMMENT?'本次操作資源的用戶Id',
??PRIMARY?KEY?(`orderId`)
)?ENGINE=InnoDB?DEFAULT?CHARSET=utf8mb4?COLLATE=utf8mb4_unicode_ci;
在我們的分布式事務(wù)中有一個(gè)全局事務(wù)ID,而我們orderId就能很好的適應(yīng)這個(gè)角色，這里我們每個(gè)資源的事務(wù)記錄表都需要記錄這個(gè)OrderId，用于和全局事務(wù)進(jìn)行關(guān)聯(lián)，并且我們這里直接將其作為主鍵，也表明了這個(gè)表中只會(huì)出現(xiàn)一次全局事務(wù)ID。這里的op_total用于記錄本次操作資源的數(shù)量，用于后續(xù)回滾，哪怕不回滾我們也可以用于后續(xù)記錄的查詢。status用于記錄我們當(dāng)前這條記錄的狀態(tài)如何，這里用了兩個(gè)狀態(tài)，后續(xù)我們可以擴(kuò)展更多的狀態(tài)，解決更多的分布式事務(wù)問題。
有了這個(gè)重做記錄之后我們只需要在每一次執(zhí)行記錄下我們的當(dāng)前資源的transaction_record，在回滾的時(shí)候根據(jù)我們的OrderId將所有的資源回滾，我們優(yōu)化之后代碼可以如下：
????????int?orderId?=?createInitOrder();
????????checkResourceEnough();
????????try?{
????????????accountService.payAccount(orderId,?userId,?opTotal);
????????????coinService.payCoin(orderId,?userId,?opTotal);
????????????couponService.payCoupon(orderId,?userId,?couponId);
????????????updateOrderStatus(orderId,?PAID);
????????}catch?(Exception?e){
????????????//這里進(jìn)行回滾
????????????accountService.rollback(orderId,?userId);
????????????coinService.rollback(orderId,?userId);
????????????couponService.rollback(orderId,?userId);
????????????updateOrderStatus(orderId,?FAILED);
????????}
這里我們將創(chuàng)建好的初始化訂單，當(dāng)作參數(shù)傳遞給我們的資源服務(wù)記錄，最后再進(jìn)行狀態(tài)更新，如果發(fā)生了異常，那么我們需要進(jìn)行手動(dòng)回滾并將訂單數(shù)據(jù)變?yōu)镕AILED, 回滾的依據(jù)就是我們的訂單Id。對(duì)于我們的支付和回滾的偽代碼有如下：
????@Transactional
????void?payAccount(int?orderId,?int?userId,?int?opTotal){
????????account.payAccount(userId,?opTotal);?//?實(shí)際的去我們account表扣減
????????transactionRecordStorage.save(orderId,?userId,?opTotal,?account.getId());?//保存事務(wù)記錄表
????}
????@Transactional
????void?rollback(int?orderId,?int?userId){
????????TransactionRecord?tr?=?transactionRecordStorage.get(orderId);?//從記錄表中查詢
????????account.rollbackBytr(tr);?//?根據(jù)記錄回滾
????}
這里的版本是比較簡(jiǎn)略的，問題還比較多后面會(huì)講優(yōu)化。
重試機(jī)制
有些同學(xué)可能會(huì)問好像我們上面的代碼基本能保證分布式事務(wù)了吧？的確上面的代碼能保證我們?cè)跊]有宕機(jī)或者其他更加嚴(yán)重的情況下基本上是沒有問題的，但是如果出現(xiàn)了宕機(jī)，比如我們剛剛把a(bǔ)ccount給支付完了，然后支付coin的時(shí)候我們的訂單機(jī)器宕機(jī)了，沒有發(fā)出去這個(gè)請(qǐng)求，這里就不會(huì)走到我們的手動(dòng)回滾請(qǐng)求，所以我們的account將會(huì)永遠(yuǎn)不會(huì)回滾，又只得靠我們的人工回滾，如果你此時(shí)還在旅游，又叫你回滾，估計(jì)你會(huì)繼續(xù)氣暈。或者說我們?cè)倩貪L的時(shí)候出現(xiàn)錯(cuò)誤，怎么辦？我們沒有有效的手段進(jìn)行針對(duì)回滾的回滾。
所以我們需要額外的重試機(jī)制來保證，首先我們需要定義什么樣的數(shù)據(jù)需要重試，這里的話我們根據(jù)業(yè)務(wù)差不多一分鐘能將所有的都資源都支付完，如果我們的訂單狀態(tài)為init 并且 創(chuàng)建時(shí)間超過一分鐘，那么就認(rèn)為發(fā)生了上述錯(cuò)誤的事件。接下來可以通過我們的重試機(jī)制進(jìn)行回滾，這里有兩個(gè)常見重試機(jī)制：
定時(shí)任務(wù)：定時(shí)任務(wù)是我們最常見的重試機(jī)制，基本所有的分布式事務(wù)框架中也都是通過定時(shí)任務(wù)去做的，這里我們需要使用分布式的定時(shí)任務(wù)，分布式的定時(shí)任務(wù)可以使用單機(jī)任務(wù)+分布式鎖 或者 直接使用開源的分布式任務(wù)中間件如elastic-job。我們?cè)诜植际饺蝿?wù)的邏輯中每次查詢我們的處于訂單狀態(tài)為init 并且 創(chuàng)建時(shí)間超過一分鐘的訂單，我們對(duì)其進(jìn)行回滾，回滾完成之后將訂單狀態(tài)置為FAILED。
消息隊(duì)列：目前我們業(yè)務(wù)上使用的是消息隊(duì)列，將下單操作放入消息隊(duì)列中去做，如果我們出現(xiàn)了各種異常，那么我們依靠消息隊(duì)列的重試機(jī)制，一般來說現(xiàn)在當(dāng)前隊(duì)列進(jìn)行重試，再丟給死信隊(duì)列去重試。這里的邏輯就需要改一下，在我們創(chuàng)建訂單的時(shí)候有可能訂單已經(jīng)存在，如果存在的話我們判斷他的狀態(tài)（init+1min）是否應(yīng)該被直接rollback，如果是則直接1min。為什么我們選擇了消息隊(duì)列進(jìn)行重試？因?yàn)槲覀兊臉I(yè)務(wù)邏輯是依靠消息隊(duì)列的，我們就不需要引入定時(shí)任務(wù)，直接依靠消息隊(duì)列即可。
冪等
判斷一個(gè)程序猿經(jīng)驗(yàn)是否老道可以從他寫代碼的時(shí)候能否考慮到冪等就可以看出。很多年輕的程序員根本不會(huì)考慮冪等的存在，甚至都不知道冪等是什么。這里先解釋一下冪等的概念：可以簡(jiǎn)單的認(rèn)為任意多次執(zhí)行所產(chǎn)生的影響和一次執(zhí)行的影響相同。
為什么我們完成分布式事務(wù)的時(shí)候需要冪等？大家可以想想如果在執(zhí)行回滾操作的時(shí)候宕機(jī)了，我們上面的重試機(jī)制就會(huì)開始工作，比如我們的券這個(gè)資源已經(jīng)回滾，但是我們重試操作的時(shí)候我并不知道券已經(jīng)回滾了，這個(gè)時(shí)候就再次嘗試回滾券，如果沒有做冪等操作會(huì)怎么辦，有可能導(dǎo)致用戶資產(chǎn)會(huì)多增加，這樣就會(huì)對(duì)公司造成很多損失。
所以冪等在我們重試的時(shí)候非常重要，實(shí)現(xiàn)冪等的關(guān)鍵是什么？我們想讓多次操作和一次操作是一樣的，那么我們只需要比較第一次已經(jīng)做過了，而這個(gè)標(biāo)記通過什么來完成呢？這里我們可以使用我們狀態(tài)機(jī)轉(zhuǎn)換的手段完成標(biāo)記。只有標(biāo)記這里還是不夠，為什么呢這里我們用個(gè)例子來說明一下，把上面的rollback簡(jiǎn)單優(yōu)化一下：
????@Transactional
????void?rollback(int?orderId,?int?userId){
????????TransactionRecord?tr?=?transactionRecordStorage.get(orderId);
????????if(tr.isCanceled()){
????????????return;?//如果已經(jīng)被取消了那么直接返回
????????}
????????//從記錄表中查詢
????????account.rollbackBytr(tr);?//?根據(jù)記錄回滾
????}
上面代碼我們通過判斷狀態(tài)如果是已經(jīng)被取消了，也就是被回滾了那么我們就直接返回，這里就完成了我們所說的冪等。但是這里還有個(gè)問題是如果有兩個(gè)rollback同時(shí)執(zhí)行怎么辦？你可能會(huì)問什么樣的情況可能會(huì)有兩個(gè)rollback，這里舉一個(gè)場(chǎng)景當(dāng)?shù)谝淮蝦ollback的時(shí)候請(qǐng)求在阻塞了，這個(gè)時(shí)候調(diào)用方已經(jīng)觸發(fā)超時(shí)了，然后一段時(shí)間之后第二次rollback來了，這個(gè)時(shí)候恰好第一次也不阻塞了，那么這里就會(huì)有兩個(gè)rollback請(qǐng)求發(fā)出，當(dāng)執(zhí)行狀態(tài)判斷的時(shí)候，如果兩個(gè)請(qǐng)求同時(shí)執(zhí)行狀態(tài)判斷，那么都會(huì)繞過這個(gè)檢查，最后用戶就會(huì)退兩次錢，這樣的情況我們一定要避免。
那么怎么才能避免呢？聰明的同學(xué)馬上就會(huì)想到使用分布式鎖呀，一提到分布式鎖馬上想到的就是Redis加鎖，ZK加鎖等等，我在這篇文章也做了介紹：聊聊分布式鎖，但是我們這里直接使用數(shù)據(jù)庫(kù)行鎖即可，也就是用下面的sql語句查詢：
select?*?from?transaction?where?orderId?=?"#{orderId}"?for?update;
其他的代碼不變，通過這種形式我們完成了冪等。這時(shí)候有可能會(huì)有同學(xué)會(huì)問到，如果TransactionRecord不存在怎么辦？因?yàn)槲覀冎卦嚨臅r(shí)候我們?cè)趺粗浪腡ry是否成功，我們這里是不知道的，所以我們這里還有策略保證我們的邏輯不會(huì)出現(xiàn)空指針，這里有兩種策略來做這個(gè)事：
如果為空我們直接返回即可。
如果為空，我們保存一條Status為已執(zhí)行空回滾狀態(tài)的TransactionRecord。
上面的第一個(gè)策略比較簡(jiǎn)單，但是我們這里需要選擇第二個(gè)策略，為什么呢因?yàn)槲覀冞€需要預(yù)防一個(gè)事情：防懸掛，我們?cè)僬frollback冪等的時(shí)候，如果第一個(gè)rollback發(fā)生網(wǎng)絡(luò)阻塞，那么這里我們將rollback替換成我們第一次支付的時(shí)候發(fā)生了阻塞，導(dǎo)致了pay在rollback之后到達(dá)我們的客戶端，如果我們采用第一種方式，我們這個(gè)阻塞的Pay請(qǐng)求時(shí)無法感知整個(gè)事務(wù)因?yàn)閞ollback，然后繼續(xù)pay導(dǎo)致我們這個(gè)pay永遠(yuǎn)得不到回滾，這就是懸掛。所以我們這里采用第二個(gè)策略，保存一條記錄，我們?cè)趐ay也會(huì)檢查有沒有這條記錄，所以優(yōu)化之后的代碼為：
????@Transactional
????void?payAccount(int?orderId,?int?userId,?int?opTotal){
????????TransactionRecord?tr?=?transactionRecordStorage.getForUpdate(orderId);
????????if(tr?!=?null){
????????????return;?//如果已經(jīng)有數(shù)據(jù)了，這里直接返回
????????}
????????account.payAccount(userId,?opTotal);?//?實(shí)際的去我們account表扣減
????????transactionRecordStorage.save(orderId,?userId,?opTotal,?account.getId());?//保存事務(wù)記錄表
????}
????@Transactional
????void?rollback(int?orderId,?int?userId){
?????????TransactionRecord?tr?=?transactionRecordStorage.getForUpdate(orderId);
????????if(tr?==?null){
????????????saveNullCancelTr(orderId,?userId);?//保存空回滾的記錄
????????}
????????if(tr.isCanceled()?||?tr.isNullCancel()){
????????????return;?//如果已經(jīng)被取消了那么直接返回
????????}
????????//從記錄表中查詢
????????account.rollbackBytr(tr);?//?根據(jù)記錄回滾
????}
總結(jié)
到這里我們整個(gè)構(gòu)建分布式事務(wù)基本大功告成了，通過這種方式基本上以后遇到相關(guān)分布式事務(wù)的業(yè)務(wù)問題的時(shí)候都可以解決。這里我們?cè)倩仡櫼幌挛覀兊娜齻€(gè)要點(diǎn)：
重試記錄：通過數(shù)據(jù)記錄保存。
重試機(jī)制：定時(shí)任務(wù)或者消息隊(duì)列自帶的重試。
冪等：通過狀態(tài)機(jī)加數(shù)據(jù)庫(kù)行鎖。
我們只要能掌握好這三個(gè)點(diǎn)，其實(shí)不僅僅是對(duì)分布式事務(wù)這一塊有幫助，對(duì)其他的業(yè)務(wù)同樣也有很大的提升。
如果大家覺得這篇文章對(duì)你有幫助，你的關(guān)注和轉(zhuǎn)發(fā)是對(duì)我最大的支持，O(∩_∩)O: