擴展gRPC支持consul服務發現和Polly策略

gRPC由於需要用工具生成代碼實現,可開發性不是很高,在擴展這方面不是很友好

最近研究了下,進行了擴展,不需要額外的工具生成,直接使用預設Grpc.Tools生成的代理類即可

相關源碼在文章底部

客戶端目標:

• 能配置consul地址和服務名稱,在調用client時能正確請求到真實的服務地址
• 在調用方法時,能使用Polly策略重試,超時,和熔斷

查看gRPC生成的代碼,可以看到Client實例化有有兩個構造方法,以測試為例

      /// <summary>Creates a new client for Greeter</summary>
      /// <param name="channel">The channel to use to make remote calls.</param>
      public GreeterClient(grpc::ChannelBase channel) : base(channel)
      {
      }
      /// <summary>Creates a new client for Greeter that uses a custom <c>CallInvoker</c>.</summary>
      /// <param name="callInvoker">The callInvoker to use to make remote calls.</param>
      public GreeterClient(grpc::CallInvoker callInvoker) : base(callInvoker)
      {
      }

1.可傳入一個ChannelBase實例化

2.可傳入一個CallInvoker實例化

Channel可實現為

Channel CreateChannel(string address)
        {
            var channelOptions = new List<ChannelOption>()
                    {
                        new ChannelOption(ChannelOptions.MaxReceiveMessageLength, int.MaxValue),
                        new ChannelOption(ChannelOptions.MaxSendMessageLength, int.MaxValue),
                    };
            var channel = new Channel(address, ChannelCredentials.Insecure, channelOptions);
            return channel;
        }

在這裡,可以從consul地址按服務名獲取真實的服務地址,生成Channel

CallInvoker為一個抽象類,若要對方法執行過程干預,則需要重寫這個方法,大致實現為

public class GRpcCallInvoker : CallInvoker
    {
        public readonly Channel Channel;
        public GRpcCallInvoker(Channel channel)
        {
            Channel = GrpcPreconditions.CheckNotNull(channel); 
        }

        public override TResponse BlockingUnaryCall<TRequest, TResponse>(Method<TRequest, TResponse> method, string host, CallOptions options, TRequest request)
        {
            return Calls.BlockingUnaryCall(CreateCall(method, host, options), request);
        }

        public override AsyncUnaryCall<TResponse> AsyncUnaryCall<TRequest, TResponse>(Method<TRequest, TResponse> method, string host, CallOptions options, TRequest request)
        {
            return Calls.AsyncUnaryCall(CreateCall(method, host, options), request);
        }

        public override AsyncServerStreamingCall<TResponse> AsyncServerStreamingCall<TRequest, TResponse>(Method<TRequest, TResponse> method, string host, CallOptions options, TRequest request)
        {
            return Calls.AsyncServerStreamingCall(CreateCall(method, host, options), request);
        }

        public override AsyncClientStreamingCall<TRequest, TResponse> AsyncClientStreamingCall<TRequest, TResponse>(Method<TRequest, TResponse> method, string host, CallOptions options)
        {
            return Calls.AsyncClientStreamingCall(CreateCall(method, host, options));
        }

        public override AsyncDuplexStreamingCall<TRequest, TResponse> AsyncDuplexStreamingCall<TRequest, TResponse>(Method<TRequest, TResponse> method, string host, CallOptions options)
        {
            return Calls.AsyncDuplexStreamingCall(CreateCall(method, host, options));
        }

        protected virtual CallInvocationDetails<TRequest, TResponse> CreateCall<TRequest, TResponse>(Method<TRequest, TResponse> method, string host, CallOptions options)
            where TRequest : class 
            where TResponse : class
        {
            return new CallInvocationDetails<TRequest, TResponse>(Channel, method, host, options);
        }
    }

這裡可以傳入上面創建的Channel,在CreateCall方法里,則可以對調用方法進行控制

完整實現為

public class GRpcCallInvoker : CallInvoker
    {
        GrpcClientOptions _options;
        IGrpcConnect _grpcConnect;
        public GRpcCallInvoker(IGrpcConnect grpcConnect)
        {
            _options = grpcConnect.GetOptions();
            _grpcConnect = grpcConnect;
        }

        public override TResponse BlockingUnaryCall<TRequest, TResponse>(Method<TRequest, TResponse> method, string host, CallOptions options, TRequest request)
        {
            return Calls.BlockingUnaryCall(CreateCall(method, host, options), request);
        }

        public override AsyncUnaryCall<TResponse> AsyncUnaryCall<TRequest, TResponse>(Method<TRequest, TResponse> method, string host, CallOptions options, TRequest request)
        {
            return Calls.AsyncUnaryCall(CreateCall(method, host, options), request);
        }

        public override AsyncServerStreamingCall<TResponse> AsyncServerStreamingCall<TRequest, TResponse>(Method<TRequest, TResponse> method, string host, CallOptions options, TRequest request)
        {
            return Calls.AsyncServerStreamingCall(CreateCall(method, host, options), request);
        }

        public override AsyncClientStreamingCall<TRequest, TResponse> AsyncClientStreamingCall<TRequest, TResponse>(Method<TRequest, TResponse> method, string host, CallOptions options)
        {
            return Calls.AsyncClientStreamingCall(CreateCall(method, host, options));
        }

        public override AsyncDuplexStreamingCall<TRequest, TResponse> AsyncDuplexStreamingCall<TRequest, TResponse>(Method<TRequest, TResponse> method, string host, CallOptions options)
        {
            return Calls.AsyncDuplexStreamingCall(CreateCall(method, host, options));
        }

        protected virtual CallInvocationDetails<TRequest, TResponse> CreateCall<TRequest, TResponse>(Method<TRequest, TResponse> method, string host, CallOptions options)
            where TRequest : class
            where TResponse : class
        {
            var methodName = $"{method.ServiceName}.{method.Name}";
            var key = methodName.Substring(methodName.IndexOf(".") + 1).ToLower();
            var a = _options.MethodPolicies.TryGetValue(key, out PollyAttribute methodPollyAttr);
            if (!a)
            {
                _options.MethodPolicies.TryGetValue("", out methodPollyAttr);
            }
            CallOptions options2;
            //重寫header
            if (options.Headers != null)
            {
                options2 = options;
            }
            else
            {
                options2 = new CallOptions(_grpcConnect.GetMetadata(), options.Deadline, options.CancellationToken);
            }

            var pollyData = PollyExtension.Invoke(methodPollyAttr, () =>
            {
                var callRes = new CallInvocationDetails<TRequest, TResponse>(_grpcConnect.GetChannel(), method, host, options2);
                return new PollyExtension.PollyData<CallInvocationDetails<TRequest, TResponse>>() { Data = callRes };
            }, $"{methodName}");
            var response = pollyData.Data;
            if (!string.IsNullOrEmpty(pollyData.Error))
            {
                throw new Exception(pollyData.Error);
            }
            return response;
            //return new CallInvocationDetails<TRequest, TResponse>(Channel.Invoke(), method, host, options2);
        }
    }

其中傳入了PollyAttribute,由PollyExtension.Invoke來完成Polly策略的實現,具體代碼可在源碼里找到

從上面代碼可以看到,CallInvoker里可以傳入了IGrpcConnect,由方法IGrpcConnect.GetChannel()獲取Channel

Client實例化

.net FrameWork實現為

     public T GetClient<T>()
        {
            var a = instanceCache.TryGetValue(typeof(T), out object instance);
            if (!a)
            {
                var grpcCallInvoker = new GRpcCallInvoker(this);
                instance = System.Activator.CreateInstance(typeof(T), grpcCallInvoker);
                instanceCache.TryAdd(typeof(T), instance);
            }
            return (T)instance;
        }

core則簡單點,直接註入實現

var client = provider.GetService<Greeter.GreeterClient>();

服務端註冊

和其它服務註冊一樣,填入正確的服務地址和名稱就行了,但是在Check里得改改,gRPC的健康檢查參數是不同的,並且在consul客戶端里沒有這個參數,得自已寫

以下代碼是我封裝過的,可查看源碼

public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
        {
            if (env.IsDevelopment())
            {
                app.UseDeveloperExceptionPage();
            }

            app.UseRouting();

            app.UseEndpoints(endpoints =>
            {
                endpoints.MapGrpcService<GreeterService>();
                endpoints.MapGrpcService<HealthCheckService>();
                endpoints.MapGet("/", async context =>
                {
                    await context.Response.WriteAsync("Communication with gRPC endpoints must be made through a gRPC client. To learn how to create a client, visit: https://go.microsoft.com/fwlink/?linkid=2086909");
                });
            });

            //註冊服務
            var consulClient = new CRL.Core.ConsulClient.Consul("http://localhost:8500");
            var info = new CRL.Core.ConsulClient.ServiceRegistrationInfo
            {
                Address = "127.0.0.1",
                Name = "grpcServer",
                ID = "grpcServer1",
                Port = 50001,
                Tags = new[] { "v1" },
                Check = new CRL.Core.ConsulClient.CheckRegistrationInfo()
                {
                    GRPC = "127.0.0.1:50001",
                    Interval = "10s",
                    GRPCUseTLS = false,
                    DeregisterCriticalServiceAfter = "90m"
                }
            };
            consulClient.DeregisterService(info.ID);
            var a = consulClient.RegisterService(info);

        }

 客戶端完整封裝代碼為

core擴展方法,設置GrpcClientOptions來配置consul地址和Polly策略,直接註入了Client類型

同時添加了統一header傳遞,使整個服務都能用一個頭髮送請求,不用再在方法後面跟參數

public static class GrpcExtensions
    {
        public static void AddGrpcExtend(this IServiceCollection services, Action<GrpcClientOptions> setupAction, params Assembly[] assemblies)
        {
            services.Configure(setupAction);
            services.AddSingleton<IGrpcConnect, GrpcConnect>();
            services.AddScoped<CallInvoker, GRpcCallInvoker>();
            foreach (var assembyle in assemblies)
            {
                var types = assembyle.GetTypes();
                foreach (var type in types)
                {
                    if(typeof(ClientBase).IsAssignableFrom(type))
                    {
                        services.AddSingleton(type);
                    }
                }
            }
        }
    }

 class Program
    {
        static IServiceProvider provider;
        static Program()
        {
            var builder = new ConfigurationBuilder();

            var configuration = builder.Build();

            var services = new ServiceCollection();
            services.AddSingleton<IConfiguration>(configuration);
            services.AddOptions();

            services.AddGrpcExtend(op =>
            {
                op.Host = "127.0.0.1";
                op.Port = 50001;
                op.UseConsulDiscover("http://localhost:8500", "grpcServer");//使用consul服務發現
                op.AddPolicy("Greeter.SayHello", new CRL.Core.Remoting.PollyAttribute() { RetryCount = 3 });//定義方法polly策略
            }, System.Reflection.Assembly.GetExecutingAssembly());

            provider = services.BuildServiceProvider();
        }


        static void Main(string[] args)
        {            
            //設置允許不安全的HTTP2支持
            AppContext.SetSwitch("System.Net.Http.SocketsHttpHandler.Http2UnencryptedSupport", true);

            var grpcConnect = provider.GetService<IGrpcConnect>();
            //認證
            //https://www.cnblogs.com/stulzq/p/11897628.html
            var token = "";
            var headers = new Metadata { { "Authorization", $"Bearer {token}" } };
            grpcConnect.SetMetadata(headers);


        label1:
            var client = provider.GetService<Greeter.GreeterClient>();
            var reply = client.SayHello(
                new HelloRequest { Name = "test" });
            Console.WriteLine("Greeter 服務返回數據: " + reply.Message);

            Console.ReadLine();
            goto label1;
        }
    }

運行服務端,結果為

可以看到服務註冊成功,狀態檢查也成功

運行客戶端

客戶端正確調用並返回了結果

項目源碼:

https://github.com/CRL2020/CRL.NetStandard/tree/master/Grpc

除了gRPC實現了服務發現和Polly策略,本框架對API代理,動態API,RPC也一起實現了

API代理測試
https://github.com/CRL2020/CRL.NetStandard/tree/master/DynamicWebApi/ApiProxyTest

動態API測試

https://github.com/CRL2020/CRL.NetStandard/tree/master/DynamicWebApi/DynamicWebApiClient

RCP測試

https://github.com/CRL2020/CRL.NetStandard/tree/master/RPC/RPCClient