grpc 版本1.50

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 源碼：

省略次部分代碼

// 首先會創(chuàng)建 ClientConn檩小，初始化相關(guān)字段
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(),
    }

    // 將用戶設(shè)置的連接參數(shù)更新到客戶端連接器 ClientConn
    for _, opt := range opts {
        opt.apply(&cc.dopts)
    }

    return cc, nil
}

connect() 方法：

func (ac *addrConn) connect() error {
    ac.mu.Lock()
    // if 校驗狀態(tài)
    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
}

進(jìn)入 resetTransport() 源碼

func (ac *addrConn) resetTransport() {
    ac.mu.Lock()
    // 判斷狀態(tài)若為 shutdown妆够，則不再連接直接推出
    if ac.state == connectivity.Shutdown {
        ac.mu.Unlock()
        return
    }

    addrs := ac.addrs
    // 連接失敗负蚊，需要進(jìn)行重試的
    // Backoff 是需要等待的時間姨伤，ac.backoffIdx表示第幾次重試
    backoffFor := ac.dopts.bs.Backoff(ac.backoffIdx)
    // 計算出本次向gRPC服務(wù)嘗試建立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)

    // 更新結(jié)構(gòu)addrConn狀態(tài)為connecting
    ac.updateConnectivityState(connectivity.Connecting, nil)
    ac.mu.Unlock()

    // 向服務(wù)器連接失敗后需要做的邏輯
    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如果連接超時套耕，重試的次數(shù)+1，繼續(xù)下一次重試連接肴捉，但需要等待一段時間
        // 2. b,直接關(guān)閉，將繼續(xù)進(jìn)行重新連接
        // 2. ac.ctx.Done芝薇，走這里的話，上下文結(jié)束绣夺，這里不會再次重試了
        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()
        // 狀態(tài) != shutdown就更新為空閑狀態(tài)
        if ac.state != connectivity.Shutdown {
            ac.updateConnectivityState(connectivity.Idle, err)
        }
        ac.mu.Unlock()
        return
    }
    // 連接成功，重新設(shè)置backoff為原始值0
    ac.mu.Lock()
    ac.backoffIdx = 0
    ac.mu.Unlock()
}

如何計算重試連接等待時間佃扼？

進(jìn)入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)
}

總結(jié)：

• 連接失敗后郁季，客戶端會進(jìn)行重試連接冗恨。
• 重試次數(shù)越多，等待下一次連接時間也會變長尿招，但不能超過MaxDelay值。

更多Go云原生學(xué)習(xí)資料咖城，收錄于Github：https://github.com/metashops/GoFamily

本文由mdnice多平臺發(fā)布