2.解决服务之间的通讯

标签：hat   订阅   ext   支持   jpg   process   stream   erro   start   

作者

微信:tangy8080
电子邮箱:914661180@qq.com
更新时间：2019-06-28 14:25:40 星期五

欢迎您订阅和分享我的订阅号,订阅号内会不定期分享一些我自己学习过程中的编写的文章
如您在阅读过程中发现文章错误,可添加我的微信 tangy8080 进行反馈.感谢您的支持。

文章主题

介绍多个服务之间进行通讯

前置条件

[无]

正文

服务之间应该尽量较少调用.以减少耦合度,如果彼此调用链过于频繁.可能会引起整个调用链的异常.
https://docs.microsoft.com/en-us/dotnet/standard/microservices-architecture/architect-microservice-container-applications/communication-in-microservice-architecture
The microservice community promotes the philosophy of "smart endpoints and dumb pipes" This slogan encourages a design that‘s as decoupled as possible between microservices, and as cohesive as possible within a single microservice. As explained earlier, each microservice owns its own data and its own domain logic. But the microservices composing an end-to-end application are usually simply choreographed by using REST communications rather than complex protocols such as WS-* and flexible event-driven communications instead of centralized business-process-orchestrators.

但服务之间的调用有时候会变得"随其自然",以上一章利用consul实现k8s服务自动发现
结尾的两个服务为例说明

• configcenter 作为整个集群甚至对外（这个时候,我们牵涉到一个概念：集中授权中兴 如果不授权,那集群外部访问配置.岂不是变得非常危险了吗？该问题我们会在后面提及）的配置中心,（它以consul的KV为存储）提供统一的配置
• terminal 作为终端服务,它提供对集群外部终端的集中管控，例如通知终端更新,向终端发送通知等等

terminal需要数据库的配置信息(ip,port等)以存储各个终端的数据,这些配置存放在configcenter中.那么terminal服务如何从configcenter中取出配置呢？ 这时候就涉及到了服务间的通讯了

服务之间如何通讯

凡是涉及到通讯的,一般都会涉及到两个概念

• 通讯协议 TCP,UDP,HTTP等？
• 数据锲约 JSON,XML又或者类似根据服务端协议自动生成客户端代码

通讯协议

在微服务之间通讯目前比较流行的有两种 TCP和HTTP

• 在TCP通讯的基础上有一些框架可供您选择：Thrift,GRPC等

• HTTP

  基于Consul和HTTP协议手撸一个微服务之间的库

  这里,我选择自己手撸一个简单的微服务之间的调用的库,它的工作模式如下
  

在ConsulCaller中,我们使用了Consul库进行服务发现.当发现了服务实例时,程序会根据随机算法选取实例.然后返回给调用方.

源码地址

git://gitblit.honeysuckle.site/public/Honeysuckle.git

实现方式讲解

1. 扩展IServiceCollection，增加扩展函数AddConsulCaller.该函数通过调用方传入Consul的配置信息来进行服务配置.
   IConsulConfiguration 中定义了Consul的配置参数,包括consul的地址,端口,Token 由于简单这里不过多说明.

 public static void AddConsulCaller(this IServiceCollection services, Action<ConsulCallerOptions> optionsAction)
        {
            var consulConfiguration=new ConsulConfiguration();
            services.AddSingleton<IConsulConfiguration>(consulConfiguration);
            services.AddSingleton<IServiceDiscover, ServiceDiscover>();

            services.AddSingleton<IConsulClientFactory, ConsulClientFactory.ConsulClientFactory>();
            services.AddTransient<IConsulCaller, ConsulCaller>();

            var consulCallerOptions = new ConsulCallerOptions(consulConfiguration);
            optionsAction.Invoke(consulCallerOptions);
        }

1. 我定义了一个IConsulClientFactory接口,它包含一个Get函数.用于返回IConsulClient对象.该对象支持发现consul中的服务实例

 public IConsulClient Get(IConsulConfiguration config)
        {
            return new ConsulClient(c =>
            {
                c.Address = new Uri($"http://{config.Host}:{config.Port}");

                if (!string.IsNullOrEmpty(config?.Token))
                {
                    c.Token = config.Token;
                }
            });
        }

3.另外一个重要的接口是IServiceDiscover,它包含一个GetServices的函数,该函数根据服务名称返回服务实例列表

        private readonly IConsulClient _consul;
        private const string VersionPrefix = "version-";

        public ServiceDiscover(IConsulClientFactory consulClientFactory, IConsulConfiguration consulConfiguration)
        {
            _consul = consulClientFactory.Get(consulConfiguration);
        }

        public  List<Service> GetServices(string key)
        {
            var queryResult =  _consul.Health.Service(key, string.Empty, true).Result;
            
            var services = new List<Service>();
            foreach (var serviceEntry in queryResult.Response)
            {
                if (IsValid(serviceEntry))
                {
                    var nodes =  _consul.Catalog.Nodes().Result;
                    if (nodes.Response == null)
                    {
                        services.Add(BuildService(serviceEntry, null));
                    }
                    else
                    {
                        var serviceNode = nodes.Response.FirstOrDefault(n => n.Address == serviceEntry.Service.Address);
                        services.Add(BuildService(serviceEntry, serviceNode));
                    }
                }
                else
                {
                    Console.WriteLine($"Unable to use service Address: {serviceEntry.Service.Address} and Port: {serviceEntry.Service.Port} as it is invalid. Address must contain host only e.g. localhost and port must be greater than 0");
                }
            }
            return services;
        }


        private static Service BuildService(ServiceEntry serviceEntry, Node serviceNode)
        {
            return new Service(
                serviceEntry.Service.Service,
                new ServiceHostAndPort(serviceNode == null ? serviceEntry.Service.Address : serviceNode.Name, serviceEntry.Service.Port),
                serviceEntry.Service.ID,
                GetVersionFromStrings(serviceEntry.Service.Tags),
                serviceEntry.Service.Tags ?? Enumerable.Empty<string>());
        }


        private static bool IsValid(ServiceEntry serviceEntry)
        {
            if (string.IsNullOrEmpty(serviceEntry.Service.Address) || serviceEntry.Service.Address.Contains("http://") || serviceEntry.Service.Address.Contains("https://") || serviceEntry.Service.Port <= 0)
            {
                return false;
            }

            return true;
        }


        private static string GetVersionFromStrings(IEnumerable<string> strings)
        {
            return strings
                ?.FirstOrDefault(x => x.StartsWith(VersionPrefix, StringComparison.Ordinal))
                .TrimStart(VersionPrefix);
        }

4.最后我们提供了一个IConsulCaller接口,它提供一个Call接口,输入服务名称和一个回调函数.

如何使用？

1.在asp.net core引用中添加了ConsulCaller服务

 //使用ConsulCaller服务
            services.AddConsulCaller(options =>
            {
                options.ConsulConfiguration.Host = Configuration["Consul:Host"];
                options.ConsulConfiguration.Port = Convert.ToInt32(Configuration["Consul:Port"]);
            });

2.调用集群中的其他服务

const string serviceName = "configcenter";
            _consulCaller.Call(serviceName, (endpoint, httpclient) =>
            {
                try
                {
                    var uri = $"http://{endpoint.DownstreamHost}:{endpoint.DownstreamPort}/Consul/Get";
                   //根据返回的服务实例,实现自己的调用逻辑
                }
                catch (Exception ex)
                {
                    var errorMsg = $"{nameof(TerminalServerDbContext)}.{nameof(OnConfiguring)} _consulCaller.Call Error ";
                    Console.WriteLine(errorMsg + ex.Message);
                    Logger.Error(errorMsg, ex);
                }
                finally
                {
                    httpclient.Dispose();
                }
            });

• IConsulCaller 对象在第一步已经被注入到IServiceCollection容器中

引用链接

[无]

2.解决服务之间的通讯

标签：hat   订阅   ext   支持   jpg   process   stream   erro   start   

原文地址：https://www.cnblogs.com/gytangyao/p/11407232.html