.NET领域最硬核的gRPC 核心能力一把梭

有态度的马甲

发布于 2024-01-19 15:20:03

3580

发布于 2024-01-19 15:20:03

文章被收录于专栏：精益码农

前言，本文定位为.NET方向 grpc核心能力一把梭，全篇是姿势性和结论性的展示，方便中高级程序员快速上手.NET Grpc。

有关grpc更深层次的前世今生、底层原理、困惑点释疑请听下回分解，欢迎菜鸟老鸟们提出宝贵意见。

grpc宏观目标：高性能rpc框架
grpc框架实现宏观目标的底层3协议
- http2通信协议，基础能力
- proto buffer：打解包协议==> 二进制
- proto buffer：服务协议，IDL
通过脚手架项目分析grpc简单一元通信
grpc打乒乓球实践双向流式通信
grpc除了基于3大协议之外，扩展点体现能力，扩展点在哪？
- 调用管道: 池化tcp、 tcp探活
- 负载均衡
- 元数据 metadata
- 拦截器

一. 宏观目标

gRPC是高性能的RPC框架，有效地用于服务通信(不管是数据中心内部还是跨数据中心)。

科普rpc：程序可以像调用本地函数和本地对象一样，达成调用远程服务的效果，rpc屏蔽了底层的通信细节和打解包细节。跟许多rpc协议一样， grpc也是基于IDL（interface define lauguage）来定义服务协议。

grpc是基于http/2协议的高性能的rpc框架。

二. grpc实现跨语言的rpc调用目标

基于三协议：

底层传输协议：基于http2 （多路复用、双向流式通信）
打解包协议：基于proto Buffer 打包成二进制格式传输
接口协议：基于契约优先的开发方式(契约以proto buffer格式定义），可以使用protoc 编译器生产各种语言的本地代理类，磨平了微服务平台中各语言的编程隔阂。

下图演示了C++ grpc服务, 被跨语言客户端调用， rpc服务提供方会在调用方产生服务代理stub，客户端就像调用本地服务一样，产生远程调用的效果。

在大规模微服务中，C++grpc服务也可能作为调用的客户端，于是这个服务上可能也存在其他服务提供方的服务代理stub，上图没有体现。

三. 通过脚手架项目分析gRPC简单一元通信

我们将从使用gRPC服务模板创建一个新的dotnet项目。

VS gRPC服务模板默认使用TLS 来创建gRRPC服务，实际上不管是HTTP1.1 还是HTTP2，都不强制要求使用TLS 如果服务一开始同时支持HTTP1.1+ HTTP2 但是没有TLS，那么协商的结果将是 HTTP1.1+ TLS，这样的话gRPC调用将会失败。

3.1 The RPC Service Definition

protocol buffers既用作服务的接口定义语言(记录服务定义和负载消息)，又用作底层消息交换格式。这个说法语上面的3大底层协议2,3 呼应。

① 使用protocol buffers在.proto文件中定义服务接口。在其中，定义可远程调用的方法的入参和返回值类型。服务器实现此接口并运行gRPC服务器以处理客户端调用。

② 定义服务后，使用PB编译器protoc从.proto文件生成指定语言的数据访问/传输类stub，该文件包含服务接口中消息和方法的实现。

syntax = "proto3";             //   `syntax`指示使用的protocol buffers的版本

option csharp_namespace = "GrpcAuthor";    // `csharp_namespace`指示未来生成的存根文件所在的`命名空间`, 这是对应C#语言， java语言应填 java_package

package greet;

service Greeter {
     rpc SayHello (HelloRequest) returns (HelloReply);      // 一元rpc调用
}

message HelloRequest {
  string name = 1;
}

message HelloReply {
  string message = 1;
}

注释一看就懂。

接下来使用protoc编译器和C＃插件来对proto文件生成服务器或客户端代码。

① 由客户端和服务共享的强类型对象，表示消息的服务操作和数据元素，这个是pb序列化协议的强类型对象。
②一个强类型基类，具有远程 gRPC 服务可以继承和扩展的所需网络管道：Greeter.GreeterBase
③一个客户端存根，其中包含调用远程 gRPC 服务所需的管道：Greeter.GreeterClient 。运行时，每条消息都序列化为标准 Protobuf 二进制表示形式，在客户端和远程服务之间交换。

3.2 实现服务定义

脚手架项目使用Grpc.AspNetCore NuGet包：所需的类由构建过程自动生成, 你只需要在项目.csproj文件中添加配置节：

<ItemGroup>
  <Protobuf Include="Protos\greet.proto" GrpcServices="Server" />
</ItemGroup>

以下是继承②强基类而实现的grpc服务

public class GreeterService : Greeter.GreeterBase
{
    private readonly ILogger<GreeterService> _logger;
    public GreeterService(ILogger<GreeterService> logger)
    {
        _logger = logger;
    }

    public override Task<HelloReply> SayHello(HelloRequest request, ServerCallContext context)
    {
        return Task.FromResult(new HelloReply
        {
            Message = "Hello " + request.Name
        });
    }
}

最后在原http服务进程上注册Grpc端点


public void ConfigureServices(IServiceCollection services)
{
       services.AddGrpc();
}

 public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
 {
      app.UseEndpoints(endpoints =>
    {
          endpoints.MapGrpcService<GreeterService>();
          endpoints.MapGet("/", async context =>
          {
              await context.Response.WriteAsync("----http调用-------");
          });
    });
 }

以上在localhost：5000端口同时支持了grpc调用和http调用。

--- 启动服务---...

3.3. 创建gRPC .NET客户端

Visual Studio创建一个名为GrpcAuthorClient的新控制台项目。

安装如下nuget包: Install-Package Grpc.Net.Client // 包含.NET Core客户端; Install-Package Google.Protobuf // 包含protobuf消息API; Install-Package Grpc.Tools // 对Protobuf文件进行编译

① 拷贝服务端项目中的..proto文件

② 将选项csharp_namespace值修改为GrpcAuthorClient。

③ 更新.csproj文件的配置节

<ItemGroup>
 <Protobuf Include="Protos\author.proto" GrpcServices="Client" />
</ItemGroup>

④ Client主文件:

static void Main(string[] args)
{
       var serverAddress = "https://localhost:5001";

      using var channel = GrpcChannel.ForAddress(serverAddress);
      var client = new Greeter.GreeterClient(channel);
      var reply = client.SayHello(new HelloRequest { Name = "宋小宝！" });

      Console.WriteLine(reply.Message.ToString());

      Console.WriteLine("Press any key to exit...");
      Console.ReadKey();
}

使用服务器地址创建GrpcChannel，然后使用GrpcChannel对象实例化GreeterClient；然后使用SayHello同步方法; 服务器响应时，打印结果。

脚手架例子就可以入门，下面聊一聊另外的核心功能

四. gRPC打乒乓球：双向流式通信[1]

除了上面的一元rpc调用(Unary RPC), 还有

Client streaming RPC：客户端流式RPC，客户端以流形式（一系列消息）向服务器发起请求，客户端将等待服务器读取消息并返回响应，gRPC服务端能保证了收到的单个RPC调用中的消息顺序。
Server streaming RPC ：服务器流式RPC，客户端向服务器发送请求，并获取服务器流（一系列消息）。客户端从返回的流（一系列消息）中读取，直到没有更多消息为止， gRPC客户端能保证收到的单个RPC调用中的消息顺序。
Bidirectional streaming RPC：双向流式RPC，双方都使用读写流发送一系列消息。这两个流是独立运行的，因此客户端和服务器可以按照自己喜欢的顺序进行读写：例如，服务器可以在写响应之前等待接收所有客户端消息，或者可以先读取一条消息再写入一条消息，或读写的其他组合，同样每个流中的消息顺序都会保留。

针对脚手架项目，稍作修改成打乒乓球，考察gRpc双向流式通信、Cancellation机制、grpc元数据三个特性

双向流式可以不管对方是否回复，首先已方是可以持续发送的，己方可以等收到所有信息再回复，也可以收到一次回复一次，也可以自定义收到几次回复一次。本次演示土乒乓球对攻，故

对攻用到双向流，收到一次，回复一次。
强制设置30s的回合对攻必须分出胜负，使用Cancellation控制回合结束
对攻双方是白云和黑土，使用元数据约束

① 添加服务定义接口

rpc PingPongHello(stream Serve) returns (stream Catch);

② 服务器实现

   public override async Task PingPongHello(IAsyncStreamReader<Serve> requestStream,IServerStreamWriter<Catch> responseStream, ServerCallContext context)
        {
            try
            {
                if ("baiyun" != context.RequestHeaders.Get("node").Value)    // 接收请求头 header
                {
                  context.Status = new Status(StatusCode.PermissionDenied,"黑土只和白云打乒乓球");  // 设置响应状态码
                  await  Task.CompletedTask;
                  return;  
                }
                await context.WriteResponseHeadersAsync(new Metadata{   // 发送响应头header
                    { "node", "heitu" }
                });
                long  round = 0L;
                
                context.CancellationToken.Register(() => { 
                    Console.WriteLine($"乒乓球回合制结束, {context.Peer} : {round}");
                    context.ResponseTrailers.Add("round", round.ToString());  // 统计一个回合里双方有多少次对攻
                    context.Status = new Status(StatusCode.OK,"");  // 设置响应状态码
                });
                while (!context.CancellationToken.IsCancellationRequested)
                {
                    var asyncRequests = requestStream.ReadAllAsync(context.CancellationToken);
                    await foreach (var req in asyncRequests)
                    {
                        var send = RandomDirect();    // ToDo 想要实现一个 随时间衰减的概率算法，模拟对攻最后终止。
                        await responseStream.WriteAsync(new Catch
                        {
                            Direct = send,
                            Id = req.Id
                        });
                        Console.WriteLine($" {context.Peer} : 第{req.Id}次服务端收到 {req.Direct}, 第{req.Id + 1}次发送 {send}");
                        round++;
                    }
                }
                 
            }
            catch(Exception ex)
            {
                Console.WriteLine($"{ex.Message}");
            }
            finally
            {
                Console.WriteLine($"乒乓球回合制结束");
            }
        }

        static Direction RandomDirect()
        {
            var ran = new Random();
            var ix = ran.Next(0, 4);
            var dir= new[] { "Front", "Back","Left", "Right",  }[ix];
            System.Enum.TryParse<Direction>(dir, out var direct);
            return direct;
        }

③ 客户端

var serverAddress = "http://localhost:5000";
           var handler = new SocketsHttpHandler
           {
               PooledConnectionIdleTimeout = Timeout.InfiniteTimeSpan,
               KeepAlivePingDelay = TimeSpan.FromSeconds(60),
               KeepAlivePingTimeout = TimeSpan.FromSeconds(30),    // tcp心跳探活
               EnableMultipleHttp2Connections = true               // 启用并发tcp连接
           };
           using var channel = GrpcChannel.ForAddress(serverAddress, new GrpcChannelOptions { 
               Credentials = ChannelCredentials.Insecure,
               MaxReceiveMessageSize = 1024 * 1024 * 10,
               MaxSendMessageSize = 1024 * 1024 * 10,  
               HttpHandler = handler 
           });
           var client = new PingPong.PingPongClient(channel);
           AsyncDuplexStreamingCall<Serve,Catch>   duplexCall = null;
            Console.WriteLine($"{DateTime.Now:yyyy-MM-dd HH:mm:ss.fff}, 白云先发球");
           using (var cancellationTokenSource = new CancellationTokenSource(30*1000))
           {
               try
               {
                   duplexCall = client.PingPongHello(new Metadata
                   {
                       { "node", "baiyun" }
                   }, null, cancellationTokenSource.Token );
                   
                   var headers = await duplexCall.ResponseHeadersAsync;
                   if ("heitu" != headers.Get("node").Value)    // 接收响应头
                   {
                      throw new RpcException(new Status(StatusCode.PermissionDenied, "白云只和黑土打乒乓球"));
                   }
                   var direct = RandomDirect();
                   await duplexCall.RequestStream.WriteAsync(new Serve { Id= 1, Direct = direct }) ;
                   await foreach (var resp in duplexCall.ResponseStream.ReadAllAsync())
                   {
                       Console.WriteLine($"第{resp.Id}次攻防，客户端发送{direct},客户端收到 {resp.Direct}");
                        direct = RandomDirect();
                       await duplexCall.RequestStream.WriteAsync(new Serve { Id= resp.Id+1 ,Direct =  direct });
                   }
                   Console.WriteLine($"{DateTime.Now:yyyy-MM-dd HH:mm:ss.fff}打乒乓球结束");
                   if (duplexCall != null)
                   {
                       var tr = duplexCall.GetTrailers();   // 接受响应尾
                       var round  = tr.Get("round").Value.ToString();
                       Console.Write($" 进行了 {round} 次攻防）");
                   }
               }
               catch (RpcException ex)
               {
                   var trailers = ex.Trailers;
                   _ = trailers.GetValue("round");
               }
               catch(Exception ex) 
               {
                   Console.WriteLine($"{DateTime.Now:yyyy-MM-dd HH:mm:ss.fff}打乒乓球（30s回合制）结束 未分出胜负，{ex.Message}");
               }
           }

https://github.com/zaozaoniao/GrpcAuthor

五：grpc扩展点

grpc：是基于http2 多路复用能力，在单tcp连接上发起高效rpc调用的框架。根据grpc调用的生命周期：可在如下阶段扩展能力

服务可寻址
附加在grpc header/trailer的元数据
连接/调用凭证
连接/调用重试机制----> 拦截器
调用状态码：https://grpc.github.io/grpc/core/md_doc_statuscodes.html

下面挑选几个核心的扩展点着重聊一聊。

5.1 负载均衡

哪些调用能做负载均衡？

只有[gRPC调用]能实现对多服务提供方节点的负载平衡, 一旦建立了gRPC流式调用，所有通过该流式调用发送的消息都将发送到一个端点。

grpc负载均衡的时机？

grpc诞生的初衷是点对点通信，现在常用于内网服务之间的通信，在微服务背景下，服务调用也有负载均衡的问题，也正因为连接建立之后是“点对点通信”，所以不方便基于L4做负载均衡。

根据grpc的调用姿势， grpc的负载均衡可在如下环节：

① 客户端负载均衡：对于每次rpc call，选择一个服务终结点，直接调用无延迟，但客户端需要周期性寻址。

② L7做服务端负载均衡：L7负载层能理解HTTP/2，并且能在一个HTTP/2连接上跨多个服务提供方节点将[多路复用的gRPC调用]分发给上游服务节点。使用代理比客户端负载平衡更简单，但会给gRPC调用增加额外的延迟。

常见的是客户端负载均衡。

https://grpc.io/blog/grpc-load-balancing/

5.2 调用通道

grpc 利用http2 使用单一tcp连接提供到指定主机端口上年的grpc调用，通道是与远程服务器的长期tcp连接的抽象。客户端对象可以重用相同的通道，与rpc调用行为相比，创建通道是一项昂贵的操作，因此应该为尽可能多的调用重复使用单个通道。

根据http2 上默认并发流的限制(100)， .NET支持在单tcp连接并发流到达上限的时候，产生新的tcp连接，故通道是一个池化的tcp并发流的概念, grpc通道具有状态，包括已连接和空闲.
像websockets这类长时间利用tcp连接的机制一样，都需要心跳保活机制, 可以快速的进行grpc调用，而不用等待tcp连接建立而延迟。
可以指定通道参数来修改gRPC的默认行为，例如打开或关闭消息压缩, 添加连接凭据。

var handler = new SocketsHttpHandler
{
    PooledConnectionIdleTimeout = Timeout.InfiniteTimeSpan,
    KeepAlivePingDelay = TimeSpan.FromSeconds(60),
    KeepAlivePingTimeout = TimeSpan.FromSeconds(30),  // tcp心跳探活
    EnableMultipleHttp2Connections = true      // 启用并发tcp连接
};

var channel = GrpcChannel.ForAddress("https://localhost:5001", new GrpcChannelOptions
{
    Credentials = ChannelCredentials.Insecure,     // 连接凭据
    HttpHandler = handler
});

https://learn.microsoft.com/en-us/aspnet/core/grpc/performance?view=aspnetcore-7.0

5.3 Metadata

元数据是以键值对列表的形式提供的有关特定RPC调用的信息（身份认证信息、访问令牌、代理信息），在grpc调用双方，一般元数据存储在header或trailer 中。

客户端发起调用时会有metadata参数可供使用：

// 上例中的 proto被编译之后产生了如下 sdk
public virtual HelloReply SayHello(HelloRequest request, Metadata headers = null, DateTime? deadline = null, CancellationToken cancellationToken = default(global::System.Threading.CancellationToken))
{
        return SayHello(request, new CallOptions(headers, deadline, cancellationToken));
}

对于身份认证元数据，有更通用的方式：builder.Services.AddGrpcClient<Greeter.GreeterClient>().AddCallCredentials((x,y) =>{ })

grpc 服务端可发送的是 header 和trailer， trailer只能在服务端响应完毕发送，至于为什么有header，还有trailer，请看再谈 gRPC 的 Trailers 设计[2]，总体而言grpc流式通信需要在调用结束给客户端传递一些之前给不了的信息。

await context.WriteResponseHeadersAsync(new Metadata{   // 发送响应头
        { "node", "B" }
 });

 context.ResponseTrailers.Add("count", cnt);  // 发送响应尾
 context.Status = Status.DefaultSuccess;  // 设置响应状态码

5.4 自定义拦截器和可能使用到的HttpClient

拦截器与 .net httpclientDelegate 、 axio的请求拦截器类似，都是在发起调用的时候，做一些过滤或者追加的行为。https://learn.microsoft.com/en-us/aspnet/core/grpc/interceptors?view=aspnetcore-8.0

builder.Services
    .AddGrpcClient<Greeter.GreeterClient>(o =>
    {
        o.Address = new Uri("https://localhost:5001");
    })
    .AddInterceptor<LoggingInterceptor>();     // 默认在客户端之间共享

// 以下是一个客户端日志拦截器，在一元异步调用时拦截
public class ClientLoggingInterceptor : Interceptor
{
    private readonly ILogger _logger;

    public ClientLoggingInterceptor(ILoggerFactory loggerFactory)
    {
        _logger = loggerFactory.CreateLogger<ClientLoggingInterceptor>();
    }

    public override AsyncUnaryCall<TResponse> AsyncUnaryCall<TRequest, TResponse>(
        TRequest request,
        ClientInterceptorContext<TRequest, TResponse> context,
        AsyncUnaryCallContinuation<TRequest, TResponse> continuation)
    {
        _logger.LogInformation("Starting call. Type/Method: {Type} / {Method}",
            context.Method.Type, context.Method.Name);     // 拦截动作： 在continuation之前做日志记录。
        return continuation(request, context);
    }
}

总结

gRPC是具有可插拔身份验证和负载平衡功能的高性能RPC框架。使用protocol buffers定义结构化数据; 针对不同语言编译出的代理sdk屏蔽底层通信和打接包细节，完成了本地实现远程调用的效果（调用方不care是远程通信）。

Additional Resources

• https://developers.google.com/protocol-buffers/docs/csharptutorial • https://www.grpc.io/docs/what-is-grpc/core-concepts/ • https://docs.microsoft.com/en-us/dotnet/architecture/grpc-for-wcf-developers/why-grpc