负载均衡算法-golang实现
随机负载
请求随机分配到各个服务器。
优点：使用简单；
缺点：不适合机器配置不同的场景；
通过系统的随机算法，根据后端服务器的列表大小值来随机选取其中的一台服务器进行访问。由概率统计理论可以得知，随着客户端调用服务端的次数增多，
其实际效果越来越接近于平均分配调用量到后端的每一台服务器，也就是轮询的结果。
package load_balance
import ( "errors" "math/rand")
//随机负载均衡type RandomBalance struct { curIndex int
rss []string}
func (r *RandomBalance) Add(params ...string) error { if len(params) == 0 { return errors.New("params len 1 at least") } addr := params[0] r.rss = append(r.rss, addr)
return nil}
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(string) (string, error) { return r.Next(), nil}
轮询负载
将所有请求依次分发到每台服务器上，适合服务器硬件配置相同的场景。
优点：每台服务器的请求数目相同；
缺点：服务器压力不一样，不适合服务器配置不同的情况；
将请求按顺序轮流地分配到后端服务器上，它均衡地对待后端的每一台服务器，而不关心服务器实际的连接数和当前的系统负载。
package load_balance
import "errors"
//轮询负载均衡type RoundRobinBalance struct { curIndex int rss []string}
func (r *RoundRobinBalance) Add(params ...string) error { if len(params) == 0 { return errors.New("params 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 }
curAddr := r.rss[r.curIndex] r.curIndex = (r.curIndex + 1) % lens return curAddr}
func (r *RoundRobinBalance) Get(string) (string, error) { return r.Next(), nil}
加权轮询负载
与加权轮询法一样，加权随机法也根据后端机器的配置，系统的负载分配不同的权重。不同的是，它是按照权重随机请求后端服务器，而非顺序。
package load_balance
import ( "errors" "strconv")
type WeightRoundRobinBalance struct { curIndex int rss []*WeightNode rsw []int}
type WeightNode struct { addr string Weight int //初始化时对节点约定的权重 currentWeight int //节点临时权重，每轮都会变化 effectiveWeight int //有效权重, 默认与weight相同 , totalWeight = sum(effectiveWeight) //出现故障就-1}
//1, currentWeight = currentWeight + effectiveWeight//2, 选中最大的currentWeight节点为选中节点//3, currentWeight = currentWeight - totalWeight
func (r *WeightRoundRobinBalance) Add(params ...string) error { if len(params) != 2 { return errors.New("params 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}
func (r *WeightRoundRobinBalance) Next() string { var best *WeightNode total := 0 for i := 0; i < len(r.rss); i++ { w := r.rss[i] //1 计算所有有效权重 total += w.effectiveWeight //2 修改当前节点临时权重 w.currentWeight += w.effectiveWeight //3 有效权重默认与权重相同，通讯异常时-1, 通讯成功+1，直到恢复到weight大小 if w.effectiveWeight < w.Weight { w.effectiveWeight++ }
//4 选中最大临时权重节点 if best == nil || w.currentWeight > best.currentWeight { best = w } }
if best == nil { return "" } //5 变更临时权重为 临时权重-有效权重之和 best.currentWeight -= total return best.addr}
func (r *WeightRoundRobinBalance) Get(string) (string, error) { return r.Next(), nil}
func (r *WeightRoundRobinBalance) Update() {
}
一致性hash
一致性Hash是一种特殊的Hash算法，由于其均衡性、持久性的映射特点，被广泛的应用于负载均衡领域，如nginx和memcached都采用了一致性Hash来作为集群负载均衡的方案。
一致性hash算法具备以下特性：
客户端请求最终会落到他所在hash环的顺时针方向的第一个节点上，在上图中，key3，key14，key1024请求最终会落到node1上面，而key5，key76982，key18会落到node3上面。
当集群中删除节点nodex时，原本应该落在nodex上面的请求，会被转移到nodex顺时针的下一个节点，新增节点同理，可以看到，无论新增还是删除一个节点，受影响的都只有一个节点的数据，如下图：
以上的一致性hash算法相比较普通的hash算法有了很大的改进，但是依然存在问题，以删除节点为例，删除一个节点后，集群中大部分的请求key都会落到node2这个节点上，已经并不是"负载均衡"了。一般的hash环空间会很大，而如果当集群中节点数量不是很多的时候，节点在环上面的位置可能会挤在很小的一部分区域，这样就导致一大部分请求会落到某个节点上（数据倾斜），为了解决这个问题，一致性hash算法引入虚拟节点。
package load_balance
import ( "errors" "hash/crc32" "sort" "strconv" "sync")
//1 单调性（唯一） 2平衡性 (数据 目标元素均衡) 3分散性(散列)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 ConsistentHashBalance struct { mux sync.RWMutex hash Hash replicas int //复制因子 keys UInt32Slice //已排序的节点hash切片 hashMap map[uint32]string //节点哈希和key的map, 键是hash值，值是节点key}
func NewConsistentHashBalance(replicas int, fn Hash) *ConsistentHashBalance { m := &ConsistentHashBalance{ 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 *ConsistentHashBalance) IsEmpty() bool { return len(c.keys) == 0}
// Add 方法用来添加缓存节点，参数为节点key，比如使用IPfunc (c *ConsistentHashBalance) 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 *ConsistentHashBalance) 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}