前面對于分布式事務(wù)也講了好幾篇了（可靠消息最終一致性分布式事務(wù) - TCC分布式事務(wù) - 2PC、3PC
https://github.com/kklldog/AgileDT 開源不易，大家多多 ???

回顧

前面一篇文章(可靠消息最終一致性 )我們詳細(xì)介紹了基于可靠消息的分布式事務(wù)。為了更好的理解 AgileDT 的代碼，我們還是有必要簡單的來回顧下。

該方案總體流程上可分為以下步驟：

主動方在真正的業(yè)務(wù)開始前先向可靠消息服務(wù)發(fā)送一個“待確認(rèn)”的消息

可靠消息服務(wù)收到待確認(rèn)消息后持久化消息到數(shù)據(jù)庫

如果以上操作成功則主動方開始真正的業(yè)務(wù)，如果失敗則直接放棄執(zhí)行業(yè)務(wù)

如果業(yè)務(wù)執(zhí)行成功則發(fā)送“確認(rèn)”消息給可靠消息服務(wù)，如果執(zhí)行失敗則發(fā)送“取消”給可靠消息服務(wù)。

如果可靠消息服務(wù)收到“確認(rèn)”消息則更新數(shù)據(jù)庫里的消息記錄的狀態(tài)為“待發(fā)送”，如果收到的消息為“取消”則更新消息狀態(tài)為“已取消”

如果上一步更新的數(shù)據(jù)庫為“待發(fā)送”，那么會開始往MQ投遞消息，并且更改數(shù)據(jù)庫里的消息記錄的狀態(tài)為“已發(fā)送”

上一步往MQ投遞消息成功后，MQ會給被動方推送消息。

被動方收到消息后開始處理業(yè)務(wù)

如果業(yè)務(wù)處理成功，則被動方對MQ進(jìn)行ACK回復(fù)，則這條消息會從MQ內(nèi)移除掉

如果業(yè)務(wù)處理成功，則發(fā)送“已完成”消息給可靠消息服務(wù)

可靠消息服務(wù)收到“已完成”消息后更新數(shù)據(jù)庫消息記錄未“已完成”

廢話不多說了，下面讓我們演示下如何使用 AgileDT 來快速實現(xiàn)一個基于可靠消息的分布式事務(wù)。
以下我們還是以經(jīng)典的訂單下單完成給會員贈送積分的場景來演示。

使用 AgileDT

依賴組件

• mysql

• rabbitmq

目前支持 mysql 數(shù)據(jù)庫，但是數(shù)據(jù)訪問組件使用的是 freesql 所以后續(xù)要實現(xiàn)支持別的數(shù)據(jù)庫也很簡單。目前框架使用的可靠消息服務(wù)為 rabbitmq 。

運行服務(wù)端

在服務(wù)新建一個數(shù)據(jù)庫并且新建一張表

//?crate?event_message?table?on?mysql create?table?if?not?exists?event_message (event_id?varchar(36)?not?nullprimary?key,biz_msg?varchar(4000)?null,status?enum('Prepare',?'Done',?'WaitSend',?'Sent',?'Finish',?'Cancel')?not?null,create_time?datetime(3)?null,event_name?varchar(255)?null );

使用docker-compose運行服務(wù)端

version:?"3"??#?optional?since?v1.27.0 services:agile_dt:image:?"kklldog/agile_dt"ports:-?"5000:5000"environment:-?db:provider=mysql-?db:conn=?Database=agile_dt;Data?Source=192.168.0.115;User?Id=root;Password=mdsd;port=3306-?mq:userName=admin-?mq:password=123456-?mq:host=192.168.0.115-?mq:port=5672

安裝客戶端

在主動方跟被動方都需要安裝AgileDT的客戶端庫

Install-Package?AgileDT.Client

主動方使用方法

在業(yè)務(wù)數(shù)據(jù)庫添加事務(wù)消息表

//?crate?event_message?table?on?mysql create?table?if?not?exists?event_message (event_id?varchar(36)?not?nullprimary?key,biz_msg?varchar(4000)?null,status?enum('Prepare',?'Done',?'WaitSend',?'Sent',?'Finish',?'Cancel')?not?null,create_time?datetime(3)?null,event_name?varchar(255)?null );

修改配置文件

在appsettings.json文件添加以下節(jié)點："agiledt":?{"server":?"http://localhost:5000","db":?{"provider":?"mysql","conn":?"Database=agile_order;Data?Source=192.168.0.125;User?Id=dev;Password=dev@123f;port=13306"//"conn":?"Database=agile_order;Data?Source=192.168.0.115;User?Id=root;Password=mdsd;port=3306"},"mq":?{"host":?"192.168.0.125",//"host":?"192.168.0.115","userName":?"admin","password":?"123456","port":?5672}}

注入 AgileDT 客戶端服務(wù)

public?void?ConfigureServices(IServiceCollection?services){services.AddAgileDT();...}

實現(xiàn)IEventService方法
處理主動方業(yè)務(wù)邏輯的類需要實現(xiàn)IEventService接口，并且標(biāo)記那個方法是真正的業(yè)務(wù)方法。AgileDT在啟動的時候會掃描這些類型，并且使用AOP技術(shù)生成代理類，在業(yè)務(wù)方法前后插入對應(yīng)的邏輯來跟可靠消息服務(wù)通訊。這里要注意的幾個地方：

• 實現(xiàn)IEventService接口

• 使用DtEventBizMethod注解標(biāo)記業(yè)務(wù)入口方法

• 使用DtEventName注解來標(biāo)記事務(wù)的方法名稱，如果不標(biāo)記則使用類名

注意：業(yè)務(wù)方法最終一定要使用事務(wù)來同步修改消息表的status字段為done狀態(tài)，這個操作框架沒辦法幫你實現(xiàn)
注意：業(yè)務(wù)方法如果失敗請拋出Exception，如果不拋異常框架一律認(rèn)為執(zhí)行成功

public?interface?IAddOrderService:IEventService{bool?AddOrder(Order?order);}[DtEventName("orderservice.order_added")]public?class?AddOrderService?:?IAddOrderService{private?readonly?ILogger<AddOrderService>?_logger;public?AddOrderService(ILogger<AddOrderService>?logger){_logger?=?logger;}public?string?EventId?{?get;set;}[DtEventBizMethod]public?virtual?bool?AddOrder(Order?order){var?ret?=?false;//3.?寫?Order?跟?修改?event?的狀態(tài)必選寫在同一個事務(wù)內(nèi)FreeSQL.Instance.Ado.Transaction(()?=>{order.EventId?=?EventId;//在訂單表新增一個eventid字段，使order跟event_message表關(guān)聯(lián)起來var?ret0?=?FreeSQL.Instance.Insert(order).ExecuteAffrows();var?ret1?=?FreeSQL.Instance.Update<OrderService.Data.entities.EventMessage>().Set(x?=>?x.Status,?MessageStatus.Done).Where(x?=>?x.EventId?==?EventId).ExecuteAffrows();ret?=?ret0?>?0?&&?ret1?>?0;});return?ret;}///?<summary>///?構(gòu)造后續(xù)業(yè)務(wù)處理需要的消息內(nèi)容///?</summary>///?<returns></returns>public?string?GetBizMsg(){//這里可以構(gòu)造傳遞到MQ的業(yè)務(wù)消息的內(nèi)容，比如傳遞訂單編號啊?,以便后續(xù)的被動方處理業(yè)務(wù)時候使用var?order?=?FreeSQL.Instance.Select<Order>().Where(x?=>?x.EventId?==?EventId).First();return?order?.Id;}}

在實現(xiàn)好 IAddOrderService 接口后，你可以像平常一樣使用 IAddOrderService 來注入實現(xiàn)類。比如在 Controller 的構(gòu)造函數(shù)注入進(jìn)去。因為 AgileDT 在啟動的時候會自動幫你注冊。

注意：IAddOrderService 跟實現(xiàn)類的生命周期是 Scoped 。

被動方使用方法

在業(yè)務(wù)方數(shù)據(jù)庫建表或者在業(yè)務(wù)表上加字段
對于被動方來說這里不是必須要建一個表。但是至少要有個地方來存儲event_id的信息，最簡單的是直接在業(yè)務(wù)主表上加event_id字段。

修改配置文件

在appsettings.json文件添加以下節(jié)點："agiledt":?{"server":?"http://localhost:5000","db":?{"provider":?"mysql","conn":?"Database=agile_order;Data?Source=192.168.0.125;User?Id=dev;Password=dev@123f;port=13306"//"conn":?"Database=agile_order;Data?Source=192.168.0.115;User?Id=root;Password=mdsd;port=3306"},"mq":?{"host":?"192.168.0.125",//"host":?"192.168.0.115","userName":?"admin","password":?"123456","port":?5672}}

注入AgileDT服務(wù)

public?void?ConfigureServices(IServiceCollection?services){services.AddAgileDT();...}

實現(xiàn)IEventMessageHandler接口
被動方需要接收MQ投遞過來的消息，這些處理類需要實現(xiàn)IEventMessageHandler接口。AgileDT啟動的時候會去掃描這些類，然后跟MQ建立綁定關(guān)系。

• 這里必須使用DtEventName注解標(biāo)記需要處理的事件名稱

• Reveive 方法必須是冪等的

public?interface?IOrderAddedMessageHandler:?IEventMessageHandler{}[DtEventName("orderservice.order_added")]public?class?OrderAddedMessageHandler:?IOrderAddedMessageHandler{static?object?_lock?=?new?object();public?bool?Receive(EventMessage?message){var?bizMsg?=?message.BizMsg;var?eventId?=?message.EventId;string?orderId?=?bizMsg;lock?(_lock){var?entity?=?FreeSQL.Instance.Select<PointHistory>().Where(x?=>?x.EventId?==?eventId).First();if?(entity?==?null){var?ret?=?FreeSQL.Instance.Insert(new?PointHistory{Id?=?Guid.NewGuid().ToString(),EventId?=?message.EventId,OrderId?=?orderId,Points?=?20,CreateTime?=?DateTime.Now}).ExecuteAffrows();return?ret?>?0;}else{return?true;}}}}

總結(jié)

通過以上演示，我們快速的實現(xiàn)了一個訂單下單會員贈送積分的服務(wù)。可以看到使用 AgileDT 可以很快速的實現(xiàn)一個分布式事務(wù)。特別是在實現(xiàn)過一個分布式事務(wù)后，后面實現(xiàn)起來就特別簡單，只要實現(xiàn)幾個接口就可以了。AgileDT 才剛剛起步，希望大家多多支持，多多??? ?，多多 PR 。

https://github.com/kklldog/AgileDT

.Net Core with 微服務(wù) - 可靠消息最終一致性分布式事務(wù)

.Net Core with 微服務(wù) - 分布式事務(wù) - TCC

.Net Core with 微服務(wù) - 分布式事務(wù) - 2PC、3PC

總結(jié)

以上是生活随笔為你收集整理的.Net Core with 微服务 - 使用 AgileDT 快速实现基于可靠消息的分布式事务的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。