.NET Core 事件总线,分布式事务解决方案：CAP

╰+哭是因爲堅強的太久メ 2023-09-28 05:07 190阅读 0赞

#### 数据库代码 ####

CREATE DATABASE OrderTest;
    GO
    USE OrderTest;
    GO
    CREATE TABLE Product(
    	Id INT PRIMARY KEY IDENTITY,
    	ProductName VARCHAR(30) NOT NULL,
    	Price INT NOT NULL,
    	Stock INT NOT NULL
    );
    CREATE TABLE [Order]
    (
    	Id INT PRIMARY KEY IDENTITY,
    	ProductId INT NOT NULL,
    	UserName VARCHAR(30) NOT NULL,
    	Quantity INT NOT NULL
    );
    GO
    INSERT INTO dbo.Product
    (
        ProductName,
        Price,
        Stock
    )
    VALUES ('衣服',30,30 ),('鞋子',20,20 ),('帽子',10,10 ),('外套',25,30 )

官网：[https://cap.dotnetcore.xyz/user-guide/zh/getting-started/quick-start/][https_cap.dotnetcore.xyz_user-guide_zh_getting-started_quick-start]

#### 背景 ####

在构建分布式应用的过程中也会遇到分布式事务的问题，那么 CAP 就是在这样的背景下诞生的。

[https://github.com/dotnetcore][https_github.com_dotnetcore]

#### CAP 介绍 ####

Github：[https://github.com/dotnetcore/CAP][https_github.com_dotnetcore_CAP]

开源协议：MIT

CAP 是一个在分布式系统中（SOA，MicroService）实现事件总线及最终一致性（分布式事务）的一个开源的 C\# 库，她具有轻量级，高性能，易使用等特点。

你可以轻松的在基于 .NET Core 技术的分布式系统中引入CAP，包括但限于 ASP.NET Core 和 ASP.NET Core on .NET Framework。

CAP 以 NuGet 包的形式提供，对项目无任何入侵，你仍然可以以你喜爱的方式来构建分布式系统。

CAP 具有 Event Bus 的所有功能，并且CAP提供了更加简化的方式来处理EventBus中的发布/订阅。

CAP 具有消息持久化的功能，也就是当你的服务进行重启或者宕机时，她可以保证消息的可靠性。

CAP 实现了分布式事务中的最终一致性，你不用再去处理这些琐碎的细节。

CAP 提供了基于 Microsoft DI 的 API 服务，她可以和你的 ASP.NET Core 系统进行无缝结合，并且能够和你的业务代码集成支持强一致性的事务处理。

CAP 是开源免费的。CAP基于MIT协议开源，你可以免费的在你的私人或者商业项目中使用，不会有人向你收取任何费用。

#### Getting Started ####

目前， CAP 同时支持使用 RabbitMQ，Kafka，Azure Service Bus 等进行底层之间的消息发送，你不需要具备这些消息队列的使用经验，仍然可以轻松的集成到项目中。

CAP 目前支持使用 Sql Server，MySql，PostgreSql，MongoDB 数据库的项目。

CAP 同时支持使用 EntityFrameworkCore 和 ADO.NET 的项目，你可以根据需要选择不同的配置方式。

下面是CAP在系统中的一个不完全示意图：

![img][]

> 图中实线部分代表用户代码，虚线部分代表CAP内部实现。

下面，我们看一下 CAP 怎么集成到项目中：

##### Step 1: #####

你可以运行下面的命令来安装CAP `NuGet` 包：

PM> Install-Package DotNetCore.CAP

根据底层消息队列，你可以选择引入不同的包：

PM> Install-Package DotNetCore.CAP.Kafka
    PM> Install-Package DotNetCore.CAP.RabbitMQ
    PM> Install-Package DotNetCore.CAP.AzureServiceBus

CAP 目前支持使用`SQL Server, PostgreSql, MySql, MongoDB` 的项目，你可以选择引入不同的包：

PM> Install-Package DotNetCore.CAP.SqlServer
    PM> Install-Package DotNetCore.CAP.MySql
    PM> Install-Package DotNetCore.CAP.PostgreSql
    PM> Install-Package DotNetCore.CAP.MongoDB     //需要 MongoDB 4.0+ 集群

##### Step 2: #####

在 `Startup.cs` 文件中，添加如下配置：

public void ConfigureServices(IServiceCollection services)
    {
        
        services.AddControllersWithViews();
    
        var sqlConnection = Configuration.GetConnectionString("OrderTestConnection");
        services.AddDbContext<OrderDbContext>(p=>p.UseSqlServer(sqlConnection));
        RabbitMQOptions rabbitOptions = new RabbitMQOptions();
        Configuration.GetSection("RabbitMQ").Bind(rabbitOptions);
        services.AddCap(p =>
                        {
        
                            p.UseEntityFramework<OrderDbContext>();
                            p.UseRabbitMQ(mq =>
                                          {
        
                                              mq.HostName = rabbitOptions.HostName;
                                              mq.VirtualHost = rabbitOptions.VirtualHost;
                                              mq.UserName = rabbitOptions.UserName;
                                              mq.Password = rabbitOptions.Password;
                                              mq.Port = rabbitOptions.Port;
                                          });
                            p.UseSqlServer(sqlConnection);
                        });
    
    }

##### Step 3: #####

`appsetting.json` 文件配置：

{
        
      "Logging": {
        
        "LogLevel": {
        
          "Default": "Information",
          "Microsoft": "Warning",
          "Microsoft.Hosting.Lifetime": "Information"
        }
      },
        "AllowedHosts": "*",
        "ConnectionStrings": {
        
            "OrderTestConnection": "server=.;uid=sa;pwd=123456;database=ordertest;"
        },
        "RabbitMQ": {
        
            "HostName": "172.16.20.157",
            "VirtualHost": "/",
            "UserName": "admin",
            "Password": "admin",
            "Port": "5672"
        } 
    }

#### 发布事件/消息 ####

在 Controller 中注入 `ICapPublisher` 然后使用 `ICapPublisher` 进行消息发布:

public class PublishController : Controller
    {
        
        private readonly ICapPublisher _capBus;
        private readonly OrderDbContext _context;
        public OrderController(ICapPublisher capBus, OrderDbContext context)
        {
        
            _capBus = capBus;
            _context = context;
        }
    
    
        public IActionResult CreateOrder()
        {
        
            return View();
        }
    
        [HttpPost]
        public IActionResult Submit(Order order)
        {
        
            using var tran = _context.Database.BeginTransaction(_capBus);
            try
            {
        
                _context.Order.Add(order);
                _context.SaveChanges();
    
                //发送消息
                _capBus.Publish("order.create",order);
    
                tran.Commit();
                return Redirect("/home/index");
            }
            catch (Exception e)
            {
        
                Console.WriteLine(e.Message);
                tran.Rollback();
                return BadRequest(e.Message);
            }
        }
    }

#### 订阅事件/消息 ####

在 `Controller` 中：  
如果是在Controller中，直接添加`[CapSubscribe("")]` 来订阅相关消息。

public class ConsumerController : Controller
    {
        private readonly OrderDbContext _context;
        public ConsumerController( OrderDbContext context)
        {
            _context = context;
        }
    
    
        [CapSubscribe(name:"order.create")] //这个要与publisher 方法中的name要一致
        public void OrderCreateConsumer(Order order)
        {
            //减库存
            var product = _context.Product.FirstOrDefault(p=>p.Id==order.ProductId);
            product.Stock -= order.Quantity;
            _context.SaveChanges();
        }
    }

在 `xxxService`中：  
如果你的方法没有位于Controller 中，那么你订阅的类需要继承 `ICapSubscribe`，然后添加`[CapSubscribe("")]`标记：

namespace xxx.Service
    {
        
        public interface ISubscriberService
        {
        
        	public void CheckReceivedMessage(DateTime time);
        }
        
        
        public class SubscriberService: ISubscriberService, ICapSubscribe
        {
        
        	[CapSubscribe("xxxx.services.show.time")]
        	public void CheckReceivedMessage(DateTime time)
        	{
        
        		
        	}
        }
    }

然后在 `Startup.cs` 中的 ConfigureServices() 中注入**你的** `ISubscriberService` 类

public void ConfigureServices(IServiceCollection services)
    {
        
        services.AddTransient<ISubscriberService,SubscriberService>();
    }

结束了，怎么样，是不是很简单？

#### 异常解决 ####

##### RabbitMQ : SubscriberNotFoundException \#63 #####

es() 中注入**你的** `ISubscriberService` 类

public void ConfigureServices(IServiceCollection services)
    {
        
        services.AddTransient<ISubscriberService,SubscriberService>();
    }

结束了，怎么样，是不是很简单？

#### 异常解决 ####

##### RabbitMQ : SubscriberNotFoundException \#63 #####

[https://github.com/dotnetcore/CAP/issues/63][https_github.com_dotnetcore_CAP_issues_63]

[https_cap.dotnetcore.xyz_user-guide_zh_getting-started_quick-start]: https://cap.dotnetcore.xyz/user-guide/zh/getting-started/quick-start/
[https_github.com_dotnetcore]: https://github.com/dotnetcore
[https_github.com_dotnetcore_CAP]: https://github.com/dotnetcore/CAP
[img]: https://img-blog.csdnimg.cn/img_convert/807d7121f81353d4ec6583aa76679fee.png
[https_github.com_dotnetcore_CAP_issues_63]: https://github.com/dotnetcore/CAP/issues/63