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

本文涉及的产品
云数据库 RDS SQL Server,独享型 2核4GB
云数据库 MongoDB,通用型 2核4GB
简介: .NET Core 事件总线,分布式事务解决方案:CAP

数据库代码


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在系统中的一个不完全示意图:

1673442616030.jpg

图中实线部分代表用户代码,虚线部分代表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>();
}


结束了,怎么样,是不是很简单?

相关实践学习
MongoDB数据库入门
MongoDB数据库入门实验。
快速掌握 MongoDB 数据库
本课程主要讲解MongoDB数据库的基本知识,包括MongoDB数据库的安装、配置、服务的启动、数据的CRUD操作函数使用、MongoDB索引的使用(唯一索引、地理索引、过期索引、全文索引等)、MapReduce操作实现、用户管理、Java对MongoDB的操作支持(基于2.x驱动与3.x驱动的完全讲解)。 通过学习此课程,读者将具备MongoDB数据库的开发能力,并且能够使用MongoDB进行项目开发。 &nbsp; 相关的阿里云产品:云数据库 MongoDB版 云数据库MongoDB版支持ReplicaSet和Sharding两种部署架构,具备安全审计,时间点备份等多项企业能力。在互联网、物联网、游戏、金融等领域被广泛采用。 云数据库MongoDB版(ApsaraDB for MongoDB)完全兼容MongoDB协议,基于飞天分布式系统和高可靠存储引擎,提供多节点高可用架构、弹性扩容、容灾、备份回滚、性能优化等解决方案。 产品详情: https://www.aliyun.com/product/mongodb
相关文章
|
5天前
|
Cloud Native API C#
C#的现代化:.NET Core引领的技术革命
【6月更文挑战第9天】`.NET Core引领C#现代化,实现跨平台革命,提升性能并支持云原生应用。异步编程模型优化体验,统一API简化开发流程。C#应用场景扩展,开发效率提高,技术创新加速,预示其未来在技术领域将持续发挥关键作用。`
23 10
|
8天前
|
开发框架 .NET Linux
【.NET Developer】已发布好的.NET Core项目文件如何打包为Docker镜像文件
该文介绍了如何不使用VS2019手动创建ASP.NET Core Blazor项目的Dockerfile并构建Docker镜像。首先,创建名为Dockerfile的文件,并复制提供的Dockerfile内容,该文件指定了基础镜像和工作目录。然后,通过CMD在项目目录下运行`docker build -t 自定义镜像名 .`来生成镜像。最后,使用`docker run`命令启动容器并验证项目运行。此外,文章还提到了将镜像推送到Azure Container Registry (ACR)的步骤。
|
8天前
|
Linux C# C++
【.NET Developer】创建ASP.NET Core Blazor项目并打包为Linux镜像发布到Azure应用服务
本文介绍了如何使用VS2019和.NET框架创建一个Blazor应用,并将其部署到Azure应用服务。首先,Blazor是一个使用C#而非JavaScript构建交互式Web UI的框架,支持共享服务器和客户端应用逻辑,以及与Docker和Azure集成。任务包括创建Blazor项目,配置Dockerfile为Linux容器,本地测试,发布到Azure Container Registry (ACR),然后在Azure App Service for Container上部署。在部署过程中,需确保Docker设置正确,开启ACR的Admin访问权限,并监控镜像拉取和容器启动日志。
|
9天前
|
XML 开发框架 人工智能
C#/.NET/.NET Core拾遗补漏合集(24年5月更新)
C#/.NET/.NET Core拾遗补漏合集(24年5月更新)
|
9天前
|
开发框架 .NET API
ASP.NET Core Web中使用AutoMapper进行对象映射
ASP.NET Core Web中使用AutoMapper进行对象映射
|
9天前
|
前端开发 Java C#
GitHub突破5k Star!这件事情我坚持了3年,努力打造C#/.NET/.NET Core全面的学习、工作、面试指南知识库
GitHub突破5k Star!这件事情我坚持了3年,努力打造C#/.NET/.NET Core全面的学习、工作、面试指南知识库
|
9天前
|
存储 开发框架 缓存
【.NET Core】你真的了解HttpRuntime类吗
【.NET Core】你真的了解HttpRuntime类吗
10 0
|
9天前
|
前端开发 C#
【.NET Core】你认识Attribute之CallerMemberName、CallerFilePath、CallerLineNumber三兄弟
【.NET Core】你认识Attribute之CallerMemberName、CallerFilePath、CallerLineNumber三兄弟
12 0
|
10天前
|
安全 C#
【.NET Core】深入理解IO - 读取器和编写器
【.NET Core】深入理解IO - 读取器和编写器
23 5
|
10天前
|
存储 网络协议 程序员
【.NET Core】.NET中的流(Stream)
【.NET Core】.NET中的流(Stream)
24 7