一、CAP簡介

　　CAP 是一個在分散式系統中（SOA，MicroService）實現事件匯流排及最終一致性（分散式事務）的一個開源的 C# 庫，她具有輕量級，高效能，易使用等特點。我們可以輕鬆的在基於 .NET Core 技術的分散式系統中引入CAP，包括但限於 ASP.NET Core 和 ASP.NET Core on .NET Framework。

　　CAP 的應用場景主要有以下兩個：

• 分散式事務中的最終一致性（非同步確保）的方案
• 具有高可用性的 EventBus

　　CAP 同時支援使用 RabbitMQ 或 Kafka 進行底層之間的訊息傳送，我們不需要具備 RabbitMQ 或者 Kafka 的使用經驗，仍然可以輕鬆的將CAP整合到專案中。

　　CAP 目前支援使用 Sql Server，MySql，PostgreSql 資料庫的專案；

　　CAP 同時支援使用 EntityFrameworkCore 和 Dapper 的專案，可以根據需要選擇不同的配置方式；

　　CAP的作者為園友savorboard（楊曉東），成都地區的.NET社群領導者，棒棒噠！

二、案例結構

　　此次試驗仍然和上一篇基於MassTransit的案例一樣（其實是我懶得再改，直接拿來複用），共有四個MicroService應用程式，當用戶下訂單時會通過CAP作為事件匯流排釋出訊息，作為訂閱者的庫存和配送服務會接收到訊息並消費訊息。此次試驗會採用RabbitMQ作為訊息佇列，採用MSSQL作為關係型資料庫（同時CAP也是支援MSSQL的）。

　　準備工作：為所有服務通過NuGet安裝CAP及其相關包

PM> Install-Package DotNetCore.CAP

　下面是RabbitMQ的支援包

PM> Install-Package DotNetCore.CAP.RabbitMQ

　下面是MSSQL的支援包

PM> Install-Package DotNetCore.CAP.SqlServer

三、具體實現

3.1 OrderService

　　（1）啟動配置：這裡主要需要給CAP指定資料庫（它會在這個資料庫中建立本地訊息表Published和Received）以及使用到的訊息佇列（這裡是RabbitMQ）

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

        // Repository
        services.AddScoped<IOrderRepository, OrderRepository>();

        // EF DbContext
        services.AddDbContext<OrderDbContext>();

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

        // CAP
        services.AddCap(x =>
        {
            x.UseEntityFramework<OrderDbContext>(); // EF

            x.UseSqlServer(Configuration["DB:OrderDB"]); // SQL Server

            x.UseRabbitMQ(cfg =>
            {
                cfg.HostName = Configuration["MQ:Host"];
                cfg.VirtualHost = Configuration["MQ:VirtualHost"];
                cfg.Port = Convert.ToInt32(Configuration["MQ:Port"]);
                cfg.UserName = Configuration["MQ:UserName"];
                cfg.Password = Configuration["MQ:Password"]; 
            }); // RabbitMQ

            // Below settings is just for demo
            x.FailedRetryCount = 2;
            x.FailedRetryInterval = 5;
        });

        ......
    }

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

        app.UseMvc();

        // CAP
        app.UseCap();

        ......
    }

　　（2）Controller：這裡會呼叫Repository去實現業務邏輯和傳送訊息

    [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.CreateOrderByDapper(orderDTO).GetAwaiter().GetResult();

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

　　（3）Repository：這裡實現了兩種方式：EF和Dapper（基於ADO.NET），其中EF方式中不需要傳transaction（當CAP檢測到 Publish 是在EF事務區域內的時候，將使用當前的事務上下文進行訊息的儲存），而基於ADO.NET方式中需要傳transaction（由於不能獲取到事務上下文，所以需要使用者手動的傳遞事務上下文到CAP中）。

    public class OrderRepository : IOrderRepository
    {
        public OrderDbContext DbContext { get; }
        public ICapPublisher CapPublisher { get; }
        public string ConnStr { get; } // For Dapper use

        public OrderRepository(OrderDbContext DbContext, ICapPublisher CapPublisher, string ConnStr)
        {
            this.DbContext = DbContext;
            this.CapPublisher = CapPublisher;
            this.ConnStr = ConnStr;
        }

        public async Task<bool> CreateOrderByEF(IOrder order)
        {
            using (var trans = DbContext.Database.BeginTransaction())
            {
                var orderEntity = new Order()
                {
                    ID = GenerateOrderID(),
                    OrderUserID = order.OrderUserID,
                    OrderTime = order.OrderTime,
                    OrderItems = null,
                    ProductID = order.ProductID // For demo use
                };

                DbContext.Orders.Add(orderEntity);
                await DbContext.SaveChangesAsync();

                // When using EF, no need to pass transaction
                var orderMessage = new OrderMessage()
                {
                    ID = orderEntity.ID,
                    OrderUserID = orderEntity.OrderUserID,
                    OrderTime = orderEntity.OrderTime,
                    OrderItems = null,
                    ProductID = orderEntity.ProductID // For demo use
                };
                
                await CapPublisher.PublishAsync(EventConstants.EVENT_NAME_CREATE_ORDER, orderMessage);

                trans.Commit();
            }

            return true;
        }

        public async Task<bool> CreateOrderByDapper(IOrder order)
        {
            using (var conn = new SqlConnection(ConnStr))
            {
                conn.Open();
                using (var trans = conn.BeginTransaction())
                {
                    // business code here
                    string sqlCommand = @"INSERT INTO [dbo].[Orders](OrderID, OrderTime, OrderUserID, ProductID)
                                                                VALUES(@OrderID, @OrderTime, @OrderUserID, @ProductID)";

                    order.ID = GenerateOrderID();
                    await conn.ExecuteAsync(sqlCommand, param: new
                    {
                        OrderID = order.ID,
                        OrderTime = DateTime.Now,
                        OrderUserID = order.OrderUserID,
                        ProductID = order.ProductID
                    }, transaction: trans);

                    // For Dapper/ADO.NET, need to pass transaction
                    var orderMessage = new OrderMessage()
                    {
                        ID = order.ID,
                        OrderUserID = order.OrderUserID,
                        OrderTime = order.OrderTime,
                        OrderItems = null,
                        MessageTime = DateTime.Now,
                        ProductID = order.ProductID // For demo use
                    };

                    await CapPublisher.PublishAsync(EventConstants.EVENT_NAME_CREATE_ORDER, orderMessage, trans);

                    trans.Commit();
                }
            }

            return true;
        }

        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();
        }
    }

　　這裡摘抄一段CAP wiki中關於事務的一段介紹：

　　事務在 CAP 具有重要作用，它是保證訊息可靠性的一個基石。 在傳送一條訊息到訊息佇列的過程中，如果不使用事務，我們是沒有辦法保證我們的業務程式碼在執行成功後訊息已經成功的傳送到了訊息佇列，或者是訊息成功的傳送到了訊息佇列，但是業務程式碼確執行失敗。

　　這裡的失敗原因可能是多種多樣的，比如連線異常，網路故障等等。

　　只有業務程式碼和CAP的Publish程式碼必須在同一個事務中，才能夠保證業務程式碼和訊息程式碼同時成功或者失敗。

換句話說，CAP會確保我們這段邏輯中業務程式碼和訊息程式碼都成功了，才會真正讓事務commit。

3.2 StorageService

　　（1）啟動配置：這裡主要是指定Subscriber

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

        // EF DbContext
        services.AddDbContext<StorageDbContext>();

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

        // Subscriber
        services.AddTransient<IOrderSubscriberService, OrderSubscriberService>();

        // CAP
        services.AddCap(x =>
        {
            x.UseEntityFramework<StorageDbContext>(); // EF

            x.UseSqlServer(Configuration["DB:StorageDB"]); // SQL Server

            x.UseRabbitMQ(cfg =>
            {
                cfg.HostName = Configuration["MQ:Host"];
                cfg.VirtualHost = Configuration["MQ:VirtualHost"];
                cfg.Port = Convert.ToInt32(Configuration["MQ:Port"]);
                cfg.UserName = Configuration["MQ:UserName"];
                cfg.Password = Configuration["MQ:Password"];
            }); // RabbitMQ

            // Below settings is just for demo
            x.FailedRetryCount = 2;
            x.FailedRetryInterval = 5;
        });

        ......
    }

    // 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)
    {
        ......

        app.UseMvc();

        // CAP
        app.UseCap();

        ......
    }

　　（2）實現Subscriber

　　首先定義一個介面，建議放到公共類庫中

    public interface IOrderSubscriberService
    {
        Task ConsumeOrderMessage(OrderMessage message);
    }

　　然後實現這個介面，記得讓其實現ICapSubscribe介面，然後我們就可以使用 CapSubscribeAttribute 來訂閱 CAP 釋出出來的訊息。

    public class OrderSubscriberService : IOrderSubscriberService, ICapSubscribe
    {
        private readonly string _connStr;
        
        public OrderSubscriberService(string connStr)
        {
            _connStr = connStr;
        }

        [CapSubscribe(EventConstants.EVENT_NAME_CREATE_ORDER)]
        public async Task ConsumeOrderMessage(OrderMessage message)
        {
            await Console.Out.WriteLineAsync($"[StorageService] Received message : {JsonHelper.SerializeObject(message)}");
            await UpdateStorageNumberAsync(message);
        }

        private async Task<bool> UpdateStorageNumberAsync(OrderMessage order)
        {
            //throw new Exception("test"); // just for demo use
            using (var conn = new SqlConnection(_connStr))
            {
                string sqlCommand = @"UPDATE [dbo].[Storages] SET StorageNumber = StorageNumber - 1
                                                                WHERE StorageID = @ProductID";

                int count = await conn.ExecuteAsync(sqlCommand, param: new
                {
                    ProductID = order.ProductID
                });

                return count > 0;
            }
        }
    }

　　*.CAP約定訊息端在方法實現的過程中需要實現冪等性，所謂冪等性就是指使用者對於同一操作發起的一次請求或者多次請求的結果是一致的，不會因為多次點選而產生了副作用。這裡我沒有考慮，實際中需要首先進行驗證，避免二次更新。

3.3 DeliveryService

　　（1）啟動配置：與StorageService高度類似，只是使用的不是同一個資料庫

　　（2）實現Subscriber

    public class OrderSubscriberService : IOrderSubscriberService, ICapSubscribe
    {
        private readonly string _connStr;

        public OrderSubscriberService(string connStr)
        {
            _connStr = connStr;
        }

        [CapSubscribe(EventConstants.EVENT_NAME_CREATE_ORDER)]
        public async Task ConsumeOrderMessage(OrderMessage message)
        {
            await Console.Out.WriteLineAsync($"[DeliveryService] Received message : {JsonHelper.SerializeObject(message)}");
            await AddDeliveryRecordAsync(message);
        }

        private async Task<bool> AddDeliveryRecordAsync(OrderMessage order)
        {
            //throw new Exception("test"); // just for demo use
            using (var conn = new SqlConnection(_connStr))
            {
                string sqlCommand = @"INSERT INTO [dbo].[Deliveries] (DeliveryID, OrderID, ProductID, OrderUserID, CreatedTime)
                                                            VALUES (@DeliveryID, @OrderID, @ProductID, @OrderUserID, @CreatedTime)";

                int count = await conn.ExecuteAsync(sqlCommand, param: new
                {
                    DeliveryID = Guid.NewGuid().ToString(),
                    OrderID = order.ID,
                    OrderUserID = order.OrderUserID,
                    ProductID = order.ProductID,
                    CreatedTime = DateTime.Now
                });

                return count > 0;
            }
        }
    }

3.4 快速測試

　　（1）啟動3個微服務，Check 資料庫表狀態

　　首先會看到在各個資料庫中均建立了本地訊息表，這兩個表的含義如下：

　　Cap.Published：這個表主要是用來儲存 CAP 傳送到MQ(Message Queue)的客戶端訊息，也就是說你使用 ICapPublisher 介面 Publish 的訊息內容。

　　Cap.Received：這個表主要是用來儲存 CAP 接收到 MQ(Message Queue) 的客戶端訂閱的訊息，也就是使用 CapSubscribe[] 訂閱的那些訊息。

　　然後看看各個表的資料，目前只有庫存表有資料，因為我們要做的只是更新。

　　（2）通過Postman發一個Post請求

　　（3）Check控制檯輸出的日誌資訊

　　（4）Check資料庫中的業務表和訊息表資料：可以看到傳送者和接收者都執行成功了，如果其中任何一個參與者發生了異常或者連線不上，CAP會有預設的重試機制（預設是50次最大重試次數，每次重試間隔60s），當失敗總次數達到預設失敗總次數後，就不會進行重試了，我們可以在 Dashboard 中檢視訊息失敗的原因，然後進行人工重試處理。

　　另外，由於CAP會在資料庫中建立訊息表，因此難免會考慮到其效能。CAP提供了一個數據清理的機制，預設情況下會每隔一個小時將訊息表的資料進行清理刪除，避免資料量過多導致效能的降低。清理規則為 ExpiresAt （欄位名）不為空並且小於當前時間的資料。

四、小結

　　本篇首先簡單介紹了一下CAP這個開源專案，然後基於上一篇中的下訂單的小案例來進行了基於CAP的改造，並通過一個例項的執行來看到了結果。當然，這個例項並不完美，很多點都沒有考慮（比如訊息端消費時的冪等性）和失敗重試的場景實踐等等等等。由於時間和精力的關係，目前只使用到這兒，以後有機會能夠應用上會研究下CAP的原始碼，最後感謝楊曉東為.NET社群帶來了一個優秀的開源專案！

示例程式碼

　　Click Here => 點我點我

參考資料

作者：周旭龍

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