負載均衡原理分析與源碼解讀
上一篇文章一起學習了 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 接口:
再來回顧下這個流程,其實主要做了如下幾件事:
-
在自定義的 Resolver 中監聽服務狀態的變更
-
通過 UpdateState 來更新狀態
-
獲取自定義的 Balancer
-
執行自定義 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 支持我們!
微信交流羣
關注『微服務實踐』公衆號並點擊 交流羣 獲取社區羣二維碼。
本文由 Readfog 進行 AMP 轉碼,版權歸原作者所有。
來源:https://mp.weixin.qq.com/s/FwBlCB0RKUOeq6PiKr2GQw