负载均衡算法实现

在这篇文章中，我将实现最常见的四种负载均衡算法，即随机负载、轮询负载、加权负载和一致性hash负载

随机负载

随机挑选目标服务器ip

type RandomBalance struct {
   curIndex int
   rss      []string
}

// 添加新的服务ip
func (r *RandomBalance) Add(params ...string) error {
   if len(params) == 0 {
      return errors.New("param len 1 at least")
   }
   addr := params[0]
   r.rss = append(r.rss, addr)
   return nil
}

// 采用rand.Intn随机取一个服务ip
func (r *RandomBalance) Next() string {
   if len(r.rss) == 0 {
      return ""
   }
   r.curIndex = rand.Intn(len(r.rss))
   return r.rss[r.curIndex]
}

func (r *RandomBalance) Get(key string) (string, error) {
   return r.Next(), nil
}

轮询负载

ABC三台服务器,A B C A B C依次轮询

type RoundRobinBalance struct {
   curIndex int
   rss      []string
}

func (r *RoundRobinBalance) Add(params ...string) error {
   if len(params) == 0 {
      return errors.New("param len 1 at least")
   }
   addr := params[0]
   r.rss = append(r.rss, addr)
   return nil
}

func (r *RoundRobinBalance) Next() string {
   if len(r.rss) == 0 {
      return ""
   }
   lens := len(r.rss)
   if r.curIndex >= lens {
      r.curIndex = 0
   }
    // 保存一个服务ip的游标
   curAddr := r.rss[r.curIndex]
   r.curIndex = (r.curIndex + 1) % lens
   return curAddr
}

func (r *RoundRobinBalance) Get(key string) (string, error) {
   return r.Next(), nil
}

加权负载

给目标设置访问权重,按照权重轮询

nginx的加权负载均衡策略

计算策略：

1. currentWeight += effectiveWeight
2. 选出最大的currentWeight节点作为选中节点
3. currentWeight -= totalWeight
   • 其中effectiveWeight的值不变，只有当服务异常的时候减少

type WeightRoundRobinBalance struct {
   curIndex int
   rss      []*WeightNode
   rsw      []int
}

type WeightNode struct {
   addr            string
   weight          int //权重值
   currentWeight   int //节点当前权重
   effectiveWeight int //有效权重
}

func (r *WeightRoundRobinBalance) Add(params ...string) error {
   if len(params) != 2 {
      return errors.New("param len need 2")
   }
   parInt, err := strconv.ParseInt(params[1], 10, 64)
   if err != nil {
      return err
   }
   node := &WeightNode{addr: params[0], weight: int(parInt)}
   node.effectiveWeight = node.weight
   r.rss = append(r.rss, node)
   return nil
}

// Next 参考了 nginx 的加权负载均衡的策略
func (r *WeightRoundRobinBalance) Next() string {
   total := 0
   var best *WeightNode
   for i := 0; i < len(r.rss); i++ {
      w := r.rss[i]
      //step 1 统计所有有效权重之和
      total += w.effectiveWeight

      //step 2 变更节点临时权重为的节点临时权重+节点有效权重
      w.currentWeight += w.effectiveWeight

      //step 3 有效权重默认与权重相同，通讯异常时-1, 通讯成功+1，直到恢复到weight大小
      if w.effectiveWeight < w.weight {
         w.effectiveWeight++
      }
      //step 4 选择最大临时权重点节点
      if best == nil || w.currentWeight > best.currentWeight {
         best = w
      }
   }
   if best == nil {
      return ""
   }
   //step 5 变更临时权重为 临时权重-有效权重之和
   best.currentWeight -= total
   return best.addr
}

func (r *WeightRoundRobinBalance) Get(key string) (string, error) {
   return r.Next(), nil
}

一致性hash负载

请求固定URL访问指定IP

type Hash func(data []byte) uint32

type UInt32Slice []uint32

func (s UInt32Slice) Len() int {
   return len(s)
}

func (s UInt32Slice) Less(i, j int) bool {
   return s[i] < s[j]
}

func (s UInt32Slice) Swap(i, j int) {
   s[i], s[j] = s[j], s[i]
}

type ConsistentHashBanlance struct {
   mux      sync.RWMutex
   hash     Hash
   replicas int               // 复制因子
   keys     UInt32Slice       // 已排序的节点hash切片
   hashMap  map[uint32]string // 节点哈希和Key的map,键是hash值，值是节点key
}

func NewConsistentHashBanlance(replicas int, fn Hash) *ConsistentHashBanlance {
   m := &ConsistentHashBanlance{
      replicas: replicas,
      hash:     fn,
      hashMap:  make(map[uint32]string),
   }
   if m.hash == nil {
      //最多32位,保证是一个2^32-1环
      m.hash = crc32.ChecksumIEEE
   }
   return m
}

// 验证是否为空
func (c *ConsistentHashBanlance) IsEmpty() bool {
   return len(c.keys) == 0
}

// Add 方法用来添加缓存节点，参数为节点key，比如使用IP
func (c *ConsistentHashBanlance) Add(params ...string) error {
   if len(params) == 0 {
      return errors.New("param len 1 at least")
   }
   addr := params[0]
   c.mux.Lock()
   defer c.mux.Unlock()
   // 结合复制因子计算所有虚拟节点的hash值，并存入m.keys中，同时在m.hashMap中保存哈希值和key的映射
   for i := 0; i < c.replicas; i++ {
      hash := c.hash([]byte(strconv.Itoa(i) + addr))
      c.keys = append(c.keys, hash)
      c.hashMap[hash] = addr
   }
   // 对所有虚拟节点的哈希值进行排序，方便之后进行二分查找
   sort.Sort(c.keys)
   return nil
}

// Get 方法根据给定的对象获取最靠近它的那个节点
func (c *ConsistentHashBanlance) Get(key string) (string, error) {
   if c.IsEmpty() {
      return "", errors.New("node is empty")
   }
   hash := c.hash([]byte(key))

   // 通过二分查找获取最优节点，第一个"服务器hash"值大于"数据hash"值的就是最优"服务器节点"
   idx := sort.Search(len(c.keys), func(i int) bool { return c.keys[i] >= hash })

   // 如果查找结果 大于 服务器节点哈希数组的最大索引，表示此时该对象哈希值位于最后一个节点之后，那么放入第一个节点中
   if idx == len(c.keys) {
      idx = 0
   }
   c.mux.RLock()
   defer c.mux.RUnlock()
   return c.hashMap[c.keys[idx]], nil
}

func (c *ConsistentHashBanlance) SetConf(conf LoadBalanceConf) {
   c.conf = conf
}