数据库代码
代码语言:javascript复制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 就是在这样的背景下诞生的。
https://github.com/dotnetcore
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
根据底层消息队列,你可以选择引入不同的包:
代码语言:javascript复制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