通過Dapr實現一個簡單的基於.net的微服務電商系統(十九)——分布式事務之Saga模式

在領域驅動設計中，由於領域邊界的存在，以往的分層設計中業務會按照其固有的領域知識被切分到不同的限界中，並且引入了領域事件這一概念來降低單個業務的復雜度，通過非耦合的事件驅動來完成復雜的業務。但是事件驅動帶來了一些新的問題，由於以往一個原子性極強的邏輯被拆散到了一個一個小的領域中，原子性事務數據的強一致性無法被保證。為了解決這個問題，一般會采用事務補償的方式來確保最終一致。

當我們采用多個本地事務組合去進行業務處理時，由於業務其本身的復雜性，往往需要在多個事務中協調。而事件協調器(saga)就是一個專門降低其復雜度的設計，開發人員原則上只需要將事件和補償按照一定順序注冊到協調處理器中，原則上協調器會按照注冊的事件依次執行，若出現事件執行失敗時，也會按照補償列表進行相應的回滾。

去年曾經在一篇文章里聊過Saga,鏈接在此。相比去年的那個小DEMO。我在此基礎上基於Dapr的狀態管理完成了另外一個開源項目Oxygen-Saga,地址：https://github.com/sd797994/Oxygen-Saga。當然這個項目既可以引入到dapr的環境中完成Saga事務，本身也可以獨立的基於rabbitmq+redis來完成非Dapr環境下的Saga事務。今天主要講解一下如何在Dapr環境下引入Saga來實現一個分布式事務。

　　首先我們還是看看目前的一個訂單下單邏輯，可以看到基於Actor服務，我們的請求是在訂單服務里直連商品服務Actor完成庫存減扣的，並通過Actor的周期性持久化機制完成事務落庫，所以實際上是把分布式事務的復雜性通過Actor屏蔽掉了，並沒有涉及到真正的分布式事務。

　　從代碼層面也可以看到，整個事務其實是在同一個方法里以同步調用Actor的方式完成的。

        #region 私有遠程服務包裝器方法
        async Task<List<OrderGoodsSnapshot>> GetGoodsListByIds(IEnumerable<Guid> input)
        {
            return (await goodsQueryService.GetGoodsListByIds(new GetGoodsListByIdsDto(input))).GetData<List<OrderGoodsSnapshot>>();
        }
        async Task<bool> DeductionGoodsStock(CreateOrderDeductionGoodsStockDto input)
        {
            var data = input.CopyTo<CreateOrderDeductionGoodsStockDto, DeductionStockDto>();
            data.ActorId = input.GoodsId.ToString();
            return (await goodsActorService.DeductionGoodsStock(data)).GetData<bool>();
        }
        async Task<bool> UnDeductionGoodsStock(CreateOrderDeductionGoodsStockDto input)
        {
            var data = input.CopyTo<CreateOrderDeductionGoodsStockDto, DeductionStockDto>();
            data.ActorId = input.GoodsId.ToString();
            return (await goodsActorService.UnDeductionGoodsStock(data)).GetData<bool>();
        }
        #endregion

        public async Task<ApiResult> CreateOrder(OrderCreateDto input)
        {
            var mockUser = (await accountQueryService.GetMockAccount()).GetData<CurrentUser>();
            //申明一個創建訂單領域服務實例,將遠程rpc調用作為匿名函數傳遞進去
            var createOrderService = new CreateOrderService(GetGoodsListByIds, DeductionGoodsStock, UnDeductionGoodsStock);
            return await ApiResult.Ok("訂單創建成功!").RunAsync(async () =>
            {
                var order = await createOrderService.CreateOrder(mockUser.Id, mockUser.UserName, mockUser.Address, mockUser.Tel, input.Items.CopyTo<OrderCreateDto.OrderCreateItemDto, OrderItem>().ToList());//通過訂單服務創建訂單
                repository.Add(order);
                if (await new CheckOrderCanCreateSpecification(repository).IsSatisfiedBy(order))
                    await unitofWork.CommitAsync();
                await eventBus.SendEvent(EventTopicDictionary.Order.CreateOrderSucc, new OperateOrderSuccessEvent(order, mockUser.UserName));//發送訂單創建成功事件
            },
            //失敗回滾
            createOrderService.UnCreateOrder);
        }

　　而在Saga方案里就變得不一樣了，我們需要借助Saga的流程管理器在多個實例中流轉我們的事務，通過事件訂閱和發布來最終完成一個分布式事務。具體的調用流程如下：