数据库代码
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/
背景
在构建分布式应用的过程中也会遇到分布式事务的问题，那么 CAP 就是在这样的背景下诞生的。
CAP 介绍
Github：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在系统中的一个不完全示意图：
图中实线部分代表用户代码，虚线部分代表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>(); }
结束了，怎么样，是不是很简单？
