HTTP 2.0 in gRPC

Posted 2021-04-11 码农如酒

gRPC为什么使用HTTP/2呢？先了解HTTP/2的新特性，再利用gRPC源码深入了解。

HTTP/2 新特性

• 二进制分帧（HTTP Frames）：通信协议采用Frame格式；

• 多路复用：同一个连接支持多次请求；

• 头部压缩：减少传输数据；

• 服务端推送（Server Push）：服务端可以推送Stream给客户端；

• 流量控制（Flow Control）:窗口滑动机制，通知对方可接受的数据大小；

提醒：HTTP/2详细，请自行了解。

图解gRPC调用

说明：

1. HTTP/2 建立的连接是多路复用的，节点之间没必要建立连接池；

2. HTTP/2 Client发送Stream，传输Frame；客户端发送的Stream ID为自增的单数；

3. HTTP/2 Server接收Frame，处理Stream；根据Client端请求的Stream ID进行响应；

4. Server端每次新建Goroutine处理新的Stream ID；注：控制并发量，Groutine超时机制；

gRPC Client端源码解析

Client端连接初始化流程：

1. 建立tcp连接，初始化http2Client对象；

2. 开启keepalive和reader协程；

3. 确认HTTP/2建立标识；

4. 启动流控机制协程；

func newHTTP2Client(connectCtx, ctx context.Context, addr TargetInfo, opts ConnectOptions, onPrefaceReceipt func(), onGoAway func(GoAwayReason), onClose func()) (_ *http2Client, err error) { //......此处省略部分代码 conn, err := dial(connectCtx, opts.Dialer, addr.Addr) //......此处省略部分代码 t := &http2Client{ //.......此处省略部分代码 } t.controlBuf = newControlBuffer(t.ctxDone) //......此处省略部分代码 if t.keepaliveEnabled { go t.keepalive() }
 go t.reader() //确认HTTP/2建立标识 n, err := t.conn.Write(clientPreface) //......此处省略部分代码 go func() { t.loopy = newLoopyWriter(clientSide, t.framer, t.controlBuf, t.bdpEst) err := t.loopy.run() //......此处省略部分代码 }() return t, nil}

invoke函数功能：

1. 初始化局部对象——ClientStream，并发送HeadersFrame至ControlBuffer;

2. 发送req数据至ControlBuffer；

3. 接收ClientStream从服务端响应的数据；

func invoke(ctx context.Context, method string, req, reply interface{}, cc *ClientConn, opts ...CallOption) error { cs, err := newClientStream(ctx, unaryStreamDesc, cc, method, opts...) if err != nil { return err } if err := cs.SendMsg(req); err != nil { return err } return cs.RecvMsg(reply)}

gRPC Server端源码解析

Client端与Server端建立连接流程：

1. Server端初始化HTTP/2连接，并开启keepalive、流控等；

2. Server端对Client端请求Stream处理——独立的Goroutine；

3. 连接关闭处理，并退出Goroutine；

func (s *Server) handleRawConn(rawConn net.Conn) { //......此处省略部分代码 st := s.newHTTP2Transport(conn, authInfo) if st == nil { return }
 rawConn.SetDeadline(time.Time{}) if !s.addConn(st) { return }  go func() { s.serveStreams(st) s.removeConn(st) }()}
func newHTTP2Server(conn net.Conn, config *ServerConfig) (_ ServerTransport, err error) { //......此处省略部分代码 t := &http2Server{ //......此处省略部分代码 } t.controlBuf = newControlBuffer(t.ctxDone) //......此处省略部分代码 go func() { t.loopy = newLoopyWriter(serverSide, t.framer, t.controlBuf, t.bdpEst) t.loopy.ssGoAwayHandler = t.outgoingGoAwayHandler if err := t.loopy.run(); err != nil { errorf("transport: loopyWriter.run returning. Err: %v", err) } t.conn.Close() close(t.writerDone) }() go t.keepalive() return t, nil}

HandleStreams函数处理Client端发送的Frame，重点处理的Frame：

• MetaHeadersFrame处理，采用函数式调用开启业务逻辑协程（handleStream函数）；

• DataFrame处理，写入Stream的recvBuffer中，业务处理时可从recvBuffer中获取数据；

func (s *Server) serveStreams(st transport.ServerTransport) { defer st.Close() var wg sync.WaitGroup st.HandleStreams(func(stream *transport.Stream) { wg.Add(1) go func() { defer wg.Done() s.handleStream(st, stream, s.traceInfo(st, stream)) }() }, func(ctx context.Context, method string) context.Context { if !EnableTracing { return ctx } tr := trace.New("grpc.Recv."+methodFamily(method), method) return trace.NewContext(ctx, tr) }) wg.Wait()}
func (t *http2Server) HandleStreams(handle func(*Stream), traceCtx func(context.Context, string) context.Context) { defer close(t.readerDone) for { frame, err := t.framer.fr.ReadFrame() atomic.StoreUint32(&t.activity, 1) if err != nil { if se, ok := err.(http2.StreamError); ok { //......此处省略部分代码 } warningf("transport: http2Server.HandleStreams failed to read frame: %v", err) t.Close() return } switch frame := frame.(type) { case *http2.MetaHeadersFrame: if t.operateHeaders(frame, handle, traceCtx) { t.Close() break } case *http2.DataFrame: t.handleData(frame) //......此处省略部分代码 default: errorf("transport: http2Server.HandleStreams found unhandled frame type %v.", frame) } }}

总结：

gRPC充分的利用了HTTP/2 的新特性（二进制分帧、多路复用、头部压缩、服务端推送、流量控制等），减少服务之间连接，但增加处理Steam的逻辑，每次请求服务端增加Wroker Goroutine开销，并且需要注意Wroker Goroutine并发量和执行时间；