GRPC源码解读

GRPC源码解读

使用的protobuf文件

syntax = "proto3";

package servers;

// The greeter 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;
}

Client

链接

conn, err := grpc.Dial("localhost:8080", grpc.WithInsecure())

建立一个 ClientConn 并将传入的 DialOption 应用到 ClientConn 的 dopts (dialOptions) 字段上。 设置默认的编解码类 protoCodec, backoff配置，负载均衡策略（roundRobin）。

从负载均衡器中拿到地址列表，启动goroutine链接各个地址（看下面的<链接address>），主协程设置超时时间（根据 dopts.timeout)），在超时时间内等待链接完成。

若是有地址列表监控Watcher(当负载均衡设置了naming.Resolver时会有)，异步监控地址列表的变更（变更时更新 ClientConn的链接列表 conns）。

链接address

若是 dopts.insecure 为false，检查证书相关信息。

将建立的 addrConn 放入到 ClientConn 的 conns map中，并销毁被替换的 addrConn。

重置 Transport:

链接地址，一旦出错，重试，重试间隔时间由 dopts.bs 指定。

若是没出错则调用负载均衡的Up方法（标记一下这个地址已经链接了而已）。

监控 Transport: 若是有错误发生就重置 Transport(关闭以前的，从新链接一个)。

调用方法

c := servers.NewGreeterClient(conn)
reply, err := c.SayHello(context.Background(), &servers.HelloRequest{"person"})

建立Client没有什么东西。

调用 SayHello 时主要代码就下面这句：

grpc.Invoke(ctx, "/servers.Greeter/SayHello", in, out, c.cc, opts...)

从 ClientConn 中获取 Transport。

利用获取到的transport 建立一个 Stream, Stream的id每次递增2，将 Stream 保存在transport的activeStreams中。设置各类 header 字段。将 header 按照 frame 发送给服务端以开启一个stream，frame最大16KB。

将 HelloRequest 编码写到 stream 中。

等待读取服务器的相应到 HelloReply。

最后关闭 stream。

roundRobin

Start

默认是没有指定 naming.Resolver 的，因此 Start 只是将地址放入到 addrs 字段中。

若是指定了 naming.Resolver, 会调用它返回一个 Watcher，而后启动 goroutine，循环从 Watcher 中获取地址变更的信息（增长或删除地址），一旦地址列表变更了，就将新的地址列表发给 addrCh 这个channel。

Up

标记地址是已经链接的了。

Get

获取上一次调用时拿到的地址的下一个地址。

若是获取不到地址，就建立 waitCh 而后等待有人关闭这个 waitCh。Up 和 Close 方法都会检测并关闭 waitCh。