如何用C#實現SAGA分布式事務
背景
銀行跨行轉賬業(yè)務是一個典型分布式事務場景，假設 A 需要跨行轉賬給 B，那么就涉及兩個銀行的數據，無法通過一個數據庫的本地事務保證轉賬的 ACID ，只能夠通過分布式事務來解決。
市面上使用比較多的分布式事務框架，支持 SAGA 的，大部分都是 JAVA 為主的，沒有提供 C# 的對接方式，或者是對接難度大，一定程度上讓人望而卻步。
下面就基于這個框架來實踐一下銀行轉賬的例子。
前置工作
dotnet add package Dtmcli --version 0.3.0
成功的 SAGA
先來看一下一個成功完成的 SAGA 時序圖。
上圖的微服務1，對應我們示例的 OutApi，也就是轉錢出去的那個服務。
微服務2，對應我們示例的 InApi，也就是轉錢進來的那個服務。
下面是兩個服務的正向操作和補償操作的處理。
OutApi
app.MapPost("/api/TransOut", (string branch_id, string gid, string op, TransRequest req) =>
{
// 進行 數據庫操作
Console.WriteLine($"用戶【{req.UserId}】轉出【{req.Amount}】正向操作，gid={gid}, branch_id={branch_id}, op={op}");
return Results.Ok(TransResponse.BuildSucceedResponse());
});
app.MapPost("/api/TransOutCompensate", (string branch_id, string gid, string op, TransRequest req) =>
{
// 進行 數據庫操作
Console.WriteLine($"用戶【{req.UserId}】轉出【{req.Amount}】補償操作，gid={gid}, branch_id={branch_id}, op={op}");
return Results.Ok(TransResponse.BuildSucceedResponse());
});
InApi
app.MapPost("/api/TransIn", (string branch_id, string gid, string op, TransRequest req) =>
{
Console.WriteLine($"用戶【{req.UserId}】轉入【{req.Amount}】正向操作，gid={gid}, branch_id={branch_id}, op={op}");
return Results.Ok(TransResponse.BuildSucceedResponse());
});
app.MapPost("/api/TransInCompensate", (string branch_id, string gid, string op, TransRequest req) =>
{
Console.WriteLine($"用戶【{req.UserId}】轉入【{req.Amount}】補償操作，gid={gid}, branch_id={branch_id}, op={op}");
return Results.Ok(TransResponse.BuildSucceedResponse());
});
注：示例為了簡單，沒有進行實際的數據庫操作。
到此各個子事務的處理已經 OK 了，然后是開啟 SAGA 事務，進行分支調用
var userOutReq = new TransRequest() { UserId = "1", Amount = -30 };
var userInReq = new TransRequest() { UserId = "2", Amount = 30 };
var ct = new CancellationToken();
var gid = await dtmClient.GenGid(ct);
var saga = new Saga(dtmClient, gid)
.Add(outApi + "/TransOut", outApi + "/TransOutCompensate", userOutReq)
.Add(inApi + "/TransIn", inApi + "/TransInCompensate", userInReq)
;
var flag = await saga.Submit(ct);
Console.WriteLine($"case1, {gid} saga 提交結果 = {flag}");
到這里，一個完整的 SAGA 分布式事務就編寫完成了。
搭建好 dtm 的環(huán)境后，運行上面的例子，會看到下面的輸出。
當然，上面的情況太理想了，轉出轉入都是一次性就成功了。
但是實際上我們會遇到許許多多的問題，最常見的應該就是網絡故障了。
下面來看一個異常的 SAGA 示例
異常的 SAGA
做一個假設，用戶1的轉出是正常的，但是用戶2在轉入的時候出現了問題。
由于事務已經提交給 dtm 了，按照 SAGA 事務的協議，dtm 會重試未完成的操作。
這個時候用戶2 這邊會出現什么樣的情況呢？
轉入其實成功了，但是 dtm 收到錯誤 (網絡故障等)轉入沒有成功，直接告訴 dtm 失敗了 (應用異常等)
無論是那一種，dtm 都會進行重試操作。這個時候會發(fā)生什么呢？我們繼續(xù)往下看。
先看一下事務失敗交互的時序圖
再通過調整上面成功的例子，來比較直觀的看看出現的情況。
在 InApi 加多一個轉入失敗的處理接口
app.MapPost("/api/TransInError", (string branch_id, string gid, string op, TransRequest req) =>
{
Console.WriteLine($"用戶【{req.UserId}】轉入【{req.Amount}】正向操作--失敗，gid={gid}, branch_id={branch_id}, op={op}");
//return Results.BadRequest();
return Results.Ok(TransResponse.BuildFailureResponse());
});
失敗的返回有兩種，一種是狀態(tài)碼大于 400，一種是狀態(tài)碼是 200 并且響應體包含 FAILURE，上面的例子是第二種
調整一下調用方，把轉入正向操作替換成上面這個返回錯誤的接口。
var saga = new Saga(dtmClient, gid)
.Add(outApi + "/TransOut", outApi + "/TransOutCompensate", userOutReq)
.Add(inApi + "/TransInError", inApi + "/TransInCompensate", userInReq);
運行結果如下：
在這個例子中，只考慮補償/重試成功的情況下。
用戶1 轉出的 30 塊錢最終是回到了他的帳號上，他沒有出現損失。
用戶2 就有點苦逼了，轉入沒有成功，返回了失敗，還觸發(fā)了轉入的補償機制，結果就是把用戶2 還沒進帳的 30 塊錢給多扣了，這個就是上面的情況2，常見的空補償問題。
這個時候就要在進行轉入補償的時候做一系列的判斷，轉入有沒有成功，轉出有沒有失敗等等，把業(yè)務變的十分復雜。
如果出現了上述的情況1，會發(fā)生什么呢？
用戶2 第一次已經成功轉入 30 塊錢，返回的也是成功，但是網絡出了點問題，導致 dtm 認為失敗了，它就會進行重試，相當于用戶2 還會收到第二個轉入 30 塊錢的請求！也就是說這次轉帳，用戶2 會進賬 60 塊錢，翻倍了，也就是說這個請求不是冪等。
同樣的，要處理這個問題，在進行轉入的正向操作中也要進行一系列的判斷，同樣會把復雜度上升一個級別。
前面有提到 dtm 提供了子事務屏障的功能，保證了冪等、空補償等常見問題。
再來看看這個子事務屏障的功能有沒有幫我們簡化上面異常處理。
子事務屏障
子事務屏障，需要根據 trans_type，gid，branch_id 和 op 四個內容進行創(chuàng)建。
這4個內容 dtm 在回調時會放在 querysting 上面。
客戶端里面提供了 IBranchBarrierFactory 來供我們使用。
空補償
針對上面的異常情況(用戶2 憑空消失 30 塊錢)，對轉入的補償進行子事務屏障的改造。
app.MapPost("/api/BarrierTransInCompensate", async (string branch_id, string gid, string op, string trans_type, TransRequest req, IBranchBarrierFactory factory) =>
{
var barrier = factory.CreateBranchBarrier(trans_type, gid, branch_id, op);
using var db = Db.GeConn();
await barrier.Call(db, async (tx) =>
{
// 轉入失敗的情況下，不應該輸出下面這個
Console.WriteLine($"用戶【{req.UserId}】轉入【{req.Amount}】補償操作，gid={gid}, branch_id={branch_id}, op={op}");
// tx 參數是事務，可和本地事務一起提交回滾
await Task.CompletedTask;
});
Console.WriteLine($"子事務屏障-補償操作，gid={gid}, branch_id={branch_id}, op={op}");
return Results.Ok(TransResponse.BuildSucceedResponse());
});
Call 方法就是關鍵所在了，需要傳入一個 DbConnection 和真正的業(yè)務操作，這里的業(yè)務操作就是在控制臺輸出補償操作的信息。
同樣的，我們再調整一下調用方，把轉入補償操作替換成上面帶子事務屏障的接口。
var saga = new Saga(dtmClient, gid)
.Add(outApi + "/TransOut", outApi + "/TransOutCompensate", userOutReq)
.Add(inApi + "/TransInError", inApi + "/BarrierTransInCompensate", userInReq)
;
再來運行這個例子。
會發(fā)現轉入的補償操作并沒執(zhí)行，控制臺沒有輸出補償信息，而是輸出了
Will not exec busiCall, isNullCompensation=True, isDuplicateOrPend=False
這個就表明了，這個請求是個空補償，是不應該執(zhí)行業(yè)務方法的，既空操作。
再來看一下，轉入成功的，但是 dtm 收到了失敗的信號，不斷重試造成重復請求的情況。
冪等
針對用戶2 轉入兩次 30 塊錢的異常情況，對轉入的正向操作進行子事務屏障的改造。
app.MapPost("/api/BarrierTransIn", async (string branch_id, string gid, string op, string trans_type, TransRequest req, IBranchBarrierFactory factory) =>
{
Console.WriteLine($"用戶【{req.UserId}】轉入【{req.Amount}】請求來了?。?！ gid={gid}, branch_id={branch_id}, op={op}");
var barrier = factory.CreateBranchBarrier(trans_type, gid, branch_id, op);
using var db = Db.GeConn();
await barrier.Call(db, async (tx) =>
{
var c = Interlocked.Increment(ref _errCount);
// 模擬一個超時執(zhí)行
if (c > 0 && c < 2) await Task.Delay(10000);
Console.WriteLine($"用戶【{req.UserId}】轉入【{req.Amount}】正向操作，gid={gid}, branch_id={branch_id}, op={op}");
await Task.CompletedTask;
});
return Results.Ok(TransResponse.BuildSucceedResponse());
});
這里通過一個超時執(zhí)行來讓 dtm 進行轉入正向操作的重試。
同樣的，我們再調整一下調用方，把轉入的正向操作也替換成上面帶子事務屏障的接口。
var saga = new Saga(dtmClient, gid)
.Add(outApi + "/TransOut", outApi + "/TransOutCompensate", userOutReq)
.Add(inApi + "/BarrierTransIn", inApi + "/BarrierTransInCompensate", userInReq)
;
再來運行這個例子。
可以看到轉入的正向操作確實是觸發(fā)了多次，第一次實際上是成功，只是響應比較慢，導致 dtm 認為是失敗了，觸發(fā)了第二次請求，但是第二次請求并沒有執(zhí)行業(yè)務操作，而是輸出了
Will not exec busiCall, isNullCompensation=False, isDuplicateOrPend=True
這個就表明了，這個請求是個重復請求，是不應該執(zhí)行業(yè)務方法的，保證了冪等。
到這里，可以看出，子事務屏障確實解決了冪等和空補償的問題，大大降低了業(yè)務判斷的復雜度和出錯的可能性。
寫在最后
在這篇文章里，也通過幾個例子，完整給出了編寫一個 SAGA 事務的過程，涵蓋了正常成功完成，異常情況，以及成功回滾的情況。希望對研究分布式事務的您有所幫助。
本文示例代碼： DtmSagaSample
到此這篇關于如何用C#實現SAGA分布式事務的文章就介紹到這了,更多相關C#實現SAGA內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
這篇文章主要介紹了C#創(chuàng)建及讀取DAT文件操作，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-12-12
這兩天做了一個圖片對比工具，里面要處理兩張大的圖片，所以要對圖片先進行裁剪,下面看看我的方法吧2013-12-12
在本篇文章里小編給大家整理的是關于C#實現排序的代碼以及相關知識點，需要的朋友們參考下。2019-10-10
本篇文章通過截圖的方式向大家展示C#程序登陸實現的全過程,利用了C#三層架構的編寫方法，希望對大家今后編寫代碼有所幫助2016-11-11
這篇文章主要介紹了C# 獲取客戶端IPv4地址的示例代碼，幫助大家更好的理解和使用c#,感興趣的朋友可以了解下2020-12-12
給大家分享了C#中comboBox實現三級聯動的全部代碼，代碼經過測試，有興趣的朋友跟著做一下。2018-03-03
這篇文章主要介紹了C#生成餅形圖及添加文字說明的方法,非常實用的功能,需要的朋友可以參考下2014-07-07
這篇文章主要介紹了如何利用WPF實現雷達圖(仿英雄聯盟)的繪制，文中的示例代碼講解詳細，對我們學習或工作有一定幫助，需要的可以參考一下2022-07-07
這篇文章介紹了C#集合之可觀察集合的用法，文中通過示例代碼介紹的非常詳細。對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2022-04-04
數組從字面上理解就是存放一組數,下面這篇文章主要給大家介紹了關于如何用C#找出數組中只出現了一次的數字,文中通過實例代碼介紹的非常詳細,需要的朋友可以參考下2022-12-12