高校人力资源管理系统网站开发,Wordpress多重筛选插件,网站建设经验王者荣耀恺和,网站目录层级建设系列文章目录
1、.Net Core微服务入门系列#xff08;一#xff09;——项目搭建 2、.Net Core微服务入门全纪录#xff08;二#xff09;——Consul-服务注册与发现#xff08;上#xff09; 3、.Net Core微服务入门全纪录#xff08;三#xff09;——Consul-服务注…系列文章目录
1、.Net Core微服务入门系列一——项目搭建 2、.Net Core微服务入门全纪录二——Consul-服务注册与发现上 3、.Net Core微服务入门全纪录三——Consul-服务注册与发现下 4、.Net Core微服务入门全纪录四——Ocelot-API网关上 5、.Net Core微服务入门全纪录五——Ocelot-API网关下 6、.Net Core微服务入门全纪录六——EventBus-事件总线 7、.Net Core微服务入门全纪录七——IdentityServer4-授权认证 8、.Net Core微服务入门全纪录八——Docker Compose与容器网络 文章目录 系列文章目录前言一、EventBus-事件总线1.1 什么是事件总线1.2 为什么要用EventBus 二、CAP使用2.1 环境准备2.2 代码修改 三、运行测试四、总结 前言
关于 微服务 的概念解释网上有很多 个人理解微服务是一种系统架构模式它和语言无关和框架无关和工具无关和服务器环境无关。
微服务思想 是将传统的单体系统按照业务拆分成多个职责单一、且可独立运行的接口服务。至于服务如何拆分没有明确的定义。几乎任何后端语言都能做微服务开发。微服务也并不是完美无缺的微服务架构会带来更多的问题增加系统的复杂度引入更多的技术栈。
上一篇【.Net Core微服务入门全纪录五——Ocelot-API网关下】中已经完成了 Ocelot Consul 的搭建这一篇简单说一下 EventBus。
一、EventBus-事件总线
1.1 什么是事件总线
事件总线 是对观察者发布-订阅模式的一种实现。它是一种集中式事件处理机制允许不同的组件之间进行彼此通信而又不需要相互依赖达到一种 解耦 的目的。
如果没有接触过 EventBus 可能不太好理解。其实 EventBus 在客户端开发中应用非常广泛androidiosweb 前端等用于多个组件或者界面之间的相互通信。
1.2 为什么要用EventBus
就拿当前的项目举例我们有一个订单服务一个产品服务。客户端有一个下单功能当用户下单时调用订单服务的下单接口那么下单接口需要调用产品服务的减库存接口这涉及到服务与服务之间的调用。那么服务之间又怎么调用呢直接 RESTAPI或者效率更高的gRPC可能这两者各有各的使用场景但是他们都存在一个服务之间的耦合问题或者难以做到异步调用。
试想一下假设我们下单时调用订单服务订单服务需要调用产品服务产品服务又要调用物流服务物流服务再去调用xx服务 等等。。。如果每个服务处理时间需要2s不使用异步的话那这种体验可想而知。
如果使用 EventBus 的话那么订单服务只需要向 EventBus 发一个“下单事件”就可以了。产品服务会订阅“下单事件”当产品服务收到下单事件时自己去减库存就好了。这样就避免了两个服务之间直接调用的耦合性并且真正做到了异步调用。
既然涉及到多个服务之间的异步调用那么就不得不提分布式事务。分布式事务并不是微服务独有的问题而是所有的分布式系统都会存在的问题。
关于分布式事务可以查一下 “CAP原则” 和 “BASE理论” 了解更多。当今的分布式系统更多的会追求事务的最终一致性。
下面使用国人开发的优秀项目 “CAP”来演示一下 EventBus 的基本使用。之所以使用 “CAP”是因为它既能解决分布式系统的最终一致性同时又是一个 EventBus它具备 EventBus 的所有功能 作者介绍https://www.cnblogs.com/savorboard/p/cap.html
二、CAP使用
2.1 环境准备
在 Docker 中准备一下需要的环境首先是数据库数据库我使用 PostgreSQL用别的也行。CAP 支持SqlServerMySqlPostgreSqlMongoDB。
然后是MQ这里我使用 RabbitMQ Kafka 也可以。
Docker运行RabbitMQ
docker pull rabbitmq:management
docker run -d -p 15672:15672 -p 5672:5672 --name rabbitmq rabbitmq:management默认用户guest密码guest
环境准备就完成了Docker 就是这么方便。
2.2 代码修改
为了模拟以上业务需要修改大量代码下面代码如有遗漏的直接去github找。
NuGet安装
Microsoft.EntityFrameworkCore
Microsoft.EntityFrameworkCore.Tools
Npgsql.EntityFrameworkCore.PostgreSQLCAP相关
DotNetCore.CAP
DotNetCore.CAP.RabbitMQ
DotNetCore.CAP.PostgreSqlOrder.API/Controllers/OrdersController.cs 增加下单接口
[Route([controller])]
[ApiController]
public class OrdersController : ControllerBase
{private readonly ILoggerOrdersController _logger;private readonly IConfiguration _configuration;private readonly ICapPublisher _capBus;private readonly OrderContext _context;public OrdersController(ILoggerOrdersController logger, IConfiguration configuration, ICapPublisher capPublisher, OrderContext context){_logger logger;_configuration configuration;_capBus capPublisher;_context context;}[HttpGet]public IActionResult Get(){string result $【订单服务】{DateTime.Now.ToString(yyyy-MM-dd HH:mm:ss)}—— ${Request.HttpContext.Connection.LocalIpAddress}:{_configuration[ConsulSetting:ServicePort]};return Ok(result);}/// summary/// 下单 发布下单事件/// /summary/// param nameorder/param/// returns/returns[Route(Create)][HttpPost]public async TaskIActionResult CreateOrder(Models.Order order){using (var trans _context.Database.BeginTransaction(_capBus, autoCommit: true)){//业务代码order.CreateTime DateTime.Now;_context.Orders.Add(order);var r await _context.SaveChangesAsync() 0;if (r){//发布下单事件await _capBus.PublishAsync(order.services.createorder, new CreateOrderMessageDto() { Count order.Count, ProductID order.ProductID });return Ok();}return BadRequest();}}}
Order.API/MessageDto/CreateOrderMessageDto.cs
/// summary
/// 下单事件消息
/// /summary
public class CreateOrderMessageDto
{/// summary/// 产品ID/// /summarypublic int ProductID { get; set; }/// summary/// 购买数量/// /summarypublic int Count { get; set; }
}
Order.API/Models/Order.cs订单实体类
public class Order
{[Key][DatabaseGenerated(DatabaseGeneratedOption.Identity)]public int ID { get; set; }/// summary/// 下单时间/// /summary[Required]public DateTime CreateTime { get; set; }/// summary/// 产品ID/// /summary[Required]public int ProductID { get; set; }/// summary/// 购买数量/// /summary[Required]public int Count { get; set; }
}
Order.API/Models/OrderContext.cs数据库Context
public class OrderContext : DbContext
{public OrderContext(DbContextOptionsOrderContext options): base(options){}public DbSetOrder Orders { get; set; }protected override void OnModelCreating(ModelBuilder modelBuilder){}
}
Order.API/appsettings.json增加数据库连接字符串
ConnectionStrings: {OrderContext: User IDpostgres;Passwordpg123456;Hosthost.docker.internal;Port5432;DatabaseOrder;Poolingtrue;
}
Order.API/Startup.cs修改ConfigureServices方法添加Cap配置
public void ConfigureServices(IServiceCollection services)
{services.AddControllers();services.AddDbContextOrderContext(opt opt.UseNpgsql(Configuration.GetConnectionString(OrderContext)));//CAPservices.AddCap(x {x.UseEntityFrameworkOrderContext();x.UseRabbitMQ(host.docker.internal);});
} 以上是订单服务的修改。
Product.API/Controllers/ProductsController.cs增加减库存接口
[Route([controller])]
[ApiController]
public class ProductsController : ControllerBase
{private readonly ILoggerProductsController _logger;private readonly IConfiguration _configuration;private readonly ICapPublisher _capBus;private readonly ProductContext _context;public ProductsController(ILoggerProductsController logger, IConfiguration configuration, ICapPublisher capPublisher, ProductContext context){_logger logger;_configuration configuration;_capBus capPublisher;_context context;}[HttpGet]public IActionResult Get(){string result $【产品服务】{DateTime.Now.ToString(yyyy-MM-dd HH:mm:ss)}—— ${Request.HttpContext.Connection.LocalIpAddress}:{_configuration[ConsulSetting:ServicePort]};return Ok(result);}/// summary/// 减库存 订阅下单事件/// /summary/// param namemessage/param/// returns/returns[NonAction][CapSubscribe(order.services.createorder)]public async Task ReduceStock(CreateOrderMessageDto message){//业务代码var product await _context.Products.FirstOrDefaultAsync(p p.ID message.ProductID);product.Stock - message.Count;await _context.SaveChangesAsync();}}
Product.API/MessageDto/CreateOrderMessageDto.cs
/// summary
/// 下单事件消息
/// /summary
public class CreateOrderMessageDto
{/// summary/// 产品ID/// /summarypublic int ProductID { get; set; }/// summary/// 购买数量/// /summarypublic int Count { get; set; }
}
Product.API/Models/Product.cs产品实体类
public class Product
{[Key][DatabaseGenerated(DatabaseGeneratedOption.Identity)]public int ID { get; set; }/// summary/// 产品名称/// /summary[Required][Column(TypeName VARCHAR(16))]public string Name { get; set; }/// summary/// 库存/// /summary[Required]public int Stock { get; set; }
}
Product.API/Models/ProductContext.cs数据库Context
public class ProductContext : DbContext
{public ProductContext(DbContextOptionsProductContext options): base(options){}public DbSetProduct Products { get; set; }protected override void OnModelCreating(ModelBuilder modelBuilder){base.OnModelCreating(modelBuilder);//初始化种子数据modelBuilder.EntityProduct().HasData(new Product{ID 1,Name 产品1,Stock 100},new Product{ID 2,Name 产品2,Stock 100});}
}
Product.API/appsettings.json增加数据库连接字符串
ConnectionStrings: {ProductContext: User IDpostgres;Passwordpg123456;Hosthost.docker.internal;Port5432;DatabaseProduct;Poolingtrue;
}
Product.API/Startup.cs修改ConfigureServices方法添加Cap配置\
public void ConfigureServices(IServiceCollection services)
{services.AddControllers();services.AddDbContextProductContext(opt opt.UseNpgsql(Configuration.GetConnectionString(ProductContext)));//CAPservices.AddCap(x {x.UseEntityFrameworkProductContext();x.UseRabbitMQ(host.docker.internal);});
} 以上是产品服务的修改。
订单服务和产品服务的修改到此就完成了看着修改很多其实功能很简单。就是各自增加了自己的数据库表然后订单服务增加了下单接口下单接口会发出 “下单事件”。产品服务增加了减库存接口减库存接口会订阅 “下单事件”。然后客户端调用下单接口下单时产品服务会减去相应的库存功能就这么简单。
关于 EF数据库迁移 之类的基本使用就不介绍了。使用 Docker 重新构建镜像运行订单服务产品服务
docker build -t orderapi:1.1 -f ./Order.API/Dockerfile .
docker run -d -p 9060:80 --name orderservice orderapi:1.1 --ConsulSetting:ServicePort9060
docker run -d -p 9061:80 --name orderservice1 orderapi:1.1 --ConsulSetting:ServicePort9061
docker run -d -p 9062:80 --name orderservice2 orderapi:1.1 --ConsulSetting:ServicePort9062docker build -t productapi:1.1 -f ./Product.API/Dockerfile .
docker run -d -p 9050:80 --name productservice productapi:1.1 --ConsulSetting:ServicePort9050
docker run -d -p 9051:80 --name productservice1 productapi:1.1 --ConsulSetting:ServicePort9051
docker run -d -p 9052:80 --name productservice2 productapi:1.1 --ConsulSetting:ServicePort9052
最后 Ocelot.APIGateway/ocelot.json 增加一条路由配置 好了进行到这里整个环境就有点复杂了。确保我们的PostgreSQLRabbitMQConsulGateway服务实例都正常运行。
服务实例运行成功后数据库应该是这样的 产品表种子数据 cap.published 表和 cap.received 表是由 CAP自动生成的它内部是使用本地消息表MQ来实现异步确保。
三、运行测试
这次使用 Postman 作为客户端调用下单接口 9070 是之前的 Ocelot 网关端口 订单库 published 表 订单库 order 表 产品库 received 表 产品库 product 表 再试一下 OK完成。虽然功能很简单但是我们实现了服务的解耦异步调用和最终一致性。
四、总结
注意上面的例子纯粹是为了说明 EventBus 的使用实际中的下单流程绝对不会这么做的希望大家不要较真。
可能有人会说如果下单成功但是库存不足导致减库存失败了怎么办是不是要回滚订单表的数据如果产生这种想法说明还没有真正理解最终一致性的思想。
首先下单前肯定会检查一下库存数量既然允许下单那么必然是库存充足的。这里的事务是指订单保存到数据库和下单事件保存到 cap.published 表保存到 cap.published 表理论上就能够发送到MQ这两件事情要么一同成功要么一同失败。如果这个事务成功那么就可以认为这个业务流程是成功的至于产品服务的减库存是否成功那就是产品服务的事情了理论上也应该是成功的因为消息已经确保发到了MQ产品服务必然会收到消息CAP也提供了失败重试和失败回调机制。
如果非要数据回滚也是能实现的CAP 的 ICapPublisher.Publish 方法提供一个 callbackName参数当减库存时可以触发这个回调。其本质也是通过发布订阅完成这是不推荐的做法就不详细说了有兴趣自己研究一下。 另外CAP 无法保证消息不重复实际使用中需要自己考虑一下消息的重复过滤和幂等性。
