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

师傅领进门，修行在个人，跟着官方脚手架demo了grpc后，之后就需要扩展前后知识边界，下面总结grpc的前世今生和最佳实践。
https://www.cnblogs.com/JulianHuang/p/14441952.html

1. grpc是基于http/2协议的高性能的rpc框架
2. 为什么已经有http？ 还需要grpc？
3. 八股文都说grpc是基于http2的rpc框架，到底利用了http2的什么特性 ？
4. 一个grpc内存泄漏的例子

---

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

提取句式中关键信息：rpc框架、 http2、 高性能

1.落地宾语-rpc框架

远程过程调用 remote process call；
程序可以像调用本地服务器托管网函数和本地对象一样， 达成调用远程服务的效果（不用意识到是远程服务），rpc屏蔽了底层的通信细节和打解包细节。

跟许多rpc协议一样， grpc也是基于IDL（interface define lauguage）来定义服务协议。

rpc的开发模式， 必然强调契约优先， client和server端首先约定service的结构(包括一系列方法的组合、每个方法具体签名)。
对这个结构的描述，gRPC 默认是用Protocol Buffer去实现的。

syntax = "proto3";

option csharp_namespace = "GrpcAuthor";

package greet;

// The greeting service definition.
service Greeter {
  // Sends a greeting
  rpc SayHello (HelloRequest) returns (HelloReply);
}

// The request message containing the user's name.
message HelloRequest {
  string name = 1;
}

// The response message containing the greetings.
message HelloReply {
  string message = 1;
}

rpc框架两个关键通用部件：

• channel: 通信信道
• stub： 是client的抽象，中文名叫存根。

① 建立与grcp服务端的通信信道;
② 基于步骤①的信道使用服务名Greet建立grpc服务在客户端的存根， 就像服务Greet是本地服务一样;
③ 用步骤②的grpc服务存根发起grpc调用GreeterClient()，就像GreeterClient是本地方法一样。

var channel = GrpcChannel.ForAddress(
    "http:///my-example-host",
    new GrpcChannelOptions { Credentials = ChannelCredentials.Insecure });
var client = new Greet.GreeterClient(channel);

var response = await client.SayHelloAsync(new HelloRequest { Name = "world" });
 

rpc 是一种久远的通信框架， http是通信协议。

2. 目前市面流行http， 为什么还需要grpc？

其实rpc框架比 http协议更早出来。
tcp协议于70年代诞生，tcp是一种可靠的、面向连接的、基于字节流的传输层协议。

tcp粘包？
关于tcp的定语和修饰词也很关键，正因为是字节流，就如同躺在水流中的0/1串， 这些0/1串是没有任何边界的，应用层传到tcp层的数据不是以消息报为单位向对面主机发送，而是以字节流的形式流淌在tcp层， 在tcp传输层由tcp层的协议进行切割和组包， 对端接收的时候没能正确还原发送端的消息报。
tcp粘包问题并不是tcp协议的弊端，而是我们在应用层发送和接收数据需要对数据分段，而传输层是无边界的0/1串之间的矛盾。

在70到90年代之后，互联网并不发达，很多都是client、server点对点的传输，所以在那个年代rpc很活跃，client、server双方约定服务结构，以[类本地调用]的形态通信。

90年代，随着it产业的蓬勃发展，计算机走进了千家万户，90年代初期诞生了浏览器， 浏览器与上面的C/S结构最大的不同是访问的服务端对象是千万家不同的服务提供方。

再像C/S那样一对一提前沟通契约就不合适了， 故浏览器和web服务器作为一种特殊的C/S需要约定一种【固定的、能自表述的传输格式】， 于是诞生了适用于B/S端的http协议。 http协议不再有rpc双方已知的本地服务名和服务方法， 服务和服务方法被金演化成对远程主机的资源请求。

所以，我们该问的不是“既然有HTTP协议，为什么要有RPC”，而是应该问“为什么有rpc，还需要有http”。

---

从上面的前世今生可以知道， rpc是通信框架，http是通信协议，rpc可以基于tcp，udp，http/2协议来实现。

grpc在众多rpc框架中脱颖而出，取决于底层的http2基础设施。

3. grpc到底利用了http/2的什么特性？

回过头来看grpc的连两个定语 ① http/2 ② 高性能。

grpc底层传输使用http/2，http/2兼容http1.1语义,还有如下优势

http2	http1.1
用于数据传输的二进制分帧	http1.1是基于文本协议
同一tcp连接支持流式传输，故支持发送多个并行请求、调用	应答模型：http1.1在一个tcp连接上完成[请求/响应]是串行的
减少网络使用率的标头压缩	头部带有大量信息，每次都要重复发送

http2的二进制分帧、流式传输 能力支撑了grpc框架近乎本地的实时服务互调；

http2的多路复用（单tcp连接上并发多个请求，不多占文件描述符）、二进制编码协议、头部压缩支撑了grpc本地互调的高性能。

这里要指出：

• HTTP/2 上的通信只需要在一个TCP连接完成，在这个连接上可见的形态是帧frame和流stream。

• 而消息(或者说业务调用，业务上的逻辑发送单位)由一或多个帧组成，这些帧可以乱序发送，然后根据每个帧首部的流标识符重新组装。

• “gRPC双向流通信streaming”与”http2的流式stream传输”是一个东西吗？
  http2流式传输stream是一种底层的传输方式，其作用是支撑单连接多路复用 。
  grpc流式通信streaming，更接近业务通信级别的通信方式，grpc流式通信可用于替代高性能场景下的一元gRPC调用。

我们假设HTTP/2协议中1次RPC请求使用1个并发Stream，每个RPC消息又可 通过帧体中 Length-Prefixed Message 头部确立了边界，这样，在 Stream 中连续地发送多个 DATA 帧，就可以实现流模式 RPC。
https://juejin.cn/post/7249522846211801147

---

在编程实践上 除了实现接口业务外， 我们通常会另外实现一些拦截器（中间件）
gRPC 为在每个 ClientConn/Server 基础上实现和安装拦截器提供了一些简单的 API。 拦截器拦截每个 RPC 调用的执行。用户可以使用拦截器进行日志记录、身份验证/授权、指标收集以及许多其他可以跨 RPC 共享的功能。

在gRPC中拦截器分为普通拦截器和流拦截器两类。普通拦截器也成为一元拦截器。流拦截器用于流式调用。而客户端和服务端都有自己的普通拦截器和流拦截器类型。

//一元服务器拦截器函数类型
type UnaryServerInterceptor func(ctx context.Context, req interface{}, info *UnaryServerInfo, handler UnaryHandler) (resp interface{}, err error)

//流式服务器拦截器函数类型
type StreamServerInterceptor func(srv interface{}, ss ServerStream, info *StreamServerInfo, handler StreamHandler) error

以下是一个recover拦截器的例子：

func RecoverInterceptorDemo() grpc.UnaryServerInterceptor {
 return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (_ interface{}, err error) {
  defer func() {
   if r := recover(); r != nil {
    fmt.Println("recover panic...")
    err = status.Errorf(codes.Internal, "Panic,%v", r)
   }
  }()
  fmt.Println("in interceptor...")
  resp, err := handler(ctx, req)
  fmt.Println("leave interceptor...")
  return resp, err
 }
}

4. 一个grpc内存泄漏的示例

底层的http2协议给予了grpc很大的性能表现，但同时也带来了新的性能瓶颈， 现在现在压力给到了tcp连接。

通常情况下,一个HTTP/2 tcp连接中流的数量是有限制的，一般服务器默认为100,不同的语言有不同的应对策略。

.net

.net tcp连接上默认流数量为100，当该连接中的grpc调用导致到达”连接流的限制”，新的grpc调用会进行排队,这个时候会出现因Concurrent stream=100引起的性能问题。

.NET引入了EnableMultipleHttp2Connections可在此时产生新的连接对象。

var channel = GrpcChannel.ForAddress("https://localhost", new GrpcChannelOptions
{
    HttpHandler = new SocketsHttpHandler
    {
        EnableMultipleHttp2Connections = true,

        // ...configure other handler settings
    }
});

golang

[golang tcp连接上流数量的默认限制也是100](https://github.com/grpc/grpc-go/blob/master/internal/transport/http2_client.go）defaultMaxStreamsClient =100.
可通过ServerOption: MaxConcurrentStreams修改， 从实现上默认的并发流的限制，被设计成池的概念，

类似与C# 也有利用tcp连接池来规避该问题的方案, 按下不表。
https://segmentfault.com/a/1190000041716350?utm_source=sf-similar-article
https://github.com/shimingyah/pool

我们先来分析grpc基于http2 stream概念发起调用的源码。

创建一个gRPC 客户端连接，会创建的几个协程：

1）transport.loopyWriter.run 往服务端发送数据协程

2）transport.http2Client.reader 读取服务端数据协程

http2Client的基础结构

// 源码所在文件：google.golang.org/grpc/http2_client.go
// http2Client 实现了接口 ClientTransport
// http2Client implements the ClientTransport interface with HTTP2.
type http2Client struct {
  conn net.Conn       // underlying communication channel
  loopy *loopyWriter  // 生产和消费关联的队列在这里面，所在文件：controlbuf.go

  // controlBuf delivers all the control related tasks (e.g., window
	// updates, reset streams, and various settings) to the controller.
	controlBuf *controlBuffer // 所在文件：controlbuf.go
  
  maxConcurrentStreams  uint32
  streamQuota           int64
  streamsQuotaAvailable chan struct{}
  waitingStreams        uint32
  
  initialWindowSize int32
}

type controlBuffer struct {
  list *itemList // 队列
}

type loopyWriter struct {
  // 关联上 controlBuffer，
  // 消费 controlBuffer 中的队列 list，
  // 生产 由http2Client通过controlBuffer 进行。
  cbuf *controlBuffer
}

创建grpc客户端的行为

func newHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts ConnectOptions, onPrefaceReceipt func(), onGoAway func(GoAwayReason), onClose func()) (_ *http2Client, err error) {
  conn, err 服务器托管网:= dial(connectCtx, opts.Dialer, addr.Addr)
  t.controlBuf = newControlBuffer(t.ctxDone) // 含发送队列的初始化

  if t.keepaliveEnabled {
		t.kpDormancyCond = sync.NewCond(&t.mu)
		go t.keepalive() // 保活协程
	}
  
  // Start the reader goroutine for incoming message. Each transport has
	// a dedicated goroutine which reads HTTP2 frame from network. Then it
	// dispatches the frame to the corresponding stream entity.
  go t.reader()
  
  // Send connection preface to server.
	n, err := t.conn.Write(clientPreface)
  
  go func() {
    t.loopy = newLoopyWriter(clientSide, t.framer, t.controlBuf, t.bdpEst)
    err := t.loopy.run()
  }
}

特别说明：
每一次 gRPC 调用，客户端均会创建一个新的 Stream，
该特性使得同一 gRPC 连接可以同时处理多个调用。请求的发送并不是同步的，而是基于队列的异步发送。

// 源码所在文件：internal/transport/controlbuf.go
func (l *loopyWriter) run() (err error) {
  // 通过 get 间接调用 dequeue 和 dequeueAll
  for {
    it, err := l.cbuf.get(true)
    if err != nil {
			return err
		}
		if err = l.handle(it); err != nil {
			return err
		}
		if _, err = l.processData(); err != nil {
			return err
		}
  }
}

//  尝试去锁定流， 锁定失败会重试
func (a *csAttempt) newStream() error {
	cs := a.cs
	cs.callHdr.PreviousAttempts = cs.numRetries
	s, err := a.t.NewStream(a.ctx, cs.callHdr)
	if err != nil {
		// Return without converting to an RPC error so retry code can
		// inspect.
		return err
	}
	cs.attempt.s = s
	cs.attempt.p = &parser{r: s}
	return nil
}

func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Stream, err error) {
  // 内存上涨，是因为队列中存了大量的 headerFields 和 s
  headerFields, err := t.createHeaderFields(ctx, callHdr)
  s := t.newStream(ctx, callHdr)
  // hdr 聚合了 headerFields 和 s
  hdr := &headerFrame{
    hf:        headerFields,
    initStream: func(id uint32) error {
      t.activeStreams[id] = s
    },
    wq:         s.wq,
  }
  for {
    // 调用 executeAndPut 入队（生产）
    // 内存上涨，是因为队列中存了大量的 hdr 。
    success, err := t.controlBuf.executeAndPut(func(it interface{}) bool {
    }, hdr)
  }
}

每一个gRPC客户端连接均有一个自己的队列，gRPC 并没有直接限定队列大小，所以如果不加任何限制则会内存暴涨，直到OOM发生。

这里我们结合内存泄漏的案例加深tcp连接在grpc调用中的独特作用。

go func() {
		for {
			ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
			defer cancel()
			resp, err := c.election.Leader(ctx)
			if err != nil {
				log.WithError(err).Errorf("get leader error. test")
			} else {
				log.Infof("get test leader success :  %s", string(resp.Kvs[0].Value))
			}
			 // cancel()
		}
	}()

pprof 显示这个election.Leader函数导致的内存持续增长。

bug分析

etcd v3 发起的请求为grpc请求，

• 因为从closed信道能持续读取零值，形成死循环。

• 死循环中，用于上下文释放资源的defer cancel() 无法得到执行，因为defer是在函数执行完才会被执行。

• 每次grpc调用均会利用流，并发流资源未能快速释放， 阻塞在队列，形成缓慢内存泄漏。

在设计 gRPC 服务时，合理配置并发流的数量以及处理队列中等待请求的策略是非常重要的。这有助于确保服务的稳定性和高效性

服务器托管，北京服务器托管，服务器租用 http://www.fwqtg.net
机房租用，北京机房租用，IDC机房托管， http://www.fwqtg.net