技術文章 » 負載均衡原理分析與原始碼解讀

負載均衡原理分析與原始碼解讀

2022-08-10 12:06:05

上一篇文章一起學習了Resolver的原理和原始碼分析，本篇繼續和大家一起學習下和Resolver關係密切的Balancer的相關內容。這裡說的負載均衡主要指資料中心內的負載均衡，即RPC間的負載均衡。

傳送門 服務發現原理分析與原始碼解讀

基於go-zero v1.3.5 和 grpc-go v1.47.0

負載均衡

每一個被呼叫服務都會有多個範例，那麼服務的呼叫方應該將請求，發向被呼叫服務的哪一個服務範例，這就是負載均衡的業務場景。

負載均衡的第一個關鍵點是公平性，即負載均衡需要關注被呼叫服務範例組之間的公平性，不要出現旱的旱死，澇的澇死的情況。

負載均衡的第二個關鍵點是正確性，即對於有狀態的服務來說，負載均衡需要關心請求的狀態，將請求排程到能處理它的後端範例上，不要出現不能處理和錯誤處理的情況。

無狀態的負載均衡

無狀態的負載均衡是我們日常工作中接觸比較多的負載均衡模型，它指的是參與負載均衡的後端範例是無狀態的，所有的後端範例都是對等的，一個請求不論發向哪一個範例，都會得到相同的並且正確的處理結果，所以無狀態的負載均衡策略不需要關心請求的狀態。下面介紹兩種無狀態負載均衡演演算法。

輪詢

輪詢的負載均衡策略非常簡單，只需要將請求按順序分配給多個範例，不用再做其他的處理。例如，輪詢策略會將第一個請求分配給第一個範例，然後將下一個請求分配給第二個範例，這樣依次分配下去，分配完一輪之後，再回到開頭分配給第一個範例，再依次分配。輪詢在路由時，不利用請求的狀態資訊，屬於無狀態的負載均衡策略，所以它不能用於有狀態範例的負載均衡器，否則正確性會出現問題。在公平性方面，因為輪詢策略只是按順序分配請求，所以適用於請求的工作負載和範例的處理能力差異都較小的情況。

權重輪詢

權重輪詢的負載均衡策略是將每一個後端範例分配一個權重，分配請求的數量和範例的權重成正比輪詢。例如有兩個範例 A，B，假設我們設定 A 的權重為 20，B 的權重為 80，那麼負載均衡會將 20% 的請求數量分配給 A，80 % 的請求數量分配給 B。權重輪詢在路由時，不利用請求的狀態資訊，屬於無狀態的負載均衡策略，所以它也不能用於有狀態範例的負載均衡器，否則正確性會出現問題。在公平性方面，因為權重策略會按範例的權重比例來分配請求數，所以，我們可以利用它解決範例的處理能力差異的問題，認為它的公平性比輪詢策略要好。

有狀態負載均衡

有狀態負載均衡是指，在負載均衡策略中會儲存伺服器端的一些狀態，然後根據這些狀態按照一定的演演算法選擇出對應的範例。

P2C+EWMA

在go-zero中預設使用的是P2C的負載均衡演演算法。該演演算法的原理比較簡單，即隨機從所有可用節點中選擇兩個節點，然後計算這兩個節點的負載情況，選擇負載較低的一個節點來服務本次請求。為了避免某些節點一直得不到選擇導致不平衡，會在超過一定的時間後強制選擇一次。

在該複雜均衡演演算法中，多出採用了EWMA指數移動加權平均的演演算法，表示是一段時間內的均值。該演演算法相對於算數平均來說對於突然的網路抖動沒有那麼敏感，突然的抖動不會體現在請求的lag中，從而可以讓演演算法更加均衡。

go-zero/zrpc/internal/balancer/p2c/p2c.go:133

atomic.StoreUint64(&c.lag, uint64(float64(olag)*w+float64(lag)*(1-w)))

go-zero/zrpc/internal/balancer/p2c/p2c.go:139

atomic.StoreUint64(&c.success, uint64(float64(osucc)*w+float64(success)*(1-w)))

係數w是一個時間衰減值，即兩次請求的間隔越大，則係數w就越小。

go-zero/zrpc/internal/balancer/p2c/p2c.go:124

w := math.Exp(float64(-td) / float64(decayTime))

節點的load值是通過該連線的請求延遲 lag 和當前請求數 inflight 的乘積所得，如果請求的延遲越大或者當前正在處理的請求數越多表明該節點的負載越高。

go-zero/zrpc/internal/balancer/p2c/p2c.go:199

func (c *subConn) load() int64 {
  // plus one to avoid multiply zero
  lag := int64(math.Sqrt(float64(atomic.LoadUint64(&c.lag) + 1)))
  load := lag * (atomic.LoadInt64(&c.inflight) + 1)
  if load == 0 {
    return penalty
  }

  return load
}

原始碼分析

如下原始碼會涉及go-zero和gRPC，請根據給出的程式碼路徑進行區分

在gRPC中，Balancer和Resolver一樣也可以自定義，同樣也是通過Register方法進行註冊

grpc-go/balancer/balancer.go:53

func Register(b Builder) {
  m[strings.ToLower(b.Name())] = b
}

Register的引數Builder為介面，在Builder介面中，Build方法的第一個引數ClientConn也為介面，Build方法的返回值Balancer同樣也是介面，定義如下：

可以看出，要想實現自定義的Balancer的話，就必須要實現balancer.Builder介面。

在瞭解了gRPC提供的Balancer的註冊方式之後，我們看一下go-zero是在什麼地方進行Balancer註冊的

go-zero/zrpc/internal/balancer/p2c/p2c.go:36

func init() {
  balancer.Register(newBuilder())
}

在go-zero中並沒有實現 balancer.Builder 介面，而是使用gRPC提供的 base.baseBuilder 進行註冊，base.baseBuilder 實現了balancer.Builder 介面。建立baseBuilder的時候呼叫了 base.NewBalancerBuilder 方法，需要傳入 PickerBuilder 引數，PickerBuilder為介面，在go-zero中 p2c.p2cPickerBuilder 實現了該介面。

PickerBuilder介面Build方法返回值 balancer.Picker 也是一個介面，p2c.p2cPicker 實現了該介面。

grpc-go/balancer/base/base.go:65

func NewBalancerBuilder(name string, pb PickerBuilder, config Config) balancer.Builder {
  return &baseBuilder{
    name:          name,
    pickerBuilder: pb,
    config:        config,
  }
}

各結構之間的關係如下圖所示，其中各結構模組對應的包為：

  • balancer：grpc-go/balancer
  • base：grpc-go/balancer/base
  • p2c: go-zero/zrpc/internal/balancer/p2c

在哪裡獲取已註冊的Balancer？

通過上面的流程步驟，已經知道了如何自定義Balancer，以及如何註冊自定義的Blancer。既然註冊了肯定就會獲取，接下來看一下是在哪裡獲取已經註冊的Balancer的。

我們知道Resolver是通過解析DialContext的第二個引數target，從而得到Resolver的name，然後根據name獲取到對應的Resolver的。獲取Balancer同樣也是根據名稱，Balancer的名稱是在建立gRPC Client的時候通過設定項傳入的，這裡的p2c.Name為註冊Balancer時指定的名稱 p2c_ewma ，如下：

go-zero/zrpc/internal/client.go:50

func NewClient(target string, opts ...ClientOption) (Client, error) {
  var cli client

  svcCfg := fmt.Sprintf(`{"loadBalancingPolicy":"%s"}`, p2c.Name)
  balancerOpt := WithDialOption(grpc.WithDefaultServiceConfig(svcCfg))
  opts = append([]ClientOption{balancerOpt}, opts...)
  if err := cli.dial(target, opts...); err != nil {
    return nil, err
  }

  return &cli, nil
}

在上一篇文章中，我們已經知道當建立gRPC使用者端的時候，會觸發呼叫自定義Resolver的Build方法，在Build方法內部獲取到服務地址列表後，通過cc.UpdateState方法進行狀態更新，後面當監聽到服務狀態變化的時候同樣也會呼叫cc.UpdateState進行狀態的更新，而這裡的cc指的就是 ccResolverWrapper 物件，這一部分如果忘記的話，可以再去回顧一下講解Resolver的那篇文章，以便能絲滑接入本篇：

go-zero/zrpc/resolver/internal/kubebuilder.go:51

if err := cc.UpdateState(resolver.State{
  Addresses: addrs,
}); err != nil {
  logx.Error(err)
}

這裡有幾個重要的模組物件，如下：

  • ClientConn：grpc-go/clientconn.go:464
  • ccResolverWrapper：grpc-go/resolver_conn_wrapper.go:36
  • ccBalancerWrapper：grpc-go/balancer_conn_wrappers.go:48
  • Balancer：grpc-go/internal/balancer/gracefulswitch/gracefulswitch.go:46
  • balancerWrapper：grpc-go/internal/balancer/gracefulswitch/gracefulswitch.go:247

當監聽到服務狀態的變更後（首次啟動或者通過Watch監聽變化）呼叫 ccResolverWrapper.UpdateState 觸發更新狀態的流程，各模組間的呼叫鏈路如下所示：

獲取Balancer的動作是在 ccBalancerWrapper.handleSwitchTo 方法中觸發的，程式碼如下所示：

grpc-go/balancer_conn_wrappers.go:266

builder := balancer.Get(name)
if builder == nil {
  channelz.Warningf(logger, ccb.cc.channelzID, "Channel switches to new LB policy %q, since the specified LB policy %q was not registered", PickFirstBalancerName, name)
  builder = newPickfirstBuilder()
} else {
  channelz.Infof(logger, ccb.cc.channelzID, "Channel switches to new LB policy %q", name)
}

if err := ccb.balancer.SwitchTo(builder); err != nil {
  channelz.Errorf(logger, ccb.cc.channelzID, "Channel failed to build new LB policy %q: %v", name, err)
  return
}
ccb.curBalancerName = builder.Name()

然後在 Balancer.SwitchTo 方法中，呼叫了自定義Balancer的Build方法：

grpc-go/internal/balancer/gracefulswitch/gracefulswitch.go:121

newBalancer := builder.Build(bw, gsb.bOpts)

上文有提到Build方法的第一個引數為介面 balancer.ClientConn ，而這裡傳入的為 balancerWrapper ，所以gracefulswitch.balancerWrapper實現了該介面：

到這裡我們已經知道了獲取自定義Balancer是在哪裡觸達的，以及在哪裡獲取的自定義的Balancer，和balancer.Builder的Build方法在哪裡被呼叫。

通過上文可知這裡的balancer.Builder為baseBuilder，所以呼叫的Build方法為baseBuilder的Build方法，Build方法的定義如下：

grpc-go/balancer/base/balancer.go:39

func (bb *baseBuilder) Build(cc balancer.ClientConn, opt balancer.BuildOptions) balancer.Balancer {
  bal := &baseBalancer{
    cc:            cc,
    pickerBuilder: bb.pickerBuilder,

    subConns: resolver.NewAddressMap(),
    scStates: make(map[balancer.SubConn]connectivity.State),
    csEvltr:  &balancer.ConnectivityStateEvaluator{},
    config:   bb.config,
  }
  bal.picker = NewErrPicker(balancer.ErrNoSubConnAvailable)
  return bal
}

Build方法返回了baseBalancer，可以知道baseBalancer實現了balancer.Balancer介面：

再來回顧下這個流程，其實主要做了如下幾件事：

  1. 在自定義的Resolver中監聽服務狀態的變更
  2. 通過UpdateState來更新狀態
  3. 獲取自定義的Balancer
  4. 執行自定義Balancer的Build方法獲取Balancer

如何建立連線？

繼續回到ClientConn的updateResolverState方法，在方法的最後呼叫balancerWrapper.updateClientConnState方法更新使用者端的連線狀態：

grpc-go/clientconn.go:664

uccsErr := bw.updateClientConnState(&balancer.ClientConnState{ResolverState: s, BalancerConfig: balCfg})
if ret == nil {
ret = uccsErr // prefer ErrBadResolver state since any other error is
// currently meaningless to the caller.
}

後面的呼叫鏈路如下圖所示：

最終會呼叫baseBalancer.UpdateClientConnState方法：

grpc-go/balancer/base/balancer.go:94

func (b *baseBalancer) UpdateClientConnState(s balancer.ClientConnState) error {
  // .............
  b.resolverErr = nil
  addrsSet := resolver.NewAddressMap()
  for _, a := range s.ResolverState.Addresses {
    addrsSet.Set(a, nil)
    if _, ok := b.subConns.Get(a); !ok {
      sc, err := b.cc.NewSubConn([]resolver.Address{a}, balancer.NewSubConnOptions{HealthCheckEnabled: b.config.HealthCheck})
      if err != nil {
        logger.Warningf("base.baseBalancer: failed to create new SubConn: %v", err)
        continue
      }
      b.subConns.Set(a, sc)
      b.scStates[sc] = connectivity.Idle
      b.csEvltr.RecordTransition(connectivity.Shutdown, connectivity.Idle)
      sc.Connect()
    }
  }
  for _, a := range b.subConns.Keys() {
    sci, _ := b.subConns.Get(a)
    sc := sci.(balancer.SubConn)
    if _, ok := addrsSet.Get(a); !ok {
      b.cc.RemoveSubConn(sc)
      b.subConns.Delete(a)
    }
  }

  // ................
}

當第一次觸發呼叫UpdateClientConnState的時候，如下程式碼中 ok 為 false：

_, ok := b.subConns.Get(a);

所以會建立新的連線：

sc, err := b.cc.NewSubConn([]resolver.Address{a}, balancer.NewSubConnOptions{HealthCheckEnabled: b.config.HealthCheck})

這裡的 b.cc 即為 balancerWrapper，忘記的盆友可以往上翻看複習一下，也就是會呼叫 balancerWrapper.NewSubConn建立連線

grpc-go/internal/balancer/gracefulswitch/gracefulswitch.go:328

func (bw *balancerWrapper) NewSubConn(addrs []resolver.Address, opts balancer.NewSubConnOptions) (balancer.SubConn, error) {
  // .............

  sc, err := bw.gsb.cc.NewSubConn(addrs, opts)
  if err != nil {
    return nil, err
  }
  
  // .............
  
  bw.subconns[sc] = true
  
  // .............
}

bw.gsb.cc即為ccBalancerWrapper，所以這裡會呼叫ccBalancerWrapper.NewSubConn建立連線：

grpc-go/balancer_conn_wrappers.go:299

func (ccb *ccBalancerWrapper) NewSubConn(addrs []resolver.Address, opts balancer.NewSubConnOptions) (balancer.SubConn, error) {
  if len(addrs) <= 0 {
    return nil, fmt.Errorf("grpc: cannot create SubConn with empty address list")
  }
  ac, err := ccb.cc.newAddrConn(addrs, opts)
  if err != nil {
    channelz.Warningf(logger, ccb.cc.channelzID, "acBalancerWrapper: NewSubConn: failed to newAddrConn: %v", err)
    return nil, err
  }
  acbw := &acBalancerWrapper{ac: ac}
  acbw.ac.mu.Lock()
  ac.acbw = acbw
  acbw.ac.mu.Unlock()
  return acbw, nil
}

最終返回的是acBalancerWrapper物件，acBalancerWrapper實現了balancer.SubConn介面：

呼叫流程圖如下所示：

建立連線的預設狀態為 connectivity.Idle :

grpc-go/clientconn.go:699

func (cc *ClientConn) newAddrConn(addrs []resolver.Address, opts balancer.NewSubConnOptions) (*addrConn, error) {
  ac := &addrConn{
    state:        connectivity.Idle,
    cc:           cc,
    addrs:        addrs,
    scopts:       opts,
    dopts:        cc.dopts,
    czData:       new(channelzData),
    resetBackoff: make(chan struct{}),
  }
 
  // ...........
}

在gRPC中為連線定義了五種狀態，分別如下：

const (
  // Idle indicates the ClientConn is idle.
  Idle State = iota
  // Connecting indicates the ClientConn is connecting.
  Connecting
  // Ready indicates the ClientConn is ready for work.
  Ready
  // TransientFailure indicates the ClientConn has seen a failure but expects to recover.
  TransientFailure
  // Shutdown indicates the ClientConn has started shutting down.
  Shutdown
)

在 **baseBalancer ** 中通過b.scStates儲存建立的連線，初始狀態也為connectivity.Idle，之後通過sc.Connect()進行連線：

grpc-go/balancer/base/balancer.go:112

b.subConns.Set(a, sc)
b.scStates[sc] = connectivity.Idle
b.csEvltr.RecordTransition(connectivity.Shutdown, connectivity.Idle)
sc.Connect()

這裡sc.Connetc呼叫的是acBalancerWrapper的Connect方法，可以看到這裡建立連線是非同步進行的：

grpc-go/balancer_conn_wrappers.go:406

func (acbw *acBalancerWrapper) Connect() {
  acbw.mu.Lock()
  defer acbw.mu.Unlock()
  go acbw.ac.connect()
}

最後會呼叫addrConn.connect方法：

grpc-go/clientconn.go:786

func (ac *addrConn) connect() error {
  ac.mu.Lock()
  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
}

從connect開始的呼叫鏈路如下所示：

在baseBalancer的UpdateSubConnState方法的最後，更新了Picker為自定義的Picker：

grpc-go/balancer/base/balancer.go:221

b.cc.UpdateState(balancer.State{ConnectivityState: b.state, Picker: b.picker})

在addrConn方法的最後會呼叫ac.resetTransport()真正的進行連線的建立：

當連線已經建立好，處於Ready狀態，最後呼叫baseBalancer.UpdateSubConnState方法，此時s==connectivity.Ready為true，而oldS == connectivity.Ready為false，所以會呼叫b.regeneratePicker()方法：

if (s == connectivity.Ready) != (oldS == connectivity.Ready) ||
    b.state == connectivity.TransientFailure {
    b.regeneratePicker()
}

func (b *baseBalancer) regeneratePicker() {
  if b.state == connectivity.TransientFailure {
    b.picker = NewErrPicker(b.mergeErrors())
    return
  }
  readySCs := make(map[balancer.SubConn]SubConnInfo)

  // Filter out all ready SCs from full subConn map.
  for _, addr := range b.subConns.Keys() {
    sci, _ := b.subConns.Get(addr)
    sc := sci.(balancer.SubConn)
    if st, ok := b.scStates[sc]; ok && st == connectivity.Ready {
      readySCs[sc] = SubConnInfo{Address: addr}
    }
  }
  b.picker = b.pickerBuilder.Build(PickerBuildInfo{ReadySCs: readySCs})
}

在regeneratePicker中獲取了處於connectivity.Ready狀態可用的連線，同時更新了picker。還記得b.pickerBuilder嗎？b.b.pickerBuilder為在go-zero中自定義實現的base.PickerBuilder介面。

go-zero/zrpc/internal/balancer/p2c/p2c.go:42

func (b *p2cPickerBuilder) Build(info base.PickerBuildInfo) balancer.Picker {
  readySCs := info.ReadySCs
  if len(readySCs) == 0 {
    return base.NewErrPicker(balancer.ErrNoSubConnAvailable)
  }

  var conns []*subConn
  for conn, connInfo := range readySCs {
    conns = append(conns, &subConn{
      addr:    connInfo.Address,
      conn:    conn,
      success: initSuccess,
    })
  }

  return &p2cPicker{
    conns: conns,
    r:     rand.New(rand.NewSource(time.Now().UnixNano())),
    stamp: syncx.NewAtomicDuration(),
  }
}

最後把自定義的Picker賦值為 ClientConn.blockingpicker.picker屬性。

grpc-go/balancer_conn_wrappers.go:347

func (ccb *ccBalancerWrapper) UpdateState(s balancer.State) {
  ccb.cc.blockingpicker.updatePicker(s.Picker)
  ccb.cc.csMgr.updateState(s.ConnectivityState)
}

如何選擇已建立的連線？

現在已經知道了如何建立連線，以及連線其實是在 baseBalancer.scStates 中管理，當連線的狀態發生變化，則會更新 **baseBalancer.scStates ** 。那麼接下來我們來看一下gRPC是如何選擇一個連線進行請求的傳送的。

當gRPC使用者端發起呼叫的時候，會呼叫ClientConn的Invoke方法，一般不會主動使用該方法進行呼叫，該方法的呼叫一般是自動生成：

grpc-go/examples/helloworld/helloworld/helloworld_grpc.pb.go:39

func (c *greeterClient) SayHello(ctx context.Context, in *HelloRequest, opts ...grpc.CallOption) (*HelloReply, error) {
  out := new(HelloReply)
  err := c.cc.Invoke(ctx, "/helloworld.Greeter/SayHello", in, out, opts...)
  if err != nil {
    return nil, err
  }
  return out, nil
}

如下為發起請求的呼叫鏈路，最終會呼叫p2cPicker.Pick方法獲取連線，我們自定義的負載均衡演演算法一般都在Pick方法中實現，獲取到連線之後，通過sendMsg傳送請求。

grpc-go/stream.go:945

func (a *csAttempt) sendMsg(m interface{}, hdr, payld, data []byte) error {
  cs := a.cs
  if a.trInfo != nil {
    a.mu.Lock()
    if a.trInfo.tr != nil {
      a.trInfo.tr.LazyLog(&payload{sent: true, msg: m}, true)
    }
    a.mu.Unlock()
  }
  if err := a.t.Write(a.s, hdr, payld, &transport.Options{Last: !cs.desc.ClientStreams}); err != nil {
    if !cs.desc.ClientStreams {
      return nil
    }
    return io.EOF
  }
  if a.statsHandler != nil {
    a.statsHandler.HandleRPC(a.ctx, outPayload(true, m, data, payld, time.Now()))
  }
  if channelz.IsOn() {
    a.t.IncrMsgSent()
  }
  return nil
}

原始碼分析到此就結束了，由於篇幅有限沒法做到面面俱到，所以本文只列出了原始碼中的主要路徑。

結束語

Balancer相關的原始碼還是有點複雜的，筆者也是讀了好幾遍才理清脈絡，所以如果讀了一兩遍感覺沒有頭緒也不用著急，對照文章的脈絡多讀幾遍就一定能搞懂。

如果有疑問可以隨時找我討論，在社群群中可以搜尋dawn_zhou找到我。

希望本篇文章對你有所幫助，你的點贊是作者持續輸出的最大動力。

專案地址

https://github.com/zeromicro/go-zero

歡迎使用 go-zero 並 star 支援我們！

微信交流群

關注『微服務實踐』公眾號並點選 交流群 獲取社群群二維條碼。