gRPC 客户端是如何管理与服务端的连接的？

grpc.ClientConn 表示一个客户端实例与服务端之间的连接，主要包含如下数据结构：

1、grpc.connectivityStateManager(grpc.ClientConn.csMgr) 总体的连接状态

状态类型为 connectivity.State，有如下几种状态：

  • Idle
  • Connecting
  • Ready
  • TransientFailure
  • Shutdown

grpc.ClientConn 包含了多个 grpc.addrConn（每个 grpc.addrConn 表示客户端到一个服务端的一条连接），每个 grpc.addrConn 也有自己的连接状态。

  • 当至少有一个 grpc.addrConn.state = Ready，则 grpc.ClientConn.csMgr.state = Ready
  • 当至少有一个 grpc.addrConn.state = Connecting，则 grpc.ClientConn.csMgr.state = Connecting
  • 否则 grpc.ClientConn.csMgr.state = TransientFailure
    1. 默认实现下客户端与某一个服务端（host:port）只会建立一条连接，所有 RPC 执行都会复用这条连接。         
    2. 关于为何只建立一条连接可以看下这个 issue：Use multiple connections to avoid the server’s 
    3. SETTINGS_MAX_CONCURRENT_STREAMS limit #11704[1]        
    4. 不过如果使用 manual.Resolver，把同一个服务地址复制多遍，也能做到与一个服务端建立多个连接。
    关于 grpc.addrConn.state 的状态切换可参考设计文档：gRPC Connectivity Semantics and API

2、grpc.ccResolverWrapper 服务端地址解析模块的封装

grpc 内置的 resolver.Resolver 有：

  • dns.dnsResolver：通过域名解析服务地址
  • manual.Resolver：手动设置服务地址
  • passthrough.passthroughResolver：将 grpc.Dial 参数中的地址作为服务地址，这也是默认的

3、grpc.ccBalancerWrapper 负载均衡模块的封装

grpc 内置的 balancer.Balancer 有：

  • grpc.pickfirstBalancer：只使用一个服务地址
  • roundrobin：在多个服务地址中轮转
  • grpclb：使用一个单独的服务提供负载均衡信息（可用的服务地址列表）

可参考设计文档：Load Balancing in gRPC

4、grpc.pickerWrapper 从连接池中选择一个连接

与使用的 balancer.Balancer 具体实现相关：

  • grpc.pickfirstBalancer：grpc.picker，返回当前的连接
  • roundrobin：roundrobin.rrPicker，轮转返回一个可用连接
  • grpclb：grpclb.lbPicker，轮转返回一个可用连接

配置 resolver = dns, balancer = roundrobin 组件之间的关系如下图：
GRPC 客户端服务保障如何做？ - 图1

  • dns.dnsResolver 会启动一个 goroutine，负责解析服务地址，发送更新事件
  • grpc.ccBalancer.Wrapper 会启动一个 goroutine，负责监听服务地址更新事件和连接状态变化事件 – 服务地址更新事件会触发负载均衡组件更新连接池 – 连接状态变化事件会触发负载均衡组件更新连接池中连接的状态，以及更新 picker
  • 当执行 RPC 调用时，会通过 grpc.ClientConn.blockingpicker（即 grpc.pickerWrapper.pick，最终调用 roundrobin.rrPicker.Pick）从连接池中获取连接

代码实现

客户端

  1. package main
  3. import (
  4.     "context"
  5.     "flag"
  6.     "fmt"
  7.     "log"
  8.     "time"
  10.     pb "github.com/yangxikun/go-grpc-client-side-lb-example/pb"
  11.     _ "github.com/yangxikun/go-grpc-client-side-lb-example/resolver/dns"
  12.     "google.golang.org/grpc"
  13.     "google.golang.org/grpc/balancer/roundrobin"
  14.     "google.golang.org/grpc/resolver"
  15. )
  17. const (
  18.     defaultName = "rokety"
  19. )
  21. func main() {
  22.     log.SetFlags(log.Lshortfile | log.Ldate)
  23.     var address string
  24.     var timeout int
  25.     flag.IntVar(&timeout, "timeout", 1, "greet rpc call timeout")
  26.     flag.StringVar(&address, "address", "localhost:50051", "grpc server addr")
  27.     flag.Parse()
  28.     // Set resolver
  29.     resolver.SetDefaultScheme("custom_dns")
  30.     // Set up a connection to the server.
  31.     conn, err := grpc.Dial(address, grpc.WithInsecure(),
  32.         grpc.WithDefaultServiceConfig(
  33.             fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, roundrobin.Name)
  34.         ),
  35.         grpc.WithBlock(), grpc.WithBackoffMaxDelay(time.Second))
  36.     if err != nil {
  37.         log.Fatalf("did not connect: %v", err)
  38.     }
  39.     defer conn.Close()
  40.     c := pb.NewGreeterClient(conn)
  42.     // Contact the server and print out its response.
  43.     for range time.Tick(time.Second) {
  44.         ctx, cancel := context.WithTimeout(context.Background(), time.Duration(timeout)*time.Second)
  45.         r, err := c.SayHello(ctx, &pb.HelloRequest{Name: defaultName})
  46.         if err != nil {
  47.             log.Printf("could not greet: %v\n", err)
  48.         } else {
  49.             log.Printf("Greeting: %s", r.Message)
  50.         }
  51.         cancel()
  52.     }
  53. }

服务端

  1. package main
  3. import (
  4.     "context"
  5.     "log"
  6.     "math/rand"
  7.     "net"
  8.     "time"
  10.     pb "github.com/yangxikun/go-grpc-client-side-lb-example/pb"
  11.     "google.golang.org/grpc"
  12.     "google.golang.org/grpc/reflection"
  13. )
  15. const (
  16.     port = ":50051"
  17. )
  19. var stuckDuration time.Duration
  21. // server is used to implement helloworld.GreeterServer.
  22. type server struct{}
  24. // SayHello implements helloworld.GreeterServer
  25. func (s *server) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloReply, error) {
  26.     time.Sleep(stuckDuration)
  27.     return &pb.HelloReply{Message: "Hello " + in.Name + "! From " + GetIP()}, nil
  28. }
  30. func main() {
  31.     // simulate busy server
  32.     stuckDuration = time.Duration(rand.NewSource(time.Now().UnixNano()).Int63()%2) * time.Second
  34.     if stuckDuration == time.Second {
  35.         log.Println("I will stuck one second!!!")
  36.     }
  38.     lis, err := net.Listen("tcp", port)
  39.     if err != nil {
  40.         log.Fatalf("failed to listen: %v", err)
  41.     }
  42.     s := grpc.NewServer()
  43.     pb.RegisterGreeterServer(s, &server{})
  44.     // Register reflection service on gRPC server.
  45.     reflection.Register(s)
  46.     if err := s.Serve(lis); err != nil {
  47.         log.Fatalf("failed to serve: %v", err)
  48.     }
  49. }
  51. func GetIP() string {
  52.     ifaces, _ := net.Interfaces()
  53.     // handle err
  54.     for _, i := range ifaces {
  55.         addrs, _ := i.Addrs()
  56.         // handle err
  57.         for _, addr := range addrs {
  58.             var ip net.IP
  59.             switch v := addr.(type) {
  60.             case *net.IPNet:
  61.                 ip = v.IP
  62.             case *net.IPAddr:
  63.                 ip = v.IP
  64.             default:
  65.                 continue
  66.             }
  67.             if ip.String() != "127.0.0.1" {
  68.                 return ip.String()
  69.             }
  70.         }
  71.     }
  72.     return ""
  73. }

接下来将借助 k8s 部署客户端和服务端代码，测试代码实际效果。
启动 3 个服务端实例，客户端 timeout 设置为 2 秒，客户端日志情况：

  1. 2019/10/21 main.go:46: Greeting: Hello rokety! From 10.200.235.181
  2. 2019/10/21 main.go:46: Greeting: Hello rokety! From 10.200.251.161
  3. 2019/10/21 main.go:46: Greeting: Hello rokety! From 10.200.105.163
  4. 2019/10/21 main.go:46: Greeting: Hello rokety! From 10.200.235.181
  5. 2019/10/21 main.go:46: Greeting: Hello rokety! From 10.200.251.161
  6. 2019/10/21 main.go:46: Greeting: Hello rokety! From 10.200.105.163
  7. 2019/10/21 main.go:46: Greeting: Hello rokety! From 10.200.235.181
  8. 2019/10/21 main.go:46: Greeting: Hello rokety! From 10.200.251.161
  9. 2019/10/21 main.go:46: Greeting: Hello rokety! From 10.200.105.163

将服务端缩容为 1 个实例后，客户端日志情况：

  1. 2019/10/21 main.go:46: Greeting: Hello rokety! From 10.200.235.181
  2. 2019/10/21 main.go:46: Greeting: Hello rokety! From 10.200.251.161
  3. 2019/10/21 main.go:46: Greeting: Hello rokety! From 10.200.105.163
  4. 2019/10/21 main.go:44: could not greet: rpc error: code = Unavailable desc = transport is closing
  5. 2019/10/21 main.go:46: Greeting: Hello rokety! From 10.200.105.163
  6. 2019/10/21 main.go:46: Greeting: Hello rokety! From 10.200.105.163
  7. 2019/10/21 main.go:46: Greeting: Hello rokety! From 10.200.105.163
  8. 2019/10/21 main.go:46: Greeting: Hello rokety! From 10.200.105.163
  9. 2019/10/21 main.go:46: Greeting: Hello rokety! From 10.200.105.163
  10. 2019/10/21 main.go:46: Greeting: Hello rokety! From 10.200.105.163
  11. 2019/10/21 main.go:46: Greeting: Hello rokety! From 10.200.105.163

可以看到客户端遇到了一次连接被关闭的错误，为了避免缩容和滚动更新导致的此类错误，我们可以在客户端的代码里加上重试机制：
grpc 客户端的重试策略有 2 种实现，具体可参考涉及文档：gRPC Retry Design：

  • Retry policy：出错时立即重试
  • Hedging policy：定时发送并发的多个请求，根据请求的响应情况决定是否发送下一个同样的请求，还是返回（该策略目前未实现）
    1. 注意：
    2. 客户端程序启动时，还需要设置环境变量：GRPC_GO_RETRY=on
    3. MaxAttempts = 2，即最多尝试 2 次，也就是最多重试 1 次
    4. RetryableStatusCodes 只设置了 UNAVAILABLE，也就是解决上面出现的错误：rpc error: code = Unavailable desc = transport is closing
    5. RetryableStatusCodes 中设置 DeadlineExceeded 和 Canceled 是没有作用的，
    6. 因为在重试逻辑的代码里判断到 Context 超时或取消就会立即退出重试逻辑了。

    添加重试逻辑后的客户端代码：

    ```go package main

import ( “context” “flag” “fmt” “log” “time”

  1. pb "github.com/yangxikun/go-grpc-client-side-lb-example/pb"
  2. _ "github.com/yangxikun/go-grpc-client-side-lb-example/resolver/dns"
  3. "google.golang.org/grpc"
  4. "google.golang.org/grpc/balancer/roundrobin"
  5. "google.golang.org/grpc/resolver"

)

const ( defaultName = “rokety” )

func main() { log.SetFlags(log.Lshortfile | log.Ldate | log.Ltime) var address string var timeout int flag.IntVar(&timeout, “timeout”, 1, “greet rpc call timeout”) flag.StringVar(&address, “address”, “localhost:50051”, “grpc server addr”) flag.Parse() // Set up a connection to the server. resolver.SetDefaultScheme(“custom_dns”) conn, err := grpc.Dial(address, grpc.WithInsecure(), grpc.WithDefaultServiceConfig( fmt.Sprintf( {"LoadBalancingPolicy": "%s", "MethodConfig":[{ "Name": [{"Service": "helloworld.Greeter"}], "RetryPolicy":{ "MaxAttempts":2, "InitialBackoff": "0.1s", "MaxBackoff": "1s", "BackoffMultiplier": 2.0, "RetryableStatusCodes": ["UNAVAILABLE"] } }] },roundrobin.Name)), grpc.WithBlock(), grpc.WithBackoffMaxDelay(time.Second)) if err != nil { log.Fatalf(“did not connect: %v”, err) } defer conn.Close() c := pb.NewGreeterClient(conn)

  1. // Contact the server and print out its response.
  2. for range time.Tick(time.Second) {
  3.     ctx, cancel := context.WithTimeout(context.Background(), time.Duration(timeout)*time.Second)
  4.     r, err := c.SayHello(ctx, &pb.HelloRequest{Name: defaultName})
  5.     if err != nil {
  6.         log.Printf("could not greet: %v\n", err)
  7.     } else {
  8.         log.Printf("Greeting: %s", r.Message)
  9.     }
  10.     cancel()
  11. }

}

  1. 服务端实例恢复为 3 个，修改客户端 timeout 设置为 1 秒后，客户端日志情况：
  2. ```go
  3. 2019/10/21 13:02:54 main.go:46: Greeting: Hello rokety! From 10.200.38.220
  4. 2019/10/21 13:02:56 main.go:44: could not greet: rpc error: code = DeadlineExceeded desc = context deadline exceeded
  5. 2019/10/21 13:02:57 main.go:44: could not greet: rpc error: code = DeadlineExceeded desc = context deadline exceeded
  6. 2019/10/21 13:02:57 main.go:46: Greeting: Hello rokety! From 10.200.38.220
  7. 2019/10/21 13:02:59 main.go:44: could not greet: rpc error: code = DeadlineExceeded desc = context deadline exceeded
  8. 2019/10/21 13:03:00 main.go:44: could not greet: rpc error: code = DeadlineExceeded desc = context deadline exceeded

可以看到有 2 个服务端的响应是一直超时的，但实际业务使用中，希望避免这种错误，这时可以使用 grpc 的健康检查功能（设计文档：GRPC Health Checking Protocol），该功能要求服务端实现健康检查接口，客户端和服务端的代码都需要调整：

添加健康检查

客户端
注意：
需要在 grpc.WithDefaultServiceConfig 中配置 HealthCheckConfig
需要导入_ “google.golang.org/grpc/health”

  1. package main
  3. import (
  4.     "context"
  5.     "flag"
  6.     "fmt"
  7.     "log"
  8.     "time"
  10.     pb "github.com/yangxikun/go-grpc-client-side-lb-example/pb"
  11.     _ "github.com/yangxikun/go-grpc-client-side-lb-example/resolver/dns"
  12.     "google.golang.org/grpc"
  13.     "google.golang.org/grpc/balancer/roundrobin"
  14.     _ "google.golang.org/grpc/health"
  15.     "google.golang.org/grpc/resolver"
  16. )
  18. const (
  19.     defaultName = "rokety"
  20. )
  22. func main() {
  23.     log.SetFlags(log.Lshortfile | log.Ldate | log.Ltime)
  24.     var address string
  25.     var timeout int
  26.     flag.IntVar(&timeout, "timeout", 1, "greet rpc call timeout")
  27.     flag.StringVar(&address, "address", "localhost:50051", "grpc server addr")
  28.     flag.Parse()
  29.     // Set up a connection to the server.
  30.     resolver.SetDefaultScheme("custom_dns")
  31.     conn, err := grpc.Dial(address, grpc.WithInsecure(),
  32.         grpc.WithDefaultServiceConfig(
  33.         fmt.Sprintf(
  34.         `{"LoadBalancingPolicy": "%s",
  35.                "MethodConfig":[{
  36.                 "Name": [{"Service": "helloworld.Greeter"}], 
  37.                 "RetryPolicy":{"MaxAttempts":2, 
  38.                             "InitialBackoff": "0.1s", 
  39.                             "MaxBackoff": "1s", 
  40.                             "BackoffMultiplier": 2.0, 
  41.                             "RetryableStatusCodes": ["UNAVAILABLE", "CANCELLED"]}
  42.             }],
  43.           "HealthCheckConfig": {"ServiceName": "helloworld.Greeter"}
  44.           }`, roundrobin.Name),
  45.         grpc.WithBlock(), grpc.WithBackoffMaxDelay(time.Second))
  46.     if err != nil {
  47.         log.Fatalf("did not connect: %v", err)
  48.     }
  49.     defer conn.Close()
  50.     c := pb.NewGreeterClient(conn)
  52.     // Contact the server and print out its response.
  53.     for range time.Tick(time.Second) {
  54.         ctx, cancel := context.WithTimeout(context.Background(), time.Duration(timeout)*time.Second)
  55.         r, err := c.SayHello(ctx, &pb.HelloRequest{Name: defaultName})
  56.         if err != nil {
  57.             log.Printf("could not greet: %v\n", err)
  58.         } else {
  59.             log.Printf("Greeting: %s", r.Message)
  60.         }
  61.         cancel()
  62.     }
  63. }

服务端
注意：
实现 grpc_health_v1 的接口
注册到服务中：grpc_health_v1.RegisterHealthServer(s, &healthServer{})

  1. package main
  3. import (
  4.     "context"
  5.     "log"
  6.     "math/rand"
  7.     "net"
  8.     "time"
  10.     pb "github.com/yangxikun/go-grpc-client-side-lb-example/pb"
  11.     "google.golang.org/grpc"
  12.     "google.golang.org/grpc/codes"
  13.     "google.golang.org/grpc/health/grpc_health_v1"
  14.     "google.golang.org/grpc/reflection"
  15.     "google.golang.org/grpc/status"
  16. )
  18. const (
  19.     port = ":50051"
  20. )
  22. var stuckDuration time.Duration
  24. type healthServer struct{}
  26. func (h *healthServer) Check(ctx context.Context, req *grpc_health_v1.HealthCheckRequest) (*grpc_health_v1.HealthCheckResponse, error) {
  27.     log.Println("recv health check for service:", req.Service)
  28.     if stuckDuration == time.Second {
  29.         return &grpc_health_v1.HealthCheckResponse{Status: grpc_health_v1.HealthCheckResponse_NOT_SERVING}, nil
  30.     }
  31.     return &grpc_health_v1.HealthCheckResponse{Status: grpc_health_v1.HealthCheckResponse_SERVING}, nil
  32. }
  34. func (h *healthServer) Watch(req *grpc_health_v1.HealthCheckRequest, stream grpc_health_v1.Health_WatchServer) error {
  35.     log.Println("recv health watch for service:", req.Service)
  36.     resp := new(grpc_health_v1.HealthCheckResponse)
  37.     if stuckDuration == time.Second {
  38.         resp.Status = grpc_health_v1.HealthCheckResponse_NOT_SERVING
  39.     } else {
  40.         resp.Status = grpc_health_v1.HealthCheckResponse_SERVING
  41.     }
  42.     for range time.NewTicker(time.Second).C {
  43.         err := stream.Send(resp)
  44.         if err != nil {
  45.             return status.Error(codes.Canceled, "Stream has ended.")
  46.         }
  47.     }
  48.     return nil
  49. }
  51. // server is used to implement helloworld.GreeterServer.
  52. type server struct{}
  54. // SayHello implements helloworld.GreeterServer
  55. func (s *server) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloReply, error) {
  56.     time.Sleep(stuckDuration)
  57.     return &pb.HelloReply{Message: "Hello " + in.Name + "! From " + GetIP()}, nil
  58. }
  60. func main() {
  61.     // simulate busy server
  62.     stuckDuration = time.Duration(rand.NewSource(time.Now().UnixNano()).Int63()%2) * time.Second
  64.     if stuckDuration == time.Second {
  65.         log.Println("I will stuck one second!!!")
  66.     }
  68.     lis, err := net.Listen("tcp", port)
  69.     if err != nil {
  70.         log.Fatalf("failed to listen: %v", err)
  71.     }
  72.     s := grpc.NewServer()
  73.     pb.RegisterGreeterServer(s, &server{})
  74.     grpc_health_v1.RegisterHealthServer(s, &healthServer{})
  75.     // Register reflection service on gRPC server.
  76.     reflection.Register(s)
  77.     if err := s.Serve(lis); err != nil {
  78.         log.Fatalf("failed to serve: %v", err)
  79.     }
  80. }
  82. func GetIP() string {
  83.     ifaces, _ := net.Interfaces()
  84.     // handle err
  85.     for _, i := range ifaces {
  86.         addrs, _ := i.Addrs()
  87.         // handle err
  88.         for _, addr := range addrs {
  89.             var ip net.IP
  90.             switch v := addr.(type) {
  91.             case *net.IPNet:
  92.                 ip = v.IP
  93.             case *net.IPAddr:
  94.                 ip = v.IP
  95.             default:
  96.                 continue
  97.             }
  98.             if ip.String() != "127.0.0.1" {
  99.                 return ip.String()
  100.             }
  101.         }
  102.     }
  103.     return ""
  104. }

更新服务镜像，查看客户端日志：

  1. 2019/10/21 14:25:01 main.go:47: Greeting: Hello rokety! From 10.200.251.170
  2. 2019/10/21 14:25:02 main.go:47: Greeting: Hello rokety! From 10.200.177.145
  3. 2019/10/21 14:25:03 main.go:47: Greeting: Hello rokety! From 10.200.251.170
  4. 2019/10/21 14:25:04 main.go:47: Greeting: Hello rokety! From 10.200.177.145

3 个服务端实例，只向其中 2 个发送了请求，通过查看 3 个服务端日志，确认其中有一个会在健康检查接口中返回 HealthCheckResponse_NOT_SERVING。

参考：go语言中文网站长 https://mp.weixin.qq.com/s?__biz=MzAxMTA4Njc0OQ==&mid=2651437638&idx=1&sn=89d63780012cb505ab5350aed9180ca4&chksm=80bb64b4b7cceda29da232ad4ce81823ff72bd053f226254a1bdb5a6f8488925cc6c59b4c9a1