一、案例結構與說明

　　在上一篇中，我們瞭解了MassTransit這個開源元件的基本用法，這一篇我們結合一個小案例來了解在ASP.NET Core中如何藉助MassTransit+Quartz.Net來實現資料的最終一致性。當然，實現資料的最終一致性有很多方案，這裡只是舉一種我所學到的比較簡單易於學習的實現方式而已。

　　假設我們有一個XX保險微信商城（WechatShop，簡稱WS）系統，根據服務的劃分，針對下訂單的這個場景，我們劃分了四個微服務，他們分別是訂單服務（OrderService），庫存服務（StorageService）、配送服務（DeliveryService）以及一個只用於後臺監控事件訊息的事件後臺服務（EventService）。

　　我們來看看這個場景下的一些基本業務邏輯流程（可能並不一定正確或者覆蓋全面）：

　　（1）當終端User在微信商城中看中一個保單產品，並且下單之後，訂單服務會接收訂單資訊並更新訂單資料庫（這裡假設只更新一張訂單表）。

　　（2）然後事件後臺服務會定期（比如每隔30秒）檢查訂單資料庫的事件狀態表（比如一張單獨的Events表，裡面有某個訂單的2行記錄，每行記錄分別代表與訂單相關的服務（這裡就是庫存和配送服務）的處理狀態），如果發現相關服務的事件狀態為未處理，則會向事件匯流排（假設這裡基於RabbitMQ）傳送訊息告知對應訂閱者要處理這個訂單。

　　（3）這裡對應的訂閱者就是庫存服務和配送服務，他們收到訊息之後，會進行各自的業務邏輯處理。比如，庫存服務會根據訂單資訊去更新庫存資料庫並做一些邏輯處理比如更新保單促銷活動的記錄，配送服務會根據訂單資訊更新配送資料庫並做一些邏輯處理比如列印紙質保單並進行物流預約登記，當他們各自處理完成之後便會向事件匯流排傳送一個處理完畢的訊息。

　　（4）事件後臺服務也會作為訂閱者，接收庫存和配送服務傳送過來的訊息，如果接收到某個服務的處理完畢訊息，便會根據接收到的訊息去更新前面事件狀態表中的對應的事件記錄記錄行。比如：接收到庫存服務傳送的訊息，會更新時間狀態表中這個OrderID相關的庫存事件狀態的那一行記錄的狀態為已處理。

　　（5）事件後臺服務的定時任務中（這裡假設每隔30秒一次），會Check事件是否還有未處理完畢的事件訊息，如果沒有則休眠，否則會檢查其建立記錄的時間與現在的系統時間的間隔是否超過了最大容忍值（這裡假設1小時），如果沒有超過則繼續向事件匯流排傳送訊息，如果超過了則進行一些事務回滾逆操作和向管理員傳送一些警告資訊以便進行人工干預等操作。

　　Talk is cheap, Show me the code。下面我們來看看如何實現，由於篇幅原因可能只會列出關鍵性程式碼，詳細程式碼請自行去GitHub上下載或Clone。

　　下面是這次實驗的專案結構，需要準備如下五個專案（四個ASP.NET Core WebAPI和一個.NET Core類庫）

　　資料庫這裡實驗採用的是MSSQL，只建立了一個Order資料庫，兩張表的資料如下：

　　可以看到，在Events表的設計中，通過EventType來區分事件型別，這裡是訂單建立（CreateOrder）這個事件的兩個具體訊息（StorageService和DeliveryService），狀態（StatusValue）為1代表未處理，為2則代表已處理。

二、OrderService的實現

2.1 準備工作

　　其中，Controllers中主要用於與終端使用者（比如WebForm、MVC、SPA等）互動，Models下主要用於定義DTO、EF DbContext與Entity，Repositories目錄下主要用於定義Repository與資料庫進行互動。

2.2 具體實現

　　（1）OrderController

    [Route("api/Order")]
    public class OrderController : Controller
    {
        public IOrderRepository OrderRepository { get; }

        public OrderController(IOrderRepository OrderRepository)
        {
            this.OrderRepository = OrderRepository;
        }

        [HttpPost]
        public string Post([FromBody]OrderDTO orderDTO)
        {
            var result = OrderRepository.CreateOrder(orderDTO).GetAwaiter().GetResult();

            return result ? "Post Success" : "Post Failed";
        }
    }

　　這裡就是簡單的呼叫OrderRepository進行訂單的建立操作。

　　（2）OrderRepository

    public class OrderRepository : IOrderRepository
    {
        public OrderDbContext OrderContext { get; }

        public OrderRepository(OrderDbContext OrderContext)
        {
            this.OrderContext = OrderContext;
        }

        public async Task<bool> CreateOrder(IOrder order)
        {
            var orderEntity = new Order()
            {
                ID = GenerateOrderID(),
                OrderUserID = order.OrderUserID,
                OrderTime = order.OrderTime,
                OrderItems = null,
            };

            OrderContext.Orders.Add(orderEntity);

            // patch data
            order.ID = orderEntity.ID;
            order.StatusKey = EventConstants.EVENT_STATUS_KEY_STORAGE;

            var eventEntityA = new Event()
            {
                ID = GenerateEventID(),
                OrderID = orderEntity.ID,
                Type = EventConstants.EVENT_TYPE_CREATE_ORDER,
                StatusKey = EventConstants.EVENT_STATUS_KEY_STORAGE,
                StatusValue = EventStatusEnum.UNHANDLE,
                CreatedTime = DateTime.Now,
                EntityJson = JsonHelper.SerializeObject(order)
            };

            order.StatusKey = EventConstants.EVENT_STATUS_KEY_DELIVERY;

            var eventEntityB = new Event()
            {
                ID = GenerateEventID(),
                OrderID = orderEntity.ID,
                Type = EventConstants.EVENT_TYPE_CREATE_ORDER,
                StatusKey = EventConstants.EVENT_STATUS_KEY_DELIVERY,
                StatusValue = EventStatusEnum.UNHANDLE,
                CreatedTime = DateTime.Now,
                EntityJson = JsonHelper.SerializeObject(order)
            };

            OrderContext.Events.Add(eventEntityA);
            OrderContext.Events.Add(eventEntityB);

            int count = await OrderContext.SaveChangesAsync();

            return count == 3;
        }

        private string GenerateOrderID()
        {
            // TODO: Some business logic to generate Order ID
            return Guid.NewGuid().ToString();
        }

        private string GenerateEventID()
        {
            // TODO: Some business logic to generate Order ID
            return Guid.NewGuid().ToString();
        }
    }

　　這裡主要是通過EF Core向訂單資料庫中的Orders表和Events表新增資料。可以看到，這裡向Events表中添加了兩個記錄，分別通過StatusKey進行區分。這裡的StatusKey其實是一個冗餘欄位，只是為了後面在不同的服務之間區分是否是自己需要處理的訊息。

三、StorageService與DeliveryService的實現

3.1 StorageService的實現

　　（1）通過NuGet安裝MassTransit、MassTransit.RabbitMQ、MassTransit.Extensions.DependencyInjection

　　（2）在StartUp類中注入MassTransit的IBusControl例項，加入了熔斷、重試與限流，具體看註釋

    public class Startup
    {
        ......

        // This method gets called by the runtime. Use this method to add services to the container.
        public void ConfigureServices(IServiceCollection services)
        {
            ......

            // EFCore
            services.AddDbContextPool<OrderDbContext>(
                options => options.UseSqlServer(Configuration["DB:OrderDB"]));

            // Repository
            services.AddScoped<IStorageRepository, StorageRepsitory>();
            services.AddScoped<StoreageOrderEventHandler>();

            // MassTransit
            services.AddMassTransit(c =>
            {
                c.AddConsumer<StoreageOrderEventHandler>();
            });
        }

        public static IBusControl BusControl { get; private set; }

        // This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
        public void Configure(IApplicationBuilder app, IServiceProvider serviceProvider, IHostingEnvironment env, IApplicationLifetime lifetime)
        {
            ......

            // Register Bus
            BusControl = Bus.Factory.CreateUsingRabbitMq(cfg =>
            {
                var host = cfg.Host(new Uri(Configuration["MQ:Host"]), hst =>
                {
                    hst.Username(Configuration["MQ:UserName"]);
                    hst.Password(Configuration["MQ:Password"]);
                });

                cfg.ReceiveEndpoint(host, Configuration["MQ:Queues:Storage"], e =>
                {
                    // Retry
                    e.UseRetry(ret =>
                    {
                        ret.Interval(3, TimeSpan.FromSeconds(10)); // 每隔10s重試一次，最多重試3次
                    });

                    // RateLimit
                    e.UseRateLimit(1000, TimeSpan.FromSeconds(100)); // 100s內限1000次訪問請求

                    // CircuitBreaker
                    e.UseCircuitBreaker(cb =>
                    {
                        cb.TrackingPeriod = TimeSpan.FromMinutes(1); // 跟蹤週期：1min
                        cb.TripThreshold = 15; // 失敗比例達到15%後才會開啟熔斷器
                        cb.ActiveThreshold = 5; // 至少發生5次請求後才會開啟熔斷器
                        cb.ResetInterval = TimeSpan.FromMinutes(5); // 熔斷時間間隔：5mins
                    });

                    e.LoadFrom(serviceProvider);
                });
            });

            // Register Start & Stop for Bus
            lifetime.ApplicationStarted.Register(BusControl.Start);
            lifetime.ApplicationStarted.Register(BusControl.Stop);
        }
    }

　　這裡需要注意的就是，在注入時我們指定了接收訊息所要進行處理的類物件：StorageOrderEventHandler。

　　（3）StorageOrderEventHandler

    public class StoreageOrderEventHandler : IConsumer<IOrder>
    {
        public IStorageRepository StorageRepository { get; }
        public IBusControl EventBus { get; }

        public StoreageOrderEventHandler(IStorageRepository StorageRepository)
        {
            this.StorageRepository = StorageRepository;
            this.EventBus = Startup.BusControl;
        }

        public async Task Consume(ConsumeContext<IOrder> context)
        {
            var order = context.Message;
            if (order.StatusKey != EventConstants.EVENT_STATUS_KEY_STORAGE)
            {
                // 如果不是StorageService要處理的Event則忽略該訊息
                return;
            }

            var result = StorageRepository.CreateStorage(order).GetAwaiter().GetResult();
            if (result)
            {
                IOrderEventEntity orderEventEntity = new OrderEventEntity
                {
                    OrderID = order.ID,
                    EventType = EventConstants.EVENT_TYPE_CREATE_ORDER,
                    StatusKey = EventConstants.EVENT_STATUS_KEY_STORAGE,
                    StatusValue = EventStatusEnum.HANDLED
                };

                await EventBus.Publish(orderEventEntity);
            }
        }

　　這裡首先判斷接收到的訊息是否是需要自己處理的，不是則忽略，是則呼叫StorageRepository進行庫存記錄的更新及其他業務邏輯（與DB有關的）操作，處理完畢後向事件匯流排傳送一個訊息，這裡主要是告知哪個OrderID的哪個事件型別（EventType）的哪個具體服務（StatusKey）要更新到哪個狀態（StatusValue）。

　　（4）StorageRepository：假裝自己是一個真實的倉儲

    public class StorageRepsitory : IStorageRepository
    {
        public OrderDbContext  OrderContext{ get; }

        public StorageRepsitory(OrderDbContext OrderContext)
        {
            this.OrderContext = OrderContext;
        }

        public async Task<bool> CreateStorage(IOrder order)
        {
            var eventList = await OrderContext.Events.Where(e => e.OrderID == order.ID 
                && e.Type == EventConstants.EVENT_TYPE_CREATE_ORDER
                && e.StatusKey == EventConstants.EVENT_STATUS_KEY_STORAGE).ToListAsync();

            if (eventList == null)
            {
                return false;
            }

            foreach (var eventItem in eventList)
            {
                try
                {
                    // TODO : Add record to Storage DB
                    Console.WriteLine($"Add one record to Storage DB : { JsonHelper.SerializeObject(order) }");
                }
                catch (Exception ex)
                {
                    // TODO : Exception log
                    Console.WriteLine($"Exception: {ex.Message}");
                    return false;
                }
            }

            return true;
        }
    }

　　這裡假設會做一些業務處理，比如向庫存資料庫新增一條記錄等。由於時間和精力，這裡我只向控制檯輸出一條訊息已進行驗證。

3.2 DeliveryService的實現

　　與StorageService高度類似，篇幅關係，不再贅述，請自行瀏覽示例原始碼。這裡之展示一下DeliveryOrderEventHandler類：

    public class DeliveryOrderEventHandler : IConsumer<IOrder>
    {
        public IDeliveryRepository DeliveryRepository { get; }
        public IBusControl EventBus { get; }

        public DeliveryOrderEventHandler(IDeliveryRepository StorageRepository)
        {
            this.DeliveryRepository = StorageRepository;
            this.EventBus = Startup.BusControl;
        }

        public async Task Consume(ConsumeContext<IOrder> context)
        {
            var order = context.Message;
            if (order.StatusKey != EventConstants.EVENT_STATUS_KEY_DELIVERY)
            {
                // 如果不是DeliveryService要處理的Event則忽略該訊息
                return;
            }

            var result = DeliveryRepository.CreateDelivery(order).GetAwaiter().GetResult();
            if (result)
            {
                IOrderEventEntity orderEventEntity = new OrderEventEntity
                {
                    OrderID = order.ID,
                    EventType = EventConstants.EVENT_TYPE_CREATE_ORDER,
                    StatusKey = EventConstants.EVENT_STATUS_KEY_DELIVERY,
                    StatusValue = EventStatusEnum.HANDLED
                };

                await EventBus.Publish(orderEventEntity);
            }
        }
    }

　　可以看出，這裡的業務邏輯和StorageService一樣，只是StatusKey不同而已。

四、EventService的實現

4.1 專案結構

　　在EventService中，除了安裝MassTransit相關的package之外，還要安裝Quartz.Net的package。

4.2 StartUp類

　　（1）ConfigureService方法

    public void ConfigureServices(IServiceCollection services)
    {
        //services.AddMvc();

        // EFCore
        services.AddDbContextPool<OrderDbContext>(
            options => options.UseSqlServer(Configuration["DB:OrderDB"]));

        // Dapper-ConnString
        services.AddSingleton(Configuration["DB:OrderDB"]);

        // Repository
        services.AddSingleton<IEventRepository<IOrderEventEntity>, OrderEventDapperRepository>();

        // Quartz
        services.UseQuartz(typeof(OrderEventJob));

        // MassTransit
        services.AddMassTransit(c =>
        {
            c.AddConsumer<OrderEventHandler>();
        });
    }

　　這裡指定了Quartz.Net要處理的定時任務（通過自定義的擴充套件方法）以及MassTransit要處理的事件處理程式。有關Quartz.Net的內容不在本篇的重點，下面看看OrderEventHandler類，它主要就是根據收到的訊息去更新某個類別某個項的事件狀態記錄。

    public class OrderEventHandler : IConsumer<IOrderEventEntity>
    {
        public IEventRepository<IOrderEventEntity> EventRepository { get; }

        public OrderEventHandler(IEventRepository<IOrderEventEntity> EventRepository)
        {
            this.EventRepository = EventRepository;
        }

        public async Task Consume(ConsumeContext<IOrderEventEntity> context)
        {
            var eventResult = await EventRepository.UpdateEventStatus(EventConstants.EVENT_TYPE_CREATE_ORDER, context.Message);
        }
    }

　　在EventRepository具體類中，這裡主要是通過Dapper去操作資料庫：

    public class OrderEventDapperRepository : IEventRepository<IOrderEventEntity>
    {
        private string connStr;

        public OrderEventDapperRepository(string connStr)
        {
            this.connStr = connStr;
        }

        public async Task<IEnumerable<IOrderEventEntity>> GetEvents(string eventType)
        {
            using (var conn = new SqlConnection(connStr))
            {
                string sqlCommand = @"SELECT [EventID],
                                                        [EventType],
                                                        [OrderID],
                                                        [CreatedTime],
                                                        [StatusKey],
                                                        [StatusValue],
                                                        [EntityJson]
                                                        FROM [dbo].[Events]
                                                        WHERE EventType = @EventType 
                                                            AND StatusValue = @StatusValue";

                var result = await conn.QueryAsync<OrderEventEntity>(sqlCommand, param: new
                {
                    EventType = EventConstants.EVENT_TYPE_CREATE_ORDER,
                    StatusValue = EventStatusEnum.UNHANDLE
                });

                return result;
            }
        }

        public async Task<bool> UpdateEventStatus(string eventType, IOrderEventEntity orderEvent)
        {
            using (var conn = new SqlConnection(connStr))
            {
                string sqlCommand = @"UPDATE [dbo].[Events]
                                                            SET StatusValue = @StatusValue
                                                            WHERE OrderID = @OrderID
                                                                AND EventType = @EventType
                                                                AND StatusKey = @StatusKey";

                var result = await conn.ExecuteAsync(sqlCommand, param: new
                {
                    OrderID = orderEvent.OrderID,
                    EventType = EventConstants.EVENT_TYPE_CREATE_ORDER,
                    StatusKey = orderEvent.StatusKey,
                    StatusValue = EventStatusEnum.HANDLED
                });

                return result > 0;
            }
        }
    }

　　*Problem: 在coding過程中，本來想用EF的，結果發現DbContext預設注入的週期是Scoped，而我們的定時Job又是Singleton的，無法正常使用，所以就改用了Dapper。在這個類中，未完成的方法是進行事務回滾逆操作的一系列方法。

　　（2）Configure方法

    public static IBusControl BusControl { get; private set; }

    // This method gets called by the runtime. Use this method to configure the HTTP request pipeline.
    public void Configure(IApplicationBuilder app, IServiceProvider serviceProvider, IHostingEnvironment env, IApplicationLifetime lifetime, IScheduler scheduler)
    {
        //app.UseMvc();

        // Register EventBus
        BusControl = Bus.Factory.CreateUsingRabbitMq(cfg =>
        {
            var host = cfg.Host(new Uri(Configuration["MQ:Host"]), hst =>
            {
                hst.Username(Configuration["MQ:UserName"]);
                hst.Password(Configuration["MQ:Password"]);
            });

            // Retry
            cfg.UseRetry(ret =>
            {
                ret.Interval(3, TimeSpan.FromSeconds(10)); // 每隔10s重試一次，最多重試3次
            });

            // RateLimit
            cfg.UseRateLimit(1000, TimeSpan.FromSeconds(100)); // 100s內限1000次訪問請求

            // CircuitBreaker
            cfg.UseCircuitBreaker(cb =>
            {
                cb.TrackingPeriod = TimeSpan.FromMinutes(1); // 跟蹤週期：1min
                cb.TripThreshold = 15; // 失敗比例達到15%後才會開啟熔斷器
                cb.ActiveThreshold = 5; // 至少發生5次請求後才會開啟熔斷器
                cb.ResetInterval = TimeSpan.FromMinutes(5); // 熔斷時間間隔：5mins
            });

            cfg.ReceiveEndpoint(host, Configuration["MQ:Queues:Order"], e => 
            {
                e.LoadFrom(serviceProvider);
            });
        });

        // Register Start & Stop for Bus
        lifetime.ApplicationStarted.Register(BusControl.Start);
        lifetime.ApplicationStarted.Register(BusControl.Stop);

        // Scheduler Job
        QuartzServiceUtil.StartJob<OrderEventJob>(scheduler, TimeSpan.FromSeconds(30));
    }

　　由於無法在ConfigureService方法中正常的注入MassTransit，所以採用了一個折中的方式：靜態變數，還好我們只需要一個單例的BusControl即可。這裡我們在啟動時，開啟了一個定時任務，這個定時任務的邏輯如下，它每隔30s執行一次。

    public class OrderEventJob: IJob
    {
        public IEventRepository<IOrderEventEntity> OrderEventRepository { get; }
        public IBusControl EventBus { get; }
        private int MaxHours;

        public OrderEventJob(IEventRepository<IOrderEventEntity> OrderEventRepository, IConfiguration configuration)
        {
            this.OrderEventRepository = OrderEventRepository;
            this.EventBus = Startup.BusControl;
            this.MaxHours = Convert.ToInt32(configuration["Service:MaxHours"]);
        }

        public async Task Execute(IJobExecutionContext context)
        {
            var events = OrderEventRepository.GetEvents(EventConstants.EVENT_TYPE_CREATE_ORDER).GetAwaiter().GetResult();

            if (events == null)
            {
                await Console.Out.WriteLineAsync($"[Tip] There's no pending to process Order events.");
                return;
            }

            foreach (var eventItem in events)
            {
                if (GetDifferenceInHours(eventItem.CreatedTime) >= MaxHours)
                {
                    // TODO: 
                    // 1.Rollback previous transaction by send rollback message
                    // 2.Send Email/Messages to administrator
                    // ......

                    break;
                }

                IOrder order = JsonHelper.DeserializeJsonToObject<Order>(eventItem.EntityJson);
                await EventBus.Publish(order);
            }
        }

        private double GetDifferenceInHours(DateTime createdTime)
        {
            DateTime currentTime = DateTime.Now;
            TimeSpan ts = currentTime.Subtract(createdTime);
            double differenceInDays = ts.TotalHours;

            return differenceInDays;
        }
    }

　　這裡的MaxHours（最大容忍小時）在配置檔案中設定的是1，即1小時。在每個定時任務中，系統會去首先check未處理的事件訊息的建立時間和現在系統時間的間隔時間是否超過了1小時，超過了則會進行一系列的回滾逆操作和傳送郵件/簡訊等操作告知人工干預，這一部分由於時間和精力未實現，有興趣的可以自己去實現。如果沒超過，則會將事件狀態表記錄行中的EntityJson（這裡主要是訂單表的序列化後的JSON字串）反序列化並作為訊息進行傳送給事件匯流排從而通知訂閱者。

五、快速測試

5.1 向OrderService傳送一個訂單請求

　　首先，清空測試的訂單資料庫表，此時無一條記錄。

　　然後，通過PostMan工具向OrderService傳送一條訂單請求（前提是你得同時把這四個服務一起啟動起來）：

5.2 Check此時的訂單資料庫

　　此時已經有了對應的資料，可以看到DeliveryService和StorageService的兩個事件狀態記錄行的狀態是1（代表待處理）

5.3 一定時間後的訂單資料庫

　　經過兩個服務的處理之後，狀態變為了2（代表已處理），再看看兩個服務的控制檯資訊，分別在處理事件訊息時輸出了一行記錄：

　　在標準情況下，當所有相關的事件訊息狀態都變成已處理時，這時資料就達到了最終一致性。當然，還有一些重試的補償和事務的回滾逆操作，沒有做演示，有興趣可以自行研究。

六、小結

　　本篇主要基於一個小案例（訂單業務處理場景），首先介紹了其業務場景與基本業務流程，然後通過介紹相關的每個服務的程式碼實現，最後通過一個快速的測試演示了資料如何達到最終一致性。當然，這個小案例並不完整，沒有對重試的補償機制以及失敗後的回滾機制進行演示和測試，不過有興趣的朋友可以自行改程式碼實現。最後，再次強調實現資料的最終一致性有很多方案，這裡只是舉一種我從桂素偉老師那裡所學到的比較簡單易於學習的實現方式而已。對MassTransit感興趣想應用於生產環境的朋友，可以去了解了解saga，建議看看這篇文章：《MassTransit&Sage分散式服務開發PPT分享》

示例程式碼

　　Click Here => 點我下載

參考資料

作者：周旭龍

本文版權歸作者和部落格園共有，歡迎轉載，但未經作者同意必須保留此段宣告，且在文章頁面明顯位置給出原文連結。