读golang的grpc源码

Posted 2021-04-03 半片柠檬

grpc 源码分析<golang>

grpc服务启动代码如下：

 // 初始化一个 grpc server句柄
 server := grpc.NewServer()
 // 注册 一个 服务实例 到 sever 上
pb.RegisterSearchServiceServer(server, &SearchService{})
 // 拨一个 tcp 连接 （ 异步 的 方式 ）
lis, err := net.Listen("tcp", ":"+PORT)
 // 监听这个连接
server.Serve(lis)

grpc 监听在一个连接上，对连接的处理如下：serve() 函数

for {
    // 监听链接
     rawConn, err := lis.Accept()
     go func() {
         // 处理 连接 中 的数据
          s.handleRawConn(rawConn)
    }()
}

对链接的处理里面，按grpc 执行 newHTTP2Transport创建 一个 http2 句柄， 这里需要提一下有个 字段useHandlerImpl 用于 控制，是否 使用 标准库中的 http2，显然标准库中http2 是可以解析grpc请求的，但是做出的响应到client是无法做出正确响应的。

// 从 http 解析出来的流中 获取 方法 与及服务，拿到 对应的方法。
service := sm[:pos]
method := sm[pos+1:]
srv, ok := s.m[service]
// 一元函数中存在，就执行一元函数
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
}
// -- 显然 一元函数比流式函数优先级高

rpc 一元处理，processUnaryRPC,(流式也是类似的处理,只是流式可以通过http2的特性向端上连续推送数据)

// 设置 压缩 流 的算法
stream.SetSendCompress(cp.Type())
// 读取流 里面的 数据 <获取:是否算法， 数据， 错误信息>
pf, req, err := p.recvMsg(s.opts.maxReceiveMessageSize)
// 抛出 解密反序列化 函数

df := func(v interface{}) error {
 if pf == compressionMade {
      if dc != nil {
           req, err = dc.Do(bytes.NewReader(req))
      } else {
           tmp, _ := decomp.Decompress(bytes.NewReader(req))
           req, err = ioutil.ReadAll(tmp)
      }
}
 s.getCodec(stream.ContentSubtype()).Unmarshal(req, v)
}
// 解析 编码 流 并反序列化 成 v

这里的df 在 被注册的handle 里面被调用

var _OpenApiCallBack_serviceDesc = grpc.ServiceDesc{
     ServiceName: "lx.bs.api.OpenApiCallBack",
     HandlerType: (*OpenApiCallBackServer)(nil),
     Methods: []grpc.MethodDesc{
         {
           MethodName: "AppRemoteCall",
           Handler:    _OpenApiCallBack_AppRemoteCall_Handler,
         },
    },
     Streams:  []grpc.StreamDesc{},
     Metadata: "lanxin-bs-api.git/api/open_callback.proto",
}

// 其中handler 的实现 如下 
 in := new(lx_bs_api_open_callback.AppRemoteCall_Request)
 dec(in)  // 
 if interceptor == nil {
      return srv.(OpenApiCallBackServer).AppRemoteCall(ctx, in)
}
 info := &grpc.UnaryServerInfo{
      Server:     srv,
      FullMethod: "/lx.bs.api.OpenApiCallBack/AppRemoteCall",
}
 handler := func(ctx context.Context, req interface{}) (interface{}, error) {
     return srv.(OpenApiCallBackServer).AppRemoteCall(ctx, req)
}
 return interceptor(ctx, in, info, handler)

上面的是 grpc 服务的源码基本解析。

client 客户端 调用 使用invoke，构建 clientStream 执行sendMsg

// 对 请求 编码
data, err := encode(cs.codec, m)
// 使用算法编码
compData, err := compress(data, cs.cp, cs.comp)
// 构建grpc 头 与信息
hdr, payload := msgHeader(data, compData)
// 发送数据
a.sendMsg(m, hdr, payload, data)
// http2 发送
a.t.Write(a.s, hdr, payld, &transport.Options{Last: !cs.desc.ClientStreams})

针对 grpc client

client

使用中，可以注册 拦截器，可以设置算法方式。

server

使用者，可以注册 拦截器，设置请求处理具体方式。

grpc 多目录 编译， 只要指定搜索目录即可，通过 Mxxx=xxxx,可修改具体参数

protoc -I=/--proto_path=[搜索路径] --go_out=plugins=grpc,M[xxx=xxx]:. 

使用protoc命令编译.proto文件:

-I 参数：指定import路径，可以指定多个 -I参数，按顺序查找，默认只查找当前目录–go_out ：golang编译支持，支持以下参数:1、plugins=plugin1+plugin2 - 指定插件，目前只有grpc，即：plugins=grpc2、M 参数 - 指定导入的.proto文件路径编译后对应的golang包名(不指定本参数默认就是.proto文件中import语句的路径)3、import_prefix=xxx - 为所有import路径添加前缀，主要用于编译子目录内的多个proto文件，这个参数按理说很有用，尤其适用替代hello_http.proto编译时的M参数，但是实际使用时有个蛋疼的问题，自己尝试看看吧4、import_path=foo/bar - 用于指定未声明package或go_package的文件的包名，最右面的斜线前的字符会被忽略5、编译文件路径 .proto文件路径(支持通配符)6、同一个包内包含多个.proto文件时使用通配符同时编译所有文件，单独编译每个文件会存在变量命名冲突完整示例：protoc --go_out=plugins=grpc,Mfoo/bar.proto=bar,import_prefix=foo/,import_path=foo/bar:. ./*.proto利用protoc在当前文件夹内生成pb源代码文件：

> 命令：protoc --go_out=plugins=grpc:. ./common.proto