jiangyihz

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

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

gRPC(rpc远程调用)

gRPC-微服务间通信实践

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

EventBus(基于消息队列的集成事件)

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

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

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

搭建框架介绍

新建项目

  1. 新建解决方案 DotNetCore.Cap.Demo
  2. 新建项目 DotNetCore.Cap.Demo.Publisher 消息发布端
  3. 新建项目 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/guest
                x.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/guest
                x.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}");
        }
    }
}

总结

  • DotNetCore.Cap 是一种异步消息的通信,可以作为微服务间通信的一种方式
  • DotNetCore.Cap 为微服务架构提供了分布式事务的解决方案,保障了数据的最终一致性
  • 开发中,应根据实际业务需求和场景,选择合适可靠的微服务间通信方案

参考

https://github.com/dotnetcore/CAP/blob/master/README.md
https://book.douban.com/subject/33425123/

相关文章: