背景

在選擇負載均衡算法時菩混，我們希望滿足以下要求：

1. 具備分區(qū)和機房調(diào)度親和性
   • 每次選擇的節(jié)點盡量是負載最低的
   • 每次盡可能選擇響應最快的節(jié)點
2. 無需人工干預故障節(jié)點
   • 當一個節(jié)點有故障時驹马，負載均衡算法可以自動隔離該節(jié)點
   • 當故障節(jié)點恢復時焰坪，能夠自動恢復對該節(jié)點的流量分發(fā)

基于這些考慮蛇耀，go-zero 選擇了 p2c+EWMA 算法來實現(xiàn)有巧。

算法的核心思想

p2c

p2c (Pick Of 2 Choices) 二選一: 在多個節(jié)點中隨機選擇兩個節(jié)點。

go-zero 中的會隨機的選擇3次，如果其中一次選擇的節(jié)點的健康條件滿足要求寺惫，就中斷選擇，采用這兩個節(jié)點蹦疑。

EWMA

EWMA (Exponentially Weighted Moving-Average) 指數(shù)移動加權平均法: 是指各數(shù)值的加權系數(shù)隨時間呈指數(shù)遞減西雀，越靠近當前時刻的數(shù)值加權系數(shù)就越大，體現(xiàn)了最近一段時間內(nèi)的平均值歉摧。

• 公式:

  EWMA公式

• 變量解釋:

  • Vt: 代表的是第 t 次請求的 EWMA值
  • Vt-1: 代表的是第 t-1 次請求的 EWMA值
  • β: 是一個常量

EWMA 算法的優(yōu)勢

1. 相較于普通的計算平均值算法艇肴，EWMA 不需要保存過去所有的數(shù)值，計算量顯著減少判莉，同時也減小了存儲資源豆挽。
2. 傳統(tǒng)的計算平均值算法對網(wǎng)絡耗時不敏感, 而 EWMA 可以通過請求頻繁來調(diào)節(jié) β，進而迅速監(jiān)控到網(wǎng)絡毛刺或更多的體現(xiàn)整體平均值券盅。
   • 當請求較為頻繁時, 說明節(jié)點網(wǎng)絡負載升高了, 我們想監(jiān)測到此時節(jié)點處理請求的耗時(側面反映了節(jié)點的負載情況), 我們就相應的調(diào)小β帮哈。β越小，EWMA值 就越接近本次耗時锰镀，進而迅速監(jiān)測到網(wǎng)絡毛刺;
   • 當請求較為不頻繁時, 我們就相對的調(diào)大β值娘侍。這樣計算出來的 EWMA值 越接近平均值

β計算

go-zero 采用的是牛頓冷卻定律中的衰減函數(shù)模型計算 EWMA 算法中的 β 值:

牛頓冷卻定律中的衰減函數(shù)

其中 Δt 為兩次請求的間隔，e泳炉，k 為常數(shù)

gRPC 中實現(xiàn)自定義負載均衡器

1. 首先我們需要實現(xiàn) google.golang.org/grpc/balancer/base/base.go/PickerBuilder 接口, 這個接口是有服務節(jié)點更新的時候會調(diào)用接口里的Build方法

   type PickerBuilder interface {
       // Build returns a picker that will be used by gRPC to pick a SubConn.
       Build(info PickerBuildInfo) balancer.Picker
   }
   

2. 還要實現(xiàn) google.golang.org/grpc/balancer/balancer.go/Picker 接口憾筏。這個接口主要實現(xiàn)負載均衡，挑選一個節(jié)點供請求使用

   type Picker interface {
     Pick(info PickInfo) (PickResult, error)
   }
   

3. 最后向負載均衡 map 中注冊我們實現(xiàn)的負載均衡器

go-zero 實現(xiàn)負載均衡的主要邏輯

1. 在每次節(jié)點更新花鹅，gRPC 會調(diào)用 Build 方法氧腰，此時在 Build 里實現(xiàn)保存所有的節(jié)點信息。
2. gRPC 在獲取節(jié)點處理請求時刨肃，會調(diào)用 Pick 方法以獲取節(jié)點古拴。go-zero 在 Pick 方法里實現(xiàn)了 p2c 算法，挑選節(jié)點真友，并通過節(jié)點的 EWMA值 計算負載情況黄痪，返回負載低的節(jié)點供 gRPC 使用。
3. 在請求結束的時候 gRPC 會調(diào)用 PickResult.Done 方法盔然，go-zero 在這個方法里實現(xiàn)了本次請求耗時等信息的存儲桅打，并計算出了 EWMA值 保存了起來是嗜，供下次請求時計算負載等情況的使用。

負載均衡代碼分析

1. 保存服務的所有節(jié)點信息

   我們需要保存節(jié)點處理本次請求的耗時挺尾、EWMA 等信息鹅搪，go-zero 給每個節(jié)點設計了如下結構：

   type subConn struct {
       addr     resolver.Address
       conn     balancer.SubConn
       lag      uint64 // 用來保存 ewma 值
       inflight int64  // 用在保存當前節(jié)點正在處理的請求總數(shù)
       success  uint64 // 用來標識一段時間內(nèi)此連接的健康狀態(tài)
       requests int64  // 用來保存請求總數(shù)
       last     int64  // 用來保存上一次請求耗時, 用于計算 ewma 值
       pick     int64  // 保存上一次被選中的時間點
   }
   

2. p2cPicker 實現(xiàn)了 balancer.Picker 接口，conns 保存了服務的所有節(jié)點信息

   type p2cPicker struct {
     conns []*subConn  // 保存所有節(jié)點的信息 
     r     *rand.Rand
     stamp *syncx.AtomicDuration
     lock  sync.Mutex
   }
   

3. gRPC 在節(jié)點有更新的時候會調(diào)用 Build 方法潦嘶，傳入所有節(jié)點信息涩嚣，我們在這里把每個節(jié)點信息用 subConn 結構保存起來崇众。并歸并到一起用 p2cPicker 結構保存起來

   func (b *p2cPickerBuilder) Build(info base.PickerBuildInfo) balancer.Picker {
     ......
     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(),
     }
   }
   

4. 隨機挑選節(jié)點信息掂僵，在這里分了三種情況:

   1. 只有一個服務節(jié)點，此時直接返回供 gRPC 使用即可
   2. 有兩個服務節(jié)點幅狮，通過 EWMA值 計算負載椎侠，并返回負載低的節(jié)點返回供 gRPC 使用
   3. 有多個服務節(jié)點旬昭，此時通過 p2c 算法選出兩個節(jié)點，比較負載情況芹扭，返回負載低的節(jié)點供 gRPC 使用

   主要實現(xiàn)代碼如下：

   switch len(p.conns) {
     case 0:// 沒有節(jié)點，返回錯誤
       return emptyPickResult, balancer.ErrNoSubConnAvailable
     case 1:// 有一個節(jié)點赦抖，直接返回這個節(jié)點
       chosen = p.choose(p.conns[0], nil)
     case 2:// 有兩個節(jié)點舱卡，計算負載，返回負載低的節(jié)點
       chosen = p.choose(p.conns[0], p.conns[1])
     default:// 有多個節(jié)點队萤，p2c 挑選兩個節(jié)點轮锥，比較這兩個節(jié)點的負載，返回負載低的節(jié)點
       var node1, node2 *subConn
           // 3次隨機選擇兩個節(jié)點
       for i := 0; i < pickTimes; i++ {
         a := p.r.Intn(len(p.conns))
         b := p.r.Intn(len(p.conns) - 1)
         if b >= a {
           b++
         }
         node1 = p.conns[a]
         node2 = p.conns[b]
         // 如果這次選擇的節(jié)點達到了健康要求, 就中斷選擇
         if node1.healthy() && node2.healthy() {
           break
         }
       }
       // 比較兩個節(jié)點的負載情況要尔，選擇負載低的
       chosen = p.choose(node1, node2)
     }
   

5. load計算節(jié)點的負載情況

   上面的 choose 方法會調(diào)用 load 方法來計算節(jié)點負載舍杜。

   計算負載的公式是: load = ewma * inflight

   在這里簡單解釋下： ewma 相當于平均請求耗時，inflight 是當前節(jié)點正在處理請求的數(shù)量赵辕，相乘大致計算出了當前節(jié)點的網(wǎng)絡負載既绩。

   func (c *subConn) load() int64 {
     // 通過 EWMA 計算節(jié)點的負載情況； 加 1 是為了避免為 0 的情況
     lag := int64(math.Sqrt(float64(atomic.LoadUint64(&c.lag) + 1)))
     load := lag * (atomic.LoadInt64(&c.inflight) + 1)
     if load == 0 {
       return penalty
     }
     return load
   }
   

6. 請求結束还惠，更新節(jié)點的 EWMA 等信息

   1. 把節(jié)點正在處理請求的總數(shù)減1
   2. 保存處理請求結束的時間點饲握，用于計算距離上次節(jié)點處理請求的差值，并算出 EWMA 中的 β值
   3. 計算本次請求耗時蚕键，并計算出 EWMA值 保存到節(jié)點的 lag 屬性里
   4. 計算節(jié)點的健康狀態(tài)保存到節(jié)點的 success 屬性中

   func (p *p2cPicker) buildDoneFunc(c *subConn) func(info balancer.DoneInfo) {
     start := int64(timex.Now())
     return func(info balancer.DoneInfo) {
       // 正在處理的請求數(shù)減 1
       atomic.AddInt64(&c.inflight, -1)
       now := timex.Now()
       // 保存本次請求結束時的時間點救欧，并取出上次請求時的時間點
       last := atomic.SwapInt64(&c.last, int64(now))
       td := int64(now) - last
       if td < 0 {
         td = 0
       }
       // 用牛頓冷卻定律中的衰減函數(shù)模型計算EWMA算法中的β值
       w := math.Exp(float64(-td) / float64(decayTime))
       // 保存本次請求的耗時
       lag := int64(now) - start
       if lag < 0 {
         lag = 0
       }
       olag := atomic.LoadUint64(&c.lag)
       if olag == 0 {
         w = 0
       }
       // 計算 EWMA 值
       atomic.StoreUint64(&c.lag, uint64(float64(olag)*w+float64(lag)*(1-w)))
       success := initSuccess
       if info.Err != nil && !codes.Acceptable(info.Err) {
         success = 0
       }
       osucc := atomic.LoadUint64(&c.success)
       atomic.StoreUint64(&c.success, uint64(float64(osucc)*w+float64(success)*(1-w)))
   
       stamp := p.stamp.Load()
       if now-stamp >= logInterval {
         if p.stamp.CompareAndSwap(stamp, now) {
           p.logStats()
         }
       }
     }
   }
   

項目地址

https://github.com/tal-tech/go-zero

歡迎使用 go-zero 并 star 支持我們！

微信交流群

關注『微服務實踐』公眾號并點擊 交流群 獲取社區(qū)群二維碼嚎幸。