CAP-微服务间通信实践

微服务间通信常见的两种方式

由于微服务架构慢慢被更多人使用后，迎面而来的问题是如何做好微服务间通信的方案。我们先分析下目前最常用的两种服务间通信方案。

gRPC（rpc远程调用）

gRPC-微服务间通信实践

场景：A服务主动发起请求到B服务，同步方式
范围：只在微服务间通信应用

EventBus（基于消息队列的集成事件）

技术：NotNetCore.Cap + Rabbitmq + Database
场景：A服务要在B服务做某件事情后响应，异步方式
实现：B服务在完成某件事情后发布消息，A服务订阅此消息
范围：只在微服务间通信应用

通过对比，两种方式完全不一样。rpc是类似于http请求的及时响应机制，但是比http更轻量、快捷，它更像以前的微软的WCF，可以自动生成客户端代码，充分体现了面向实体对象的远程调用的思想；Eventbus是异步的消息机制，基于cap的思想，不关心下游订阅方服务是否消费成功，保障了主服务业务的流畅性，同时也是一款分布式事务的实现方案，可以保障分布式架构中的数据的最终一致性。

我们今天主要介绍CAP在微服务中的实践案例。

搭建框架介绍

新建项目

新建解决方案 DotNetCore.Cap.Demo
新建项目 DotNetCore.Cap.Demo.Publisher 消息发布端
新建项目 DotNetCore.Cap.Demo.Subscriber 消息订阅端

主要sdk

项目框架 netcoreapp 3.1
消息队列选用RabbitMQ
数据库存储选用PostgreSql

根据实际情况选择合适的消息队列和数据库存储

CAP 支持 Kafka、RabbitMQ、AzureServiceBus 消息队列：

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

CAP 提供了 Sql Server, MySql, PostgreSQL，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

本次demo使用的nuget包如下所示

$ dotnet list package
项目“DotNetCore.Cap.Demo.Publisher”具有以下包引用[netcoreapp3.1]: 顶级包                                                          已请求      已解决   > DotNetCore.CAP                                             3.1.1    3.1.1> DotNetCore.CAP.PostgreSql                                  3.1.1    3.1.1> DotNetCore.CAP.RabbitMQ                                    3.1.1    3.1.1> Microsoft.VisualStudio.Azure.Containers.Tools.Targets      1.10.9   1.10.9> Npgsql.EntityFrameworkCore.PostgreSQL                      3.1.4    3.1.4> Npgsql.EntityFrameworkCore.PostgreSQL.Design               1.1.0    1.1.0项目“DotNetCore.Cap.Demo.Subscriber”具有以下包引用[netcoreapp3.1]:顶级包                                                          已请求      已解决> DotNetCore.CAP                                             3.1.1    3.1.1> DotNetCore.CAP.PostgreSql                                  3.1.1    3.1.1> DotNetCore.CAP.RabbitMQ                                    3.1.1    3.1.1> Microsoft.VisualStudio.Azure.Containers.Tools.Targets      1.10.9   1.10.9> Npgsql.EntityFrameworkCore.PostgreSQL                      3.1.4    3.1.4> Npgsql.EntityFrameworkCore.PostgreSQL.Design               1.1.0    1.1.0

DotNetCore.Cap.Demo.Publisher 消息发布端

修改Startup文件

注入dotnetcore.cap组件及数据库上下文

    services.AddDbContext<PgDbContext>(p => p.UseNpgsql("数据库连接字符串"));services.AddCap(x =>{//rabbitmq在docker运行时，需要映射两个端口：5672和15672//5672供程序集访问//15672供web访问//默认用户名和密码为：guest/guestx.UseRabbitMQ(p =>{p.HostName = "localhost";p.Port = 5672;p.UserName = "guest";p.Password = "guest";});x.UseEntityFramework<PgDbContext>();//x.FailedRetryCount = 1;//x.UseDashboard();});

新建Controller作为Publisher

构造函数注入DotNetCore.CAP.ICapPublisher

提供了同步和异步的消息发布方法

发布字符串消息

await _capPublisher.PublishAsync("消息名称", "消息内容");

        [HttpGet("string")]public async Task<IActionResult> PublishString(){await _capPublisher.PublishAsync("sample.rabbitmq.demo.string", "this is text!");return Ok();}

发布对象

await _capPublisher.PublishAsync("消息名称", "消息对象");

        [HttpGet("dynamic")]public async Task<IActionResult> PublishDynamic(){await _capPublisher.PublishAsync("sample.rabbitmq.demo.dynamic", new{Name = "xiao gou",Age = 18});return Ok();}

分布式事务场景

当需要在数据库操作后，发布消息出去，DotNetCore.Cap也提供了分布式事务的解决方案。它扩展的transcation能保证只有在数据库操作和消息发送都完成后，才提交Commit

        [HttpGet("transcation")]public async Task<IActionResult> PublishWithTranscation(){using (var trans = _pgDbContext.Database.BeginTransaction(_capPublisher)){var apiConfig = new ApiConfig{ApiName = "111122",ApiDesc = "223",ReturnType = "1",ReturnExpect = "1",IsAsync = true,OperCode = "999",OperTime = DateTime.Now};await _pgDbContext.ApiConfig.AddAsync(apiConfig);await _pgDbContext.SaveChangesAsync();_capPublisher.Publish("sample.rabbitmq.demo.transcation", apiConfig);trans.Commit();return Ok();}}

DotNetCore.Cap.Demo.Subscriber 消息订阅端

修改Startup文件

注入dotnetcore.cap组件及数据库上下文

    services.AddDbContext<PgDbContext>(p => p.UseNpgsql("数据库连接字符串"));services.AddCap(x =>{//rabbitmq在docker运行时，需要映射两个端口：5672和15672//5672供程序集访问//15672供web访问//默认用户名和密码为：guest/guestx.UseRabbitMQ(p =>{p.HostName = "localhost";p.Port = 5672;p.UserName = "guest";p.Password = "guest";});x.UseEntityFramework<PgDbContext>();//x.FailedRetryCount = 1;//x.UseDashboard();});

新建Controller作为Subscriber

订阅字符串消息

[NonAction] 标签：Indicates that a controller method is not an action method.
[CapSubscribe] 标签：标志此方法为订阅方法，并以消息名称匹配发布端的消息事件

        [NonAction][CapSubscribe("sample.rabbitmq.demo.string")]public void SubscriberString(string text){Console.WriteLine($"【SubscriberString】Subscriber invoked, Info: {text}");}

订阅对象消息

方法入参person即为消息体

        [NonAction][CapSubscribe("sample.rabbitmq.demo.dynamic")]public void SubscriberDynamic(dynamic person){Console.WriteLine($"【SubscriberDynamic】Subscriber invoked, Info: {person.Name} {person.Age}");}

新建Service作为Subscriber

除了Controller可以作为消息订阅端外，也可以用继承自DotNetCore.CAP.ICapSubscribe接口的Service作为订阅端
Controller作为订阅者时，不用继承ICapSubscribe；Service作为订阅者时，必须继承ICapSubscribe
Controller和Service同时订阅了一个消息时，只触发了Service的消费；若要多个消费，需要在不同的Group下

using DotNetCore.CAP;
using System;namespace DotNetCore.Cap.Demo.Subscriber.Services
{/// <summary>/// 消费订阅服务/// </summary>public class SubscriberService : ICapSubscribe{[CapSubscribe("sample.rabbitmq.demo.string")]public void SubscriberString(string text){Console.WriteLine($"【SubscriberString】Subscriber invoked, Info: {text}");}[CapSubscribe("sample.rabbitmq.demo.dynamic")]public void SubscriberDynamic(dynamic person){Console.WriteLine($"【SubscriberDynamic】Subscriber invoked, Info: {person.Name} {person.Age}");}[CapSubscribe("sample.rabbitmq.demo.object")]public void SubscriberObject(Person person){Console.WriteLine($"【SubscriberObject】Subscriber invoked, Info: {person.Name} {person.Age}");}[CapSubscribe("sample.rabbitmq.demo.trans")]public void SubscriberTrans(ApiConfig apiConfig){Console.WriteLine($"【SubscriberTrans】Subscriber invoked, Info: {apiConfig.Id}");}}
}