client.go 代码：

func main() {
 flag.Parse()
 // Set up a connection to the server.
 conn, err := grpc.Dial(*addr, grpc.WithTransportCredentials(insecure.NewCredentials()))
 if err != nil {
  log.Fatalf("did not connect: %v", err)
 }
 defer conn.Close()
 c := pb.NewGreeterClient(conn)

 // Contact the server and print out its response.
 ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 defer cancel()
 r, err := c.SayHello(ctx, &pb.HelloRequest{Name: *name})
 if err != nil {
  log.Fatalf("could not greet: %v", err)
 }
 log.Printf("Greeting: %s", r.GetMessage())
}

Dial 源码：

func Dial(target string, opts ...DialOption) (*ClientConn, error) {
 return DialContext(context.Background(), target, opts...)
}

DialContext 源码：

省略次部分代码

// 首先会创建 ClientConn，初始化相关字段
func DialContext(ctx context.Context, target string, opts ...DialOption) (conn *ClientConn, err error) {
 cc := &ClientConn{
  target:            target,
  csMgr:             &connectivityStateManager{},
  conns:             make(map[*addrConn]struct{}),
  dopts:             defaultDialOptions(),
  blockingpicker:    newPickerWrapper(),
  czData:            new(channelzData),
  firstResolveEvent: grpcsync.NewEvent(),
 }

 // 将用户设置的连接参数更新到客户端连接器 ClientConn
 for _, opt := range opts {
  opt.apply(&cc.dopts)
 }

 return cc, nil
}

connect() 方法：

func (ac *addrConn) connect() error {
 ac.mu.Lock()
 // if 校验状态
 if ac.state == connectivity.Shutdown {
  ac.mu.Unlock()
  return errConnClosing
 }
 if ac.state != connectivity.Idle {
  ac.mu.Unlock()
  return nil
 }
 ac.updateConnectivityState(connectivity.Connecting, nil)
 ac.mu.Unlock()
 // 主要看这个方法，重试连接
 ac.resetTransport()
 return nil
}

进入 resetTransport() 源码

func (ac *addrConn) resetTransport() {
 ac.mu.Lock()
 // 判断状态若为 shutdown，则不再连接直接推出
 if ac.state == connectivity.Shutdown {
  ac.mu.Unlock()
  return
 }

 addrs := ac.addrs
 // 连接失败，需要进行重试的
 // Backoff 是需要等待的时间，ac.backoffIdx表示第几次重试
 backoffFor := ac.dopts.bs.Backoff(ac.backoffIdx)
 // 计算出本次向gRPC服务尝试建立TCP连接的最长时间
 // 若超过这个时间还是连接不上，则主动断开，等待尝试下次连接
 // This will be the duration that dial gets to finish.
 dialDuration := minConnectTimeout
 if ac.dopts.minConnectTimeout != nil {
  dialDuration = ac.dopts.minConnectTimeout()
 }

 if dialDuration < backoffFor {
  // Give dial more time as we keep failing to connect.
  dialDuration = backoffFor
 }
 connectDeadline := time.Now().Add(dialDuration)

 // 更新结构addrConn状态为connecting
 ac.updateConnectivityState(connectivity.Connecting, nil)
 ac.mu.Unlock()

 // 向服务器连接失败后需要做的逻辑
 if err := ac.tryAllAddrs(addrs, connectDeadline); err != nil {
  ac.cc.resolveNow(resolver.ResolveNowOptions{})
  // After exhausting all addresses, the addrConn enters
  // TRANSIENT_FAILURE.
  ac.mu.Lock()
  if ac.state == connectivity.Shutdown {
   ac.mu.Unlock()
   return
  }
  ac.updateConnectivityState(connectivity.TransientFailure, err)

  // Backoff.
  b := ac.resetBackoff
  ac.mu.Unlock()

  // 定时的超时间
  timer := time.NewTimer(backoffFor)
  // 1.timer.C如果连接超时，重试的次数+1，继续下一次重试连接，但需要等待一段时间
  // 2. b,直接关闭，将继续进行重新连接
  // 2. ac.ctx.Done，走这里的话，上下文结束，这里不会再次重试了
  select {
  case <-timer.C:
   ac.mu.Lock()
   ac.backoffIdx++
   ac.mu.Unlock()
  case <-b:
   timer.Stop()
  case <-ac.ctx.Done():
   timer.Stop()
   return
  }

  ac.mu.Lock()
  // 状态 != shutdown就更新为空闲状态
  if ac.state != connectivity.Shutdown {
   ac.updateConnectivityState(connectivity.Idle, err)
  }
  ac.mu.Unlock()
  return
 }
 // 连接成功，重新设置backoff为原始值0
 ac.mu.Lock()
 ac.backoffIdx = 0
 ac.mu.Unlock()
}

如何计算重试连接等待时间？

进入Backoff方法

func (bc Exponential) Backoff(retries int) time.Duration {
 if retries == 0 {
  return bc.Config.BaseDelay
 }
 backoff, max := float64(bc.Config.BaseDelay), float64(bc.Config.MaxDelay)
 for backoff < max && retries > 0 {
  // 幂次方
  backoff *= bc.Config.Multiplier
  retries--
 }
 // 不能超过最大延时时间
 if backoff > max {
  backoff = max
 }
 // Randomize backoff delays so that if a cluster of requests start at
 // the same time, they won't operate in lockstep.
 backoff *= 1 + bc.Config.Jitter*(grpcrand.Float64()*2-1)
 if backoff < 0 {
  return 0
 }
 return time.Duration(backoff)
}

连接失败后，客户端会进行重试连接。 重试次数越多，等待下一次连接时间也会变长，但不能超过MaxDelay值。

更多Go云原生学习资料，收录于Github： https://github.com/metashops/GoFamily

本文由 mdnice 多平台发布

grpc 版本1.50client.go 代码：func main() { flag.Parse() // Set up a connection to the server. conn, err := grpc.Dial(*addr, grpc.WithTransportCredentials(insecure.NewCredentials. 项目中碰到了 grpc 双向流和单向流出现了 服务端 流context报错：context cancel error，而 客户端 无感知 连接 断开，依然认为还在 连接 中，导致数据传输中断。 遇到这种情况从以下几个维度去思考问题发生的可能： 网络 是否 稳定，能否复现 客户端 和 服务端 是否 有对context进行cancel 客户端 和 服务端 是否 有去处理context的cancel信号 由于出现的问题不是在开发环境中，因此开始只能看日志进行排查，但是日志中只得到了 服务端 会 收到context的cancel信号，而 客户端 没有任何 我们以发送数据的场景，来介绍 重试 机制 。 1、 grpc 框架中的 重试 机制 的核心或者说实现主体思路？ 假设 客户端 向服务器端发起服务请求，整个过程 会 经历很多阶段，并不是所有的阶段都能 重试 ，只有某些阶段可以进行 重试 ； 当某个操作执行成功后，将该操作缓存到一个切片里，假设已经将2个阶段的操作缓存到了切片里； 执行第3个操作时，假设执行 失败 ； 那么此时，首先判断 是否 允许 重试 ： 若允许 重试 的话，重新依次执行切片里的函数，最后在 重试 一下第3个操作即可； 2、 grpc 框架 重试 机制