自适应负载均衡算法原理和实现

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

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

type subConn struct {
    addr     resolver.Address
    conn     balancer.SubConn
    lag      uint64 // 用来保存 ewma 值
    inflight int64  // 用在保存当前节点正在处理的请求总数
    success  uint64 // 用来标识一段时间内此连接的健康状态
    requests int64  // 用来保存请求总数
    last     int64  // 用来保存上一次请求耗时, 用于计算 ewma 值
    pick     int64  // 保存上一次被选中的时间点
}

type p2cPicker struct {
  conns []*subConn  // 保存所有节点的信息 
  r     *rand.Rand
  stamp *syncx.AtomicDuration
  lock  sync.Mutex
}

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(),
  }
}

switch len(p.conns) {
  case 0: // 没有节点，返回错误
    return emptyPickResult, balancer.ErrNoSubConnAvailable
  case 1: // 有一个节点，直接返回这个节点
    chosen = p.choose(p.conns[0], nil)
  case 2: // 有两个节点，计算负载，返回负载低的节点
    chosen = p.choose(p.conns[0], p.conns[1])
  default: // 有多个节点，p2c 挑选两个节点，比较这两个节点的负载，返回负载低的节点
    var node1, node2 *subConn
    // 3次随机选择两个节点
    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]
      // 如果这次选择的节点达到了健康要求, 就中断选择
      if node1.healthy() && node2.healthy() {
        break
      }
    }
    // 比较两个节点的负载情况，选择负载低的
    chosen = p.choose(node1, node2)
  }

func (c *subConn) load() int64 {
  // 通过 EWMA 计算节点的负载情况； 加 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
}

func (p *p2cPicker) buildDoneFunc(c *subConn) func(info balancer.DoneInfo) {
  start := int64(timex.Now())
  return func(info balancer.DoneInfo) {
    // 正在处理的请求数减 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
    }
    // 用牛顿冷却定律中的衰减函数模型计算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()
      }
    }
  }
}