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 = https://www.cnblogs.com/hubro/p/PollyExtension.Invoke(methodPollyAttr, () => { var callRes = new CallInvocationDetails<TRequest, TResponse>(_grpcConnect.GetChannel(), method, host, options2); return new PollyExtension.PollyData<CallInvocationDetails<TRequest, TResponse>>() { Data =https://www.cnblogs.com/hubro/p/ 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
轉載請註明出處，本文鏈接：https://www.uj5u.com/net/52845.html
標籤：.NET Core