golang版 grpc源码浅析

Author: zjfsdnu

git/TortoiseGit操作.md

@@ -0,0 +1,6 @@
+### 界面
+
+![TortoiseGit](C:\Users\test\Pictures\QQ截图20200119093046.png)
+
+
+

protobuf/golang版 grpc源码浅析.md

@@ -0,0 +1,161 @@
+项目中需要用到grpc作为通信框架，但是在我们项目组中没有一个人对这个grpc有深入的理解。只知道用了GRPC后，调用远程函数就像调用本地函数一样方便，能够像处理同步编程一样处理异步编程。至于，他是怎么实现的，以及网络数据格式是怎样的，他是怎样处理数据流的，需要像TCP一样处理粘包吗，等等一系问题，我们都一无所知。如果就这样使用了这套框架,可能会给项目带来潜在的风险，甚至由于这些风险可能会导致项目终止。所以我觉得有必要看看GRPC的源代码了。
+ 首先，我最关心的是，他是怎么处理数据流的。GRPC框架采用了HTTP2作为通讯协议，而HTTP2是基于TCP协议(传输控制协议)。TCP协议传输数据是通过数据流的方式传输，所以grpc也绕不开数据流的处理。
+ 先看看发送数据函数:
+
+```go
+func (cs *clientStream) SendMsg(m interface{}) (err error) {
+	defer func() {
+		if err != nil && err != io.EOF {
+			// Call finish on the client stream for errors generated by this SendMsg
+			// call, as these indicate problems created by this client.  (Transport
+			// errors are converted to an io.EOF error in csAttempt.sendMsg; the real
+			// error will be returned from RecvMsg eventually in that case, or be
+			// retried.)
+			cs.finish(err)
+		}
+	}()
+	if cs.sentLast {
+		return status.Errorf(codes.Internal, "SendMsg called after CloseSend")
+	}
+	if !cs.desc.ClientStreams {
+		cs.sentLast = true
+	}
+
+	// load hdr, payload, data
+	hdr, payload, data, err := prepareMsg(m, cs.codec, cs.cp, cs.comp)
+	if err != nil {
+		return err	}
+
+	// TODO(dfawley): should we be checking len(data) instead?
+	if len(payload) > *cs.callInfo.maxSendMessageSize {
+		return status.Errorf(codes.ResourceExhausted, "trying to send message larger than max (%d vs. %d)", len(payload), *cs.callInfo.maxSendMessageSize)
+	}
+	msgBytes := data // Store the pointer before setting to nil. For binary logging.
+	op := func(a *csAttempt) error {
+		err := a.sendMsg(m, hdr, payload, data)
+		// nil out the message and uncomp when replaying; they are only needed for
+		// stats which is disabled for subsequent attempts.
+		m, data = nil, nil
+		return err	}
+	err = cs.withRetry(op, func() { cs.bufferForRetryLocked(len(hdr)+len(payload), op) })
+	if cs.binlog != nil && err == nil {
+		cs.binlog.Log(&binarylog.ClientMessage{
+			OnClientSide: true,
+			Message:      msgBytes,
+		})
+	}
+	return}
+```
+
+这段代码,主要进行了消息头的封装,发送数据长度检查,以及将数据放到待发送缓冲区等工作。
+ prepareMsg这个函数，里面做了数据编码(如果设置了编码函数),数据压缩(如果设置了压缩函数)，和消息头的封装。消息头格式为:压缩标志+消息长度，总共占5字节。其中，压缩标志占1个字节,长度占4个字节。网络数据格式如下图:
+ ![grpc网络数据格式](http://image.okcode.net/F7B15F45104CF4BC81E152F39BB4CA82.jpg)
+ 讲完了理论，还是看看源码吧
+ prepareMsg函数如下:
+
+```go
+func prepareMsg(m interface{}, codec baseCodec, cp Compressor, comp encoding.Compressor) (hdr, payload, data []byte, err error) {
+	if preparedMsg, ok := m.(*PreparedMsg); ok {
+		return preparedMsg.hdr, preparedMsg.payload, preparedMsg.encodedData, nil
+	}
+	// The input interface is not a prepared msg.
+	// Marshal and Compress the data at this point
+	data, err = encode(codec, m)
+	if err != nil {
+		return nil, nil, nil, err	}
+	compData, err := compress(data, cp, comp)
+	if err != nil {
+		return nil, nil, nil, err	}
+	hdr, payload = msgHeader(data, compData)
+	return hdr, payload, data, nil}
+```
+
+msgHeader函数如下:
+
+```go
+func msgHeader(data, compData []byte) (hdr []byte, payload []byte) {
+	hdr = make([]byte, headerLen)
+	if compData != nil {
+		hdr[0] = byte(compressionMade)
+		data = compData	} else {
+		hdr[0] = byte(compressionNone)
+	}
+
+	// Write length of payload into buf
+	binary.BigEndian.PutUint32(hdr[payloadLen:], uint32(len(data)))
+	return hdr, data}
+```
+
+看完了grpc发送数据代码(当然只是看了发送数据的格式,真正调用tcp socket  的send函数并没有展示,因为没有太大意义),我们可以推测接收数据函数肯定会按照发送的数据格式,进行解压，进行粘包,组装一个完整的数据包后，然后抛给应用层。那么就让我们来看看是否和我们的猜想一致吧！
+ 额，接收网络数据函数要稍微复杂一些，函数嵌套深度达16层之多！这里就不把所有的函数都贴出来,喜欢研究源代码的童鞋自己去官方下载源码吧。在这里，我们说一下，GRPC接收数据的流程,并贴出一些比较关键的源码。
+ 接收消息头部代码
+
+```go
+func (p *parser) recvMsg(maxReceiveMessageSize int) (pf payloadFormat, msg []byte, err error) {
+	if _, err := p.r.Read(p.header[:]); err != nil {
+		return 0, nil, err	}
+
+	pf = payloadFormat(p.header[0])
+	length := binary.BigEndian.Uint32(p.header[1:])
+
+	if length == 0 {
+		return pf, nil, nil
+	}
+	if int64(length) > int64(maxInt) {
+		return 0, nil, status.Errorf(codes.ResourceExhausted, "grpc: received message larger than max length allowed on current machine (%d vs. %d)", length, maxInt)
+	}
+	if int(length) > maxReceiveMessageSize {
+		return 0, nil, status.Errorf(codes.ResourceExhausted, "grpc: received message larger than max (%d vs. %d)", length, maxReceiveMessageSize)
+	}
+	// TODO(bradfitz,zhaoq): garbage. reuse buffer after proto decoding instead
+	// of making it for each message:
+	msg = make([]byte, int(length))
+	if _, err := p.r.Read(msg); err != nil {
+		if err == io.EOF {
+			err = io.ErrUnexpectedEOF		}
+		return 0, nil, err	}
+	return pf, msg, nil}
+```
+
+注意，接收单个数据包最大长度为maxReceiveMessageSize ，默认是 4194304（4M）。
+ 首先读取消息头部(5字节)。然后解析出压缩标志和消息包长度(不含消息头部大小)。最后根据消息长度读取消息内容。
+ 上面代码不能看出GRPC框架是怎么粘包的，不着急，我们往下看：
+ 列出函数调用关系，具体函数体不在下面贴出来
+  Stream.Read()->io.ReadFull()->io.ReadAtLeast()->transportReader.Read()->recvBufferReader.Read()->recvBufferReader.readClient()
+ ReadAtLeast这个函数会读取指定长度数据后再返回：
+
+```go
+func ReadAtLeast(r Reader, buf []byte, min int) (n int, err error) {
+	if len(buf) < min {
+		return 0, ErrShortBuffer	}
+	for n < min && err == nil {
+		var nn int
+		nn, err = r.Read(buf[n:])
+		n += nn	}
+	if n >= min {
+		err = nil
+	} else if n > 0 && err == EOF {
+		err = ErrUnexpectedEOF	}
+	return}
+```
+
+这个函数的核心就是那个for循环啦，只要读取的数据没有达到最低要求，就不断尝试读取数据，直到读够数据为止。其实粘包就在这里悄悄的实现了。
+ readClient从接收数据channel里面取出数据:
+
+```go
+func (r *recvBufferReader) readClient(p []byte) (n int, err error) {
+	// If the context is canceled, then closes the stream with nil metadata.
+	// closeStream writes its error parameter to r.recv as a recvMsg.
+	// r.readAdditional acts on that message and returns the necessary error.
+	select {
+	case <-r.ctxDone:
+		r.closeStream(ContextErr(r.ctx.Err()))
+		m := <-r.recv.get()
+		return r.readAdditional(m, p)
+	case m := <-r.recv.get():
+		return r.readAdditional(m, p)
+	}}
+```
+
+GRPC框架采用HTTP2协议，而HTTP2是用过数据帧传输数据,当数据帧到达后,就放入接收数据channel。readClient就可以从接收数据channel里面去数据啦。在传输数据较小(小于16K)的情况下，单个数据帧就能传输完毕。但是如果单个数据帧不能完成数据传输，就会用到上面提到的粘包功能了(这里也可以叫做粘帧,毕竟是通过数据帧传输数据的)。
+ 相信通过以上分析，能对GRPC网络通信这块有一个大概的了解。这里面并没有提到HTTP2的Hpack思想,哈夫曼编码(Huffman Coding)等实现细节。有兴趣的朋友可以自己去了解下，这些仍然值得我们学习。