与 gRPC 客户端重新连接的正确方法

大家好，今天本人给大家带来文章《与 gRPC 客户端重新连接的正确方法》，文中内容主要涉及到，如果你对Golang方面的知识点感兴趣，那就请各位朋友继续看下去吧~希望能真正帮到你们，谢谢！

我有一个 go grpc 客户端连接到在 k8s 集群中的不同 pod 中运行的 grpc 服务器。

它运行良好，可以接收和处理请求。

我现在想知道在 grpc 服务器 pod 被回收的情况下如何最好地实现弹性。

据我所知，clientconn.go 代码应该自动处理重新连接，但我就是无法让它工作，我担心我的实现在第一个实例中是不正确的。

从 main 调用代码：

go func() {     
        if err := grpcclient.processrequests(); err != nil {
            log.error("error while processing requests")
            //do something here??
        }
    }()

我在 grpcclient 包装器模块中的代码：

func (grpcclient *grpcclient) processrequests() error {
    defer grpcclient.close()    

    for {
        request, err := reqclient.stream.recv()
        log.info("request received")
        if err == io.eof {          
            break
        }
        if err != nil {
            //when pod is recycled, this is what's hit with err:
            //rpc error: code = unavailable desc = transport is closing"

            //what is the correct pattern for recovery here so that we can await connection
            //and continue processing requests once more?
            //should i return err here and somehow restart the processrequests() go routine in the 
            //main funcition?
            break
            
        } else {
            //the happy path
            //code block to process any requests that are received
        }
    }

    return nil
}

func (reqclient *requestclient) close() {
//this is called soon after the conneciton drops
        reqclient.conn.close()
}

编辑： 艾敏·拉莱托维奇（emin laletovic）在下面优雅地回答了我的问题，并且大部分内容都得到了解答。 我必须对 waituntilready 函数进行一些更改：

func (grpcclient *gRPCClient) waitUntilReady() bool {
ctx, cancel := context.WithTimeout(context.Background(), 300*time.Second) //define how long you want to wait for connection to be restored before giving up
defer cancel()

currentState := grpcclient.conn.GetState()
stillConnecting := true

for currentState != connectivity.Ready && stillConnecting {
    //will return true when state has changed from thisState, false if timeout
    stillConnecting = grpcclient.conn.WaitForStateChange(ctx, currentState)
    currentState = grpcclient.conn.GetState()
    log.WithFields(log.Fields{"state: ": currentState, "timeout": timeoutDuration}).Info("Attempting reconnection. State has changed to:")
}

if stillConnecting == false {
    log.Error("Connection attempt has timed out.")
    return false
}

return true
}

解决方案

rpc 连接由 clientconn.go 自动处理，但这并不意味着流也会自动处理。

流一旦断开，无论是由于 rpc 连接中断还是其他原因，都无法自动重新连接，一旦 rpc 连接恢复，您需要从服务器获取新的流。

等待 rpc 连接处于 ready 状态并建立新流的伪代码可能如下所示：

func (grpcclient *grpcclient) processrequests() error {
    defer grpcclient.close()    
    
    go grpcclient.process()
    for {
      select {
        case <- grpcclient.reconnect:
           if !grpcclient.waituntilready() {
             return errors.new("failed to establish a connection within the defined timeout")
           }
           go grpcclient.process()
        case <- grpcclient.done:
          return nil
      }
    }
}

func (grpcclient *grpcclient) process() {
    reqclient := getstream() //always get a new stream
    for {
        request, err := reqclient.stream.recv()
        log.info("request received")
        if err == io.eof {          
            grpcclient.done <- true
            return
        }
        if err != nil {
            grpcclient.reconnect <- true
            return
            
        } else {
            //the happy path
            //code block to process any requests that are received
        }
    }
}

func (grpcclient *grpcclient) waituntilready() bool {
  ctx, cancel := context.withtimeout(context.background(), 60*time.second) //define how long you want to wait for connection to be restored before giving up
  defer cancel()
  return grpcclient.conn.waitforstatechange(ctx, conectivity.ready)
}

编辑：

重新审视上面的代码，应该纠正一些错误。 waitforstatechange 函数等待连接状态从传递状态更改，它不等待连接更改为传递状态。

最好跟踪当前连接状态，如果通道空闲，则使用 connect 函数进行连接。

func (grpcclient *grpcclient) processrequests() error {
        defer grpcclient.close()    
        
        go grpcclient.process()
        for {
          select {
            case <- grpcclient.reconnect:
               if !grpcclient.isreconnected(1*time.second, 60*time.second) {
                 return errors.new("failed to establish a connection within the defined timeout")
               }
               go grpcclient.process()
            case <- grpcclient.done:
              return nil
          }
        }
}

func (grpcclient *grpcclient) isreconnected(check, timeout time.duration) bool {
  ctx, cancel := context.context.withtimeout(context.background(), timeout)
  defer cancel()
  ticker := time.newticker(check)

  for{
    select {
      case <- ticker.c:
        grpcclient.conn.connect()
 
        if grpcclient.conn.getstate() == connectivity.ready {
          return true
        }
      case <- ctx.done():
         return false
    }
  }
}

当grpc连接关闭时，grpc客户端连接的状态将为 idle 或 transient_failure。以下是我的 grpc 双向流式传输自定义重新连接机制的示例。首先，我有一个 for 循环来保持重新连接，直到 grpc 服务器启动，在调用 conn.connect() 后状态将变为就绪状态。

for {
    select {
    case <-ctx.done():
        return false
    default:
            if client.conn.getstate() != connectivity.ready {
                client.conn.connect()
            }

            // reserve a short duration (customizable) for conn to change state from idle to ready if grpc server is up
            time.sleep(500 * time.millisecond)

            if client.conn.getstate() == connectivity.ready {
                return true
            }

            // define reconnect time interval (backoff) or/and reconnect attempts here
            time.sleep(2 * time.second)
    }
}

此外，还将生成一个 goroutine 以执行重新连接任务。成功重连后，会生成另一个goroutine来监听grpc服务器。

for {
    select {
    case <-ctx.done():
        return
    case <-reconnectch:
        if client.conn.getstate() != connectivity.ready && *isconnectedwebsocket {
            if o.waituntilready(client, isconnectedwebsocket, ctx) {
                err := o.generatenewprocessorderstream(client, ctx)
                if err != nil {
                    logger.logger.error("failed to establish stream connection to grpc server ...")
                }

                // re-listening server side streaming
                go o.listenprocessorderserverside(client, reconnectch, ctx, isconnectedwebsocket)
            }
        }
    }
}

请注意，监听任务是由另一个 goroutine 并发处理的。

// listening server side streaming
go o.listenProcessOrderServerSide(client, reconnectCh, websocketCtx, isConnectedWebSocket)

您可以查看我的代码示例 here。希望这会有所帮助。

图片来源：艾敏·拉莱托维奇

好了，本文到此结束，带大家了解了《与 gRPC 客户端重新连接的正确方法》，希望本文对你有所帮助！关注golang学习网公众号，给大家分享更多Golang知识！