gRPC 内容介绍

待我称王封你为后i 2022-06-13 10:15 146阅读 0赞

### gRPC概念和基本思想 ###

**概念**：

gRPC是Googel基于HTTP/2以及protobuf的。gRPC通常有四种模式，unary，client streaming， server streaming 以及bidirectional streaming，但无论哪一种模式对底层的HTTP/2来说都是stream，所以总结来看，grpc仍是一套request+response的模型。

**基本思想：**

定义一个服务，指定其可以被远程调用的方法及其参数和返回类型。gRPC默认使用

[protocol buffers][] 作为接口定义语言（可以替换protocol buffers），来描述服务接口和消息结构。

所以，在深入分析gRPC server如何工作之前，先了解下HTTP/2的特性和概念。

### HTTP/2基本概念及特性解释 ###

*  二进制分帧
    
     *  Stream： 一个双向流，一条连接可以有多个 streams。
     *  Message： 也就是逻辑上面的 request，response。
     *  Frame:：数据传输的最小单位。每个 Frame 都属于一个特定的 stream 或者整个连接。一个 message 可能有多个 frame 组成。
    
    HTTP/2将每次request、response以帧为单位进行了更细小的划分，同时抽象了流的概念，所有的帧都在特定的stream上进行传输，帧可以在数据流上乱序发送，然后再根据每个帧首部的流标识符重新组装。
    
    **帧格式介绍**：[http://httpwg.org/specs/rfc7540.html\#rfc.section.4.1][http_httpwg.org_specs_rfc7540.html_rfc.section.4.1]
 *  多路复用（Multiplexing）
    
    在HTTP/1.1协议中，浏览器客户端在同一时间，针对同一域名下的请求有一定数量限制。超过限制数量的请求会被阻塞。
    
    多路复用允许同时通过单一的HTTP/2连接发起多重请求-响应消息。HTTP/2 通过 stream 支持了连接的多路复用，提高了连接的利用率。Stream 有很多重要特性：
    
     *  一条连接可以包含多个 streams，多个 streams 发送的数据互相不影响。
     *  Stream 可以被 client 和 server 单方面或者共享使用。
     *  Stream 可以被任意一段关闭。
     *  Stream 会确定好发送 frame 的顺序，另一端会按照接受到的顺序来处理。
     *  Stream 用一个唯一 ID 来标识。
    
    这里在说一下 Stream ID，如果是 client 创建的 stream，ID 就是奇数，如果是 server 创建的，ID 就是偶数。ID 0x00 和 0x01 都有特定的使用场景。
    
    Stream ID 不可能被重复使用，如果一条连接上面 ID 分配完了，client 会新建一条连接。而 server 则会给 client 发送一个 GOAWAY frame 强制让 client 新建一条连接。
    
    
    
    ![多路复用!][SouthEast]
 *  服务器推送
    
    通常，只有在浏览器请求某个资源的时候，服务器才会向浏览器发送该资源。Server Push则允许服务器在收到浏览器的请求之前，主动向浏览器推送资源。比如说，网站首页引用了一个CSS文件。浏览器在请求首页时，服务器除了返回首页的HTML之外，可以将其引用的 CSS文件也一并推给客户端。
    
    Server push是HTTP/2中一个很强大的功能：
    
     *  服务器除了响应客户端的请求外，还可以向客户端额外推送资源。
     *  服务器推送的资源有自己独立的URL， 可以被浏览器缓存，可以达到多页面共享。
     *  资源推送遵守同源策略，服务器不可随便推送第三方资源给客户端。
     *  客户端可以拒绝推送过来的资源。
 *  头部压缩
    
     *  HTTP/2在客户端和服务器端使用“首部表”来跟踪和存储之前发送的键－值对，对于相同的数据，不再通过每次请求和响应发送；
     *  首部表在HTTP/2的连接存续期内始终存在，由客户端和服务器共同渐进地更新;
     *  每个新的首部键－值对要么被追加到当前表的末尾，要么替换表中之前的值。
    
    例如：下图中的两个请求， 请求一发送了所有的头部字段，第二个请求则只需要发送差异数据，这样可以减少冗余数据，降低开销
    
    ![头部压缩][SouthEast 1]

## gRPC Server源码分析 ##

gRPC Server主要完成以下功能：

*  将proto中定义的service的真正实现注册到grpc的Server中。
 *  管理Server的Socket以及在Socket上传输数据的过程。

### 注册流程 ###

以下以grpc/examples/helloworld为例进行分析。

*  定义proto文件

syntax = "proto3";
    
    option java_multiple_files = true;
    option java_package = "io.grpc.examples.helloworld";
    option java_outer_classname = "HelloWorldProto";
    package helloworld;
    // 定义Greeter service的方法以及方法对应的request、response
    service Greeter {
      rpc SayHello (HelloRequest) returns (HelloReply) {}
    }
    message HelloRequest {
      string name = 1;
    }
    message HelloReply {
      string message = 1;
    }

*  将Greeter server的真正实现注册到grpc Server

// 定义Server
    type server struct{}
    
    // 实现 helloworld.proto对应的方法
    func (s *server) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloReply, error) {
        return &pb.HelloReply{Message: "Hello " + in.Name}, nil
    }
    
    func main() {
        lis, err := net.Listen("tcp", port)
        if err != nil {
            log.Fatalf("failed to listen: %v", err)
        }
        s := grpc.NewServer()
      //将真正实现的GreeterServer注册到grpc server
        pb.RegisterGreeterServer(s, &server{})
        if err := s.Serve(lis); err != nil {
            log.Fatalf("failed to serve: %v", err)
        }
    }

*  注册函数的分析

//生成的hello.pb.go对Greeter service的描述
    var _Greeter_serviceDesc = grpc.ServiceDesc{
        ServiceName: "helloworld.Greeter",
        HandlerType: (*GreeterServer)(nil),
        Methods: []grpc.MethodDesc{
            {
                MethodName: "SayHello",
                Handler:    _Greeter_SayHello_Handler,
            },
        },
        Streams:  []grpc.StreamDesc{},
        Metadata: fileDescriptor0,
    }
    ——————————————————————————————————————pb.go生成的注册函数——————————————————————————————————————
    func RegisterGreeterServer(s *grpc.Server, srv GreeterServer) {
        s.RegisterService(&_Greeter_serviceDesc, srv)
    }
    ——————————————————————————————grpc暴露的注册函数————————————————————————————————————————————
    //将自定的Greeter Service注册到grpc Server
    func (s *Server) RegisterService(sd *ServiceDesc, ss interface{}) {
        ht := reflect.TypeOf(sd.HandlerType).Elem()
        st := reflect.TypeOf(ss)
      //判断是否符合proto中定义的方法
        if !st.Implements(ht) {
            grpclog.Fatalf("grpc: Server.RegisterService found the handler of type %v that does not satisfy %v", st, ht)
        }
        s.register(sd, ss)
    }
    
    func (s *Server) register(sd *ServiceDesc, ss interface{}) {
        s.mu.Lock()
        defer s.mu.Unlock()
        s.printf("RegisterService(%q)", sd.ServiceName)
        if _, ok := s.m[sd.ServiceName]; ok {
            grpclog.Fatalf("grpc: Server.RegisterService found duplicate service registration for %q", sd.ServiceName)
        }
      //初始化srv
        srv := &service{
            server: ss,
            md:     make(map[string]*MethodDesc),
            sd:     make(map[string]*StreamDesc),
            mdata:  sd.Metadata,
        }
      //保存unary method对应的handler
        for i := range sd.Methods {
            d := &sd.Methods[i]
            srv.md[d.MethodName] = d
        }
      //保存stream(client, server, bidirectional)对应对应的handler
        for i := range sd.Streams {
            d := &sd.Streams[i]
            srv.sd[d.StreamName] = d
        }
      //把服务注册到grpc server中
        s.m[sd.ServiceName] = srv
    }

### Socket和传输数据的管理 ###

//在做以下函数在说明的时候，去掉一些错误处理的代码和非主要逻辑的代码。
    //可自行查看grpc/server.go的源码进行理解
      func (s *Server) Serve(lis net.Listener) error {
        for {
          rawConn, err := lis.Accept()
          go s.handleRawConn(rawConn)
        }
      }
    //handleRawConn函数说明
    func (s *Server) handleRawConn(rawConn net.Conn) {
        conn, authInfo, err := s.useTransportAuthenticator(rawConn)
        //s.opts在上一步中并未设置useHandlerImpl,进一步分析serveNewHTTP2Transport
        if s.opts.useHandlerImpl {
            s.serveUsingHandler(conn)
        } else {
            s.serveNewHTTP2Transport(conn, authInfo)
        }
    }
    //serveNewHTTP2Transport分析
    func (s *Server) serveNewHTTP2Transport(c net.Conn, authInfo credentials.AuthInfo) {
    //创建http2的transport,可自己配置该connection上最大streams数量 
      st, err := transport.NewServerTransport("http2", c, s.opts.maxConcurrentStreams, authInfo)
        if !s.addConn(st) {
            st.Close()
            return
        }
        s.serveStreams(st)
    }
    //调用该transport上stream对应handler函数
    func (s *Server) serveStreams(st transport.ServerTransport) {
        defer s.removeConn(st)
        defer st.Close()
        var wg sync.WaitGroup
        st.HandleStreams(func(stream *transport.Stream) {
            wg.Add(1)
            go func() {
                defer wg.Done()
              //真正的steam处理函数
                s.handleStream(st, stream, s.traceInfo(st, stream))
            }()
        })
        wg.Wait()
    }
    
    func (s *Server) handleStream(t transport.ServerTransport, stream *transport.Stream, trInfo *traceInfo) {
        sm := stream.Method()
        if sm != "" && sm[0] == '/' {
            sm = sm[1:]
        }
        pos := strings.LastIndex(sm, "/")
        if pos == -1 {
            if trInfo != nil {
                trInfo.tr.LazyLog(&fmtStringer{
       "Malformed method name %q", []interface{}{sm}}, true)
                trInfo.tr.SetError()
            }
            if err := t.WriteStatus(stream, codes.InvalidArgument, fmt.Sprintf("malformed method name: %q", stream.Method())); err != nil {
                if trInfo != nil {
                    trInfo.tr.LazyLog(&fmtStringer{
       "%v", []interface{}{err}}, true)
                    trInfo.tr.SetError()
                }
                grpclog.Printf("grpc: Server.handleStream failed to write status: %v", err)
            }
            if trInfo != nil {
                trInfo.tr.Finish()
            }
            return
        }
      //从请求中分离出调的服务和方法
        service := sm[:pos]
        method := sm[pos+1:]
      //从grpc server中分离出注册的service信息，此处为greeterServer
        srv, ok := s.m[service]
        if !ok {
            if trInfo != nil {
                trInfo.tr.LazyLog(&fmtStringer{
       "Unknown service %v", []interface{}{service}}, true)
                trInfo.tr.SetError()
            }
            if err := t.WriteStatus(stream, codes.Unimplemented, fmt.Sprintf("unknown service %v", service)); err != nil {
                if trInfo != nil {
                    trInfo.tr.LazyLog(&fmtStringer{
       "%v", []interface{}{err}}, true)
                    trInfo.tr.SetError()
                }
                grpclog.Printf("grpc: Server.handleStream failed to write status: %v", err)
            }
            if trInfo != nil {
                trInfo.tr.Finish()
            }
            return
        }
        // Unary RPC or Streaming RPC?
        if md, ok := srv.md[method]; ok {
            s.processUnaryRPC(t, stream, srv, md, trInfo)
            return
        }
        if sd, ok := srv.sd[method]; ok {
            s.processStreamingRPC(t, stream, srv, sd, trInfo)
            return
        }
        if trInfo != nil {
            trInfo.tr.LazyLog(&fmtStringer{
       "Unknown method %v", []interface{}{method}}, true)
            trInfo.tr.SetError()
        }
      //client调用了server端并未实现的method
        if err := t.WriteStatus(stream, codes.Unimplemented, fmt.Sprintf("unknown method %v", method)); err != nil {
            if trInfo != nil {
                trInfo.tr.LazyLog(&fmtStringer{
       "%v", []interface{}{err}}, true)
                trInfo.tr.SetError()
            }
            grpclog.Printf("grpc: Server.handleStream failed to write status: %v", err)
        }
        if trInfo != nil {
            trInfo.tr.Finish()
        }
    }
    
    
    //最后，真实的处理stream上recv message和发送response
    func (s *Server) processUnaryRPC(t transport.ServerTransport, stream *transport.Stream, srv *service, md *MethodDesc, trInfo *traceInfo) (err error) {
        p := &parser{r: stream}
        for {
          //recvMsg从指定流获取message，返回它的message和payload format格式
            pf, req, err := p.recvMsg(s.opts.maxMsgSize)
    
            //check stream上传输的message压缩算法是否符合要求
            if err := checkRecvPayload(pf, stream.RecvCompress(), s.opts.dc); err != nil {
                switch err := err.(type) {
                case *rpcError:
                    if err := t.WriteStatus(stream, err.code, err.desc); err != nil {
                        grpclog.Printf("grpc: Server.processUnaryRPC failed to write status %v", err)
                    }
                default:
                    if err := t.WriteStatus(stream, codes.Internal, err.Error()); err != nil {
                        grpclog.Printf("grpc: Server.processUnaryRPC failed to write status %v", err)
                    }
    
                }
                return err
            }
            statusCode := codes.OK
            statusDesc := ""
            df := func(v interface{}) error {
                if pf == compressionMade {
                    var err error
                    req, err = s.opts.dc.Do(bytes.NewReader(req))
                    if err != nil {
                        if err := t.WriteStatus(stream, codes.Internal, err.Error()); err != nil {
                            grpclog.Printf("grpc: Server.processUnaryRPC failed to write status %v", err)
                        }
                        return err
                    }
                }
                if len(req) > s.opts.maxMsgSize {
                    // TODO: Revisit the error code. Currently keep it consistent with
                    // java implementation.
                    statusCode = codes.Internal
                    statusDesc = fmt.Sprintf("grpc: server received a message of %d bytes exceeding %d limit", len(req), s.opts.maxMsgSize)
                }
                if err := s.opts.codec.Unmarshal(req, v); err != nil {
                    return err
                }
                if trInfo != nil {
                    trInfo.tr.LazyLog(&payload{sent: false, msg: v}, true)
                }
                return nil
            }
          //调用之前注册的函数的handler
            reply, appErr := md.Handler(srv.server, stream.Context(), df, s.opts.unaryInt)
            if appErr != nil {
                if err, ok := appErr.(*rpcError); ok {
                    statusCode = err.code
                    statusDesc = err.desc
                } else {
                    statusCode = convertCode(appErr)
                    statusDesc = appErr.Error()
                }
                if trInfo != nil && statusCode != codes.OK {
                    trInfo.tr.LazyLog(stringer(statusDesc), true)
                    trInfo.tr.SetError()
                }
                if err := t.WriteStatus(stream, statusCode, statusDesc); err != nil {
                    grpclog.Printf("grpc: Server.processUnaryRPC failed to write status: %v", err)
                    return err
                }
                return nil
            }
            if trInfo != nil {
                trInfo.tr.LazyLog(stringer("OK"), false)
            }
            opts := &transport.Options{
                Last:  true,
                Delay: false,
            }
          //将结果返回给client端
            if err := s.sendResponse(t, stream, reply, s.opts.cp, opts); err != nil {
                switch err := err.(type) {
                case transport.ConnectionError:
                    // Nothing to do here.
                case transport.StreamError:
                    statusCode = err.Code
                    statusDesc = err.Desc
                default:
                    statusCode = codes.Unknown
                    statusDesc = err.Error()
                }
                return err
            }
            if trInfo != nil {
                trInfo.tr.LazyLog(&payload{sent: true, msg: reply}, true)
            }
            return t.WriteStatus(stream, statusCode, statusDesc)
        }
    }

至此，基于gRPC server端的源码分析全部结束。笔者对于gRPC的理解仍不够深入，如果发现理解有误的地方，欢迎指出。

## 参考资料 ##

*  gRPC官方文档中文版——[http://doc.oschina.net/grpc?t=58009][http_doc.oschina.net_grpc_t_58009]
 *  深入了解gRPc：协议——[http://www.jianshu.com/p/48ad37e8b4ed][http_www.jianshu.com_p_48ad37e8b4ed]
 *  详解HTTP/2 server push——[http://geek.csdn.net/news/detail/208615][http_geek.csdn.net_news_detail_208615]
 *  HTTP/2.0相对于HTTP/1.1的重大改进——[https://www.zhihu.com/question/34074946][https_www.zhihu.com_question_34074946]
 *  golang版本grpc服务端浅析——[https://guidao.github.io/grpc\_server.html][https_guidao.github.io_grpc_server.html]

[protocol buffers]: https://developers.google.com/protocol-buffers/
[http_httpwg.org_specs_rfc7540.html_rfc.section.4.1]: http://httpwg.org/specs/rfc7540.html#rfc.section.4.1
[SouthEast]: /images/20220613/378bd32bcce9462997adb8fb776d4764.png
[SouthEast 1]: /images/20220613/345f5e59e7864e968d10524692659452.png
[http_doc.oschina.net_grpc_t_58009]: http://doc.oschina.net/grpc?t=58009
[http_www.jianshu.com_p_48ad37e8b4ed]: http://www.jianshu.com/p/48ad37e8b4ed
[http_geek.csdn.net_news_detail_208615]: http://geek.csdn.net/news/detail/208615
[https_www.zhihu.com_question_34074946]: https://www.zhihu.com/question/34074946
[https_guidao.github.io_grpc_server.html]: https://guidao.github.io/grpc_server.html