一致性哈希 (Consistent Hashing) 算法, 主要应用在分布式系统中, 尤其是缓存, 解决负载均衡、热点、数据倾斜等问题.
一.目的
将服务器节点 Node, 映射到一个较长的环上, 当查找数据时, 在环上顺时针查找, 该数据的 Key 所临近的服务器节点 Node, 最后再到服务器节点上去查找源数据.
二.个人对算法的理解
1.取一个环, 作为哈希环.
长度一般是 2 的 32 次方 (4,294,967,296), 即 int32 的数值范围.
2.定一个哈希规则, 哈希结果的值范围在要跟哈希环的长度适配. 比如环长度为 2 的 32 次方, 则选用 CRC32 规则; Redis-Cluster Slots 的个数是 2 的 14 次方 (16384), 使用的规则是 CRC16.
3.使用哈希规则对服务器节点 Node 进行哈希运算. 比如对服务器节点的 IP、HostName 等, 进行 CRC32 哈希运算.
4.将哈希运算的结果, 对哈希环的长度进行取余运算, 然后将余数映射到哈希环上.
5.对查找数据的 Key 进行 CRC32 哈希运算, 并对哈希环的长度进行哈希取余操作, 然后顺时针查找临近的 Node. (Key 比如 user 表的 user id)
三.主要缺点
1.无法满足负载均衡.
如果节点个数比较少, 极端假设有两个节点, 一个节点 A 的 Hash 值为 0, 另一个节点 B 的 Hash 值为 10, 而一般哈希环的长度一般都相当长 (如 int32 范围), 导致的结果是 A 负载 11 ~ int32最大数值, 而仅仅负载 1~10.
2.新增节点导致大量数据迁移.
如果新增一个节点 C, Hash 值为 3,000,000,000, 那么 11 ~3,000,000,000 对应的这么大量的数据, 会从 A 迁移到 C 来.
新增的节点, 只为一个节点分摊压力, 并不是都为其他节点分摊一部分的负载压力.
四.解决的关键思想
节点越多, 分布越均匀.
负载越近似均衡, 数据倾斜越少.
如果没有实际节点, 那就通过新增虚拟节点.
于是, 基于虚拟节点的一致性哈希诞生.
五.用 go 语言实现, 基于虚拟节点的一致性哈希
1.接口抽象和结构体定义
根据迪米勒原则, 抽象以下接口
type IConsistentHash interface {
AddNode(node Node) error
RemoveNode(node Node) error
GetNode(key string) Node
}
真实节点 Node, 作为值类型.
type Node struct {
Name string // 服务器名称
IP string // IP
Port int // 端口
}
虚拟节点, 将用 map 来与真实节点的关联, RingIndexs 切片的索引就是虚拟节点的 ID.
type virtualNode struct {
// 节点的 Hash 在哈希环上的索引
RingIndexs []uint32
}
基于虚拟节点的一致性哈希.
type ConsistentHash struct {
virtualNodeAmount int // 虚拟节点个数
nodeMap map[Node]*virtualNode // 真实节点 映射 虚拟节点
hashVals []uint32 // 所用虚拟节点的 Hash 值, 主要用于检索
hashValMap map[uint32]Node // 虚拟节点 Hash 值, 映射真实节点
}
2.具体实现
2.1 虚拟节点 Key 的构造
func (*ConsistentHash) node2Str(n Node) string {
// fmt.Sprintf("%s<%s:%d>", n.Name, n.IP, n.Port)
return strings.Join([]string{
n.Name, "<", n.IP, ":", strconv.Itoa(n.Port), ">",
}, "")
}
func (c *ConsistentHash) genKey(nodeStr string, i int) string {
// nodeA#0, nodeA#1, nodeA#2, ...
// nodeB#0, nodeB#1, nodeB#2, ...
return strings.Join([]string{nodeStr, strconv.Itoa(i)}, "#")
}
2.2 哈希算法
因为算法 hash 范围正是环的长度范围, 所以不对 math.MaxInt32 取余.
func (c *ConsistentHash) hashAlg(str string) uint32 {
var bytes = []byte(str)
return crc32.ChecksumIEEE(bytes)
}
2.3 对所有虚拟节点的 Hash 值的排序
快排算法, 时间复杂度: O(NlogN), N 为虚拟节点的个数
func (c *ConsistentHash) sort() {
hashVals := make([]uint32, 0, len(c.hashValMap))
// concat all RingIndexs
for _, vn := range c.nodeMap {
hashVals = append(hashVals, vn.RingIndexs...)
}
sort.Slice(hashVals, func(i, j int) bool {
return hashVals[i] < hashVals[j]
})
c.hashVals = hashVals
}
2.5 接口实现-新增和删除节点
在新增节点时, 自动创建虚拟节点, 在删除节点时, 将节点对应的所有虚拟节点进行拔除.
并在修改节点后, 重新对所有虚拟节点的 Hash 值进行排序.
var (
ErrNodeExisted = errors.New("Node is Existed")
ErrNodeNotFound = errors.New("Node is not Found")
)
func (c *ConsistentHash) AddNode(node Node) error {
if _, ok := c.nodeMap[node]; ok {
return ErrNodeExisted
}
s := c.node2Str(node)
vn := &virtualNode{
RingIndexs: make([]uint32, 0, c.virtualNodeAmount),
}
for i := 0; i < c.virtualNodeAmount; i++ {
hash := c.hashAlg(c.genKey(s, i))
vn.RingIndexs = append(vn.RingIndexs, hash)
c.hashValMap[hash] = node
}
c.nodeMap[node] = vn
c.sort()
return nil
}
func (c *ConsistentHash) RemoveNode(node Node) error {
if _, ok := c.nodeMap[node]; !ok {
return ErrNodeNotFound
}
delete(c.nodeMap, node)
s := c.node2Str(node)
for i := 0; i < c.virtualNodeAmount; i++ {
hash := c.hashAlg(c.genKey(s, i))
if _, ok := c.hashValMap[hash]; ok {
delete(c.hashValMap, hash)
}
}
c.sort()
return nil
}
2.6 接口实现-获取节点
使用折半算法, 对节点进行查找
func (c *ConsistentHash) GetNode(key string) Node {
// c.sort()
hash := c.hashAlg(key)
l, r := 0, len(c.hashVals)-1
for l < r {
m := (l + r) / 2
if hash > c.hashVals[m] {
l = m + 1
} else if hash < c.hashVals[m] {
r = m - 1
} else {
ringIndex := c.hashVals[m]
log.Printf("[ConsistentHash/GetNode] key-HashVal:%v, ringIndex:%v.\n", hash, ringIndex)
return c.hashValMap[ringIndex]
}
}
index := l
if hash > c.hashVals[index] {
index = (l + 1) % len(c.hashVals)
}
ringIndex := c.hashVals[index]
log.Printf("[ConsistentHash/GetNode] key-HashVal:%v, ringIndex:%v.\n", hash, ringIndex)
return c.hashValMap[ringIndex]
}
3.测试
3.1 测试代码
主要验证获取的节点是否正确, 以及对新增和删除节点的操作是否正确.
真实节点个数: 10
虚拟节点个数: 16
import (
"testing"
)
var nodes = []Node{
{Name: "appA.service.order", IP: "192.168.1.100", Port: 5000},
{Name: "appA.service.pay", IP: "192.168.1.101", Port: 5000},
{Name: "appA.service.history", IP: "192.168.1.102", Port: 5000},
{Name: "appA.service.detail", IP: "192.168.1.103", Port: 5000},
{Name: "appA.service.comment", IP: "192.168.1.104", Port: 5000},
{Name: "appA.service.mq", IP: "192.168.1.105", Port: 5000},
{Name: "appA.service.mysql", IP: "192.168.1.106", Port: 5000},
{Name: "appA.service.gateway", IP: "192.168.1.107", Port: 5000},
{Name: "appB.service.geo", IP: "192.168.1.108", Port: 5000},
{Name: "appC.service.cdn", IP: "192.168.1.109", Port: 5000},
}
var user_id = "123456"
func TestConsistentHash_GetNode(t *testing.T) {
c := NewConsistentHash(16, nodes...)
node := c.GetNode(user_id)
t.Logf("GetNode: %+v\n", node)
for node, vn := range c.nodeMap {
t.Logf("%+v, %+v\n", node.IP, vn.RingIndexs)
}
t.Log(c.hashVals)
}
func TestConsistentHash(t *testing.T) {
c := NewConsistentHash(16, nodes...)
node := c.GetNode(user_id)
t.Logf("GetNode: %+v\n", node)
err := c.RemoveNode(node)
t.Log("RemoveNode Err:", err)
err = c.AddNode(node)
t.Log("AddNode Err:", err)
node = c.GetNode(user_id)
t.Logf("GetNode: %+v\n", node)
}
3.2 测试结果
排序、新增节点、删除节点、获取节点, 貌似都 Okay.
=== RUN TestConsistentHash_GetNode
2021/06/25 [ConsistentHash/GetNode] key-HashVal:158520161, ringIndex:256098299.
consistenthash_test.go:26: GetNode: {Name:appA.service.pay IP:192.168.1.101 Port:5000}
...
consistenthash_test.go:30: [9950081 24722963 133830552 137264098 147812920 256098299 262321697 ...]
--- PASS: TestConsistentHash_GetNode (0.00s)
=== RUN TestConsistentHash
2021/06/25 [ConsistentHash/GetNode] key-HashVal:158520161, ringIndex:256098299.
consistenthash_test.go:37: GetNode: {Name:appA.service.pay IP:192.168.1.101 Port:5000}
consistenthash_test.go:39: RemoveNode Err: <nil>
consistenthash_test.go:41: AddNode Err: <nil>
2021/06/25 [ConsistentHash/GetNode] key-HashVal:158520161, ringIndex:256098299.
consistenthash_test.go:43: GetNode: {Name:appA.service.pay IP:192.168.1.101 Port:5000}
--- PASS: TestConsistentHash (0.00s)
PASS
ok
