Go：負(fù)載均衡原理分析與源碼解讀
上一篇文章一起學(xué)習(xí)了Resolver的原理和源碼分析，本篇繼續(xù)和大家一起學(xué)習(xí)下和Resolver關(guān)系密切的Balancer的相關(guān)內(nèi)容。這里說的負(fù)載均衡主要指數(shù)據(jù)中心內(nèi)的負(fù)載均衡，即RPC間的負(fù)載均衡。
傳送門 服務(wù)發(fā)現(xiàn)原理分析與源碼解讀
基于 go-zero v1.3.5 和 grpc-go v1.47.0
負(fù)載均衡
每一個被調(diào)用服務(wù)都會有多個實例，那么服務(wù)的調(diào)用方應(yīng)該將請求，發(fā)向被調(diào)用服務(wù)的哪一個服務(wù)實例，這就是負(fù)載均衡的業(yè)務(wù)場景。
負(fù)載均衡的第一個關(guān)鍵點是公平性，即負(fù)載均衡需要關(guān)注被調(diào)用服務(wù)實例組之間的公平性，不要出現(xiàn)旱的旱死，澇的澇死的情況。
負(fù)載均衡的第二個關(guān)鍵點是正確性，即對于有狀態(tài)的服務(wù)來說，負(fù)載均衡需要關(guān)心請求的狀態(tài)，將請求調(diào)度到能處理它的后端實例上，不要出現(xiàn)不能處理和錯誤處理的情況。
無狀態(tài)的負(fù)載均衡
無狀態(tài)的負(fù)載均衡是我們?nèi)粘９ぷ髦薪佑|比較多的負(fù)載均衡模型，它指的是參與負(fù)載均衡的后端實例是無狀態(tài)的，所有的后端實例都是對等的，一個請求不論發(fā)向哪一個實例，都會得到相同的并且正確的處理結(jié)果，所以無狀態(tài)的負(fù)載均衡策略不需要關(guān)心請求的狀態(tài)。下面介紹兩種無狀態(tài)負(fù)載均衡算法。
輪詢
輪詢的負(fù)載均衡策略非常簡單，只需要將請求按順序分配給多個實例，不用再做其他的處理。例如，輪詢策略會將第一個請求分配給第一個實例，然后將下一個請求分配給第二個實例，這樣依次分配下去，分配完一輪之后，再回到開頭分配給第一個實例，再依次分配。輪詢在路由時，不利用請求的狀態(tài)信息，屬于無狀態(tài)的負(fù)載均衡策略，所以它不能用于有狀態(tài)實例的負(fù)載均衡器，否則正確性會出現(xiàn)問題。在公平性方面，因為輪詢策略只是按順序分配請求，所以適用于請求的工作負(fù)載和實例的處理能力差異都較小的情況。
權(quán)重輪詢
權(quán)重輪詢的負(fù)載均衡策略是將每一個后端實例分配一個權(quán)重，分配請求的數(shù)量和實例的權(quán)重成正比輪詢。例如有兩個實例 A，B，假設(shè)我們設(shè)置 A 的權(quán)重為 20，B 的權(quán)重為 80，那么負(fù)載均衡會將 20% 的請求數(shù)量分配給 A，80 % 的請求數(shù)量分配給 B。權(quán)重輪詢在路由時，不利用請求的狀態(tài)信息，屬于無狀態(tài)的負(fù)載均衡策略，所以它也不能用于有狀態(tài)實例的負(fù)載均衡器，否則正確性會出現(xiàn)問題。在公平性方面，因為權(quán)重策略會按實例的權(quán)重比例來分配請求數(shù)，所以，我們可以利用它解決實例的處理能力差異的問題，認(rèn)為它的公平性比輪詢策略要好。
有狀態(tài)負(fù)載均衡
有狀態(tài)負(fù)載均衡是指，在負(fù)載均衡策略中會保存服務(wù)端的一些狀態(tài)，然后根據(jù)這些狀態(tài)按照一定的算法選擇出對應(yīng)的實例。
P2C+EWMA
在go-zero中默認(rèn)使用的是P2C的負(fù)載均衡算法。該算法的原理比較簡單，即隨機從所有可用節(jié)點中選擇兩個節(jié)點，然后計算這兩個節(jié)點的負(fù)載情況，選擇負(fù)載較低的一個節(jié)點來服務(wù)本次請求。為了避免某些節(jié)點一直得不到選擇導(dǎo)致不平衡，會在超過一定的時間后強制選擇一次。
在該復(fù)雜均衡算法中，采用了EWMA指數(shù)移動加權(quán)平均的算法，表示是一段時間內(nèi)的均值。該算法相對于算數(shù)平均來說對于突然的網(wǎng)絡(luò)抖動沒有那么敏感，突然的抖動不會體現(xiàn)在請求的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)))
系數(shù)w是一個時間衰減值，即兩次請求的間隔越大，則系數(shù)w就越小。
go-zero/zrpc/internal/balancer/p2c/p2c.go:124
w := math.Exp(float64(-td) / float64(decayTime))
節(jié)點的load值是通過該連接的請求延遲 lag 和當(dāng)前請求數(shù) inflight 的乘積所得，如果請求的延遲越大或者當(dāng)前正在處理的請求數(shù)越多表明該節(jié)點的負(fù)載越高。
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，請根據(jù)給出的代碼路徑進行區(qū)分
在gRPC中，Balancer和Resolver一樣也可以自定義，同樣也是通過Register方法進行注冊
grpc-go/balancer/balancer.go:53
func Register(b Builder) {
m[strings.ToLower(b.Name())] = b
}
Register的參數(shù)Builder為接口，在Builder接口中，Build方法的第一個參數(shù)ClientConn也為接口，Build方法的返回值Balancer同樣也是接口，定義如下：
可以看出，要想實現(xiàn)自定義的Balancer的話，就必須要實現(xiàn)balancer.Builder接口。
在了解了gRPC提供的Balancer的注冊方式之后，我們看一下go-zero是在什么地方進行Balancer注冊的
go-zero/zrpc/internal/balancer/p2c/p2c.go:36
func init() {
balancer.Register(newBuilder())
}
在go-zero中并沒有實現(xiàn) balancer.Builder 接口，而是使用gRPC提供的 base.baseBuilder 進行注冊，base.baseBuilder 實現(xiàn)了balancer.Builder 接口。創(chuàng)建baseBuilder的時候調(diào)用了 base.NewBalancerBuilder 方法，需要傳入 PickerBuilder 參數(shù)，PickerBuilder為接口，在go-zero中 p2c.p2cPickerBuilder 實現(xiàn)了該接口。
PickerBuilder接口Build方法返回值 balancer.Picker 也是一個接口，p2c.p2cPicker 實現(xiàn)了該接口。
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,
}
}
各結(jié)構(gòu)之間的關(guān)系如下圖所示，其中各結(jié)構(gòu)模塊對應(yīng)的包為：
balancer：grpc-go/balancer base：grpc-go/balancer/base p2c: go-zero/zrpc/internal/balancer/p2c
在哪里獲取已注冊的Balancer？
通過上面的流程步驟，已經(jīng)知道了如何自定義Balancer，以及如何注冊自定義的Blancer。既然注冊了肯定就會獲取，接下來看一下是在哪里獲取已經(jīng)注冊的Balancer的。
我們知道Resolver是通過解析DialContext的第二個參數(shù)target，從而得到Resolver的name，然后根據(jù)name獲取到對應(yīng)的Resolver的。獲取Balancer同樣也是根據(jù)名稱，Balancer的名稱是在創(chuàng)建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
}
在上一篇文章中，我們已經(jīng)知道當(dāng)創(chuàng)建gRPC客戶端的時候，會觸發(fā)調(diào)用自定義Resolver的Build方法，在Build方法內(nèi)部獲取到服務(wù)地址列表后，通過cc.UpdateState方法進行狀態(tài)更新，后面當(dāng)監(jiān)聽到服務(wù)狀態(tài)變化的時候同樣也會調(diào)用cc.UpdateState進行狀態(tài)的更新，而這里的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
當(dāng)監(jiān)聽到服務(wù)狀態(tài)的變更后（首次啟動或者通過Watch監(jiān)聽變化）調(diào)用 ccResolverWrapper.UpdateState 觸發(fā)更新狀態(tài)的流程，各模塊間的調(diào)用鏈路如下所示：
獲取Balancer的動作是在 ccBalancerWrapper.handleSwitchTo 方法中觸發(fā)的，代碼如下所示：
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 方法中，調(diào)用了自定義Balancer的Build方法：
grpc-go/internal/balancer/gracefulswitch/gracefulswitch.go:121
newBalancer := builder.Build(bw, gsb.bOpts)
上文有提到Build方法的第一個參數(shù)為接口 balancer.ClientConn ，而這里傳入的為 balancerWrapper ，所以gracefulswitch.balancerWrapper實現(xiàn)了該接口：
到這里我們已經(jīng)知道了獲取自定義Balancer是在哪里觸達的，以及在哪里獲取的自定義的Balancer，和balancer.Builder的Build方法在哪里被調(diào)用。
通過上文可知這里的balancer.Builder為baseBuilder，所以調(diào)用的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實現(xiàn)了balancer.Balancer接口：
再來回顧下這個流程，其實主要做了如下幾件事：
在自定義的Resolver中監(jiān)聽服務(wù)狀態(tài)的變更 通過UpdateState來更新狀態(tài) 獲取自定義的Balancer 執(zhí)行自定義Balancer的Build方法獲取Balancer
如何創(chuàng)建連接？
繼續(xù)回到ClientConn的updateResolverState方法，在方法的最后調(diào)用balancerWrapper.updateClientConnState方法更新客戶端的連接狀態(tài)：
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.
}
后面的調(diào)用鏈路如下圖所示：
最終會調(diào)用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)
}
}
// ................
}
當(dāng)?shù)谝淮斡|發(fā)調(diào)用UpdateClientConnState的時候，如下代碼中 ok 為 false：
_, ok := b.subConns.Get(a);
所以會創(chuàng)建新的連接：
sc, err := b.cc.NewSubConn([]resolver.Address{a}, balancer.NewSubConnOptions{HealthCheckEnabled: b.config.HealthCheck})
這里的 b.cc 即為 balancerWrapper，忘記的盆友可以往上翻看復(fù)習(xí)一下，也就是會調(diào)用 balancerWrapper.NewSubConn創(chuàng)建連接
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，所以這里會調(diào)用ccBalancerWrapper.NewSubConn創(chuàng)建連接：
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實現(xiàn)了balancer.SubConn接口：
調(diào)用流程圖如下所示：
創(chuàng)建連接的默認(rèn)狀態(tài)為 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中為連接定義了五種狀態(tài)，分別如下：
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保存創(chuàng)建的連接，初始狀態(tài)也為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調(diào)用的是acBalancerWrapper的Connect方法，可以看到這里創(chuàng)建連接是異步進行的：
grpc-go/balancer_conn_wrappers.go:406
func (acbw *acBalancerWrapper) Connect() {
acbw.mu.Lock()
defer acbw.mu.Unlock()
go acbw.ac.connect()
}
最后會調(diào)用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開始的調(diào)用鏈路如下所示：
在baseBalancer的UpdateSubConnState方法的最后，更新了Picker為自定義的Picker：
grpc-go/balancer/base/balancer.go:221
b.cc.UpdateState(balancer.State{ConnectivityState: b.state, Picker: b.picker})
在addrConn方法的最后會調(diào)用ac.resetTransport()真正的進行連接的創(chuàng)建：
當(dāng)連接已經(jīng)創(chuàng)建好，處于Ready狀態(tài)，最后調(diào)用baseBalancer.UpdateSubConnState方法，此時s==connectivity.Ready為true，而oldS == connectivity.Ready為false，所以會調(diào)用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狀態(tài)可用的連接，同時更新了picker。還記得b.pickerBuilder嗎？b.b.pickerBuilder為在go-zero中自定義實現(xiàn)的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)
}
如何選擇已創(chuàng)建的連接？
現(xiàn)在已經(jīng)知道了如何創(chuàng)建連接，以及連接其實是在 baseBalancer.scStates 中管理，當(dāng)連接的狀態(tài)發(fā)生變化，則會更新 **baseBalancer.scStates ** 。那么接下來我們來看一下gRPC是如何選擇一個連接進行請求的發(fā)送的。
當(dāng)gRPC客戶端發(fā)起調(diào)用的時候，會調(diào)用ClientConn的Invoke方法，一般不會主動使用該方法進行調(diào)用，該方法的調(diào)用一般是自動生成：
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
}
如下為發(fā)起請求的調(diào)用鏈路，最終會調(diào)用p2cPicker.Pick方法獲取連接，我們自定義的負(fù)載均衡算法一般都在Pick方法中實現(xiàn)，獲取到連接之后，通過sendMsg發(fā)送請求。
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
}
源碼分析到此就結(jié)束了，由于篇幅有限沒法做到面面俱到，所以本文只列出了源碼中的主要路徑。
結(jié)束語
Balancer相關(guān)的源碼還是有點復(fù)雜的，筆者也是讀了好幾遍才理清脈絡(luò)，所以如果讀了一兩遍感覺沒有頭緒也不用著急，對照文章的脈絡(luò)多讀幾遍就一定能搞懂。
如果有疑問可以隨時找我討論，在社區(qū)群中可以搜索dawn_zhou找到我。
希望本篇文章對你有所幫助，你的點贊是作者持續(xù)輸出的最大動力。
推薦閱讀