- 采用一致性hash算法将key分散到不同的节点,客户端可以连接到集群中任意一个节点
- https://github.com/csgopher/go-redis
- 本文涉及以下文件:
consistenthash:实现添加和选择节点方法
standalone_database:单机database
client:客户端
client_pool:实现连接池
cluster_database:对key进行路由
com:与其他节点通信
router,ping,keys,del,select:各类命令的转发具体逻辑
一致性哈希
为什么需要一致性 hash?
在采用分片方式建立分布式缓存时,我们面临的第一个问题是如何决定存储数据的节点。最自然的方式是参考 hash 表的做法,假设集群中存在 n 个节点,我们用 node = hashCode(key) % n 来决定所属的节点。
普通 hash 算法解决了如何选择节点的问题,但在分布式系统中经常出现增加节点或某个节点宕机的情况。若节点数 n 发生变化, 大多数 key 根据 node = hashCode(key) % n 计算出的节点都会改变。这意味着若要在 n 变化后维持系统正常运转,需要将大多数数据在节点间进行重新分布。这个操作会消耗大量的时间和带宽等资源,这在生产环境下是不可接受的。
算法原理
一致性 hash 算法的目的是在节点数量 n 变化时, 使尽可能少的 key 需要进行节点间重新分布。一致性 hash 算法将数据 key 和服务器地址 addr 散列到 2^32 的空间中。
我们将 2^32 个整数首尾相连形成一个环,首先计算服务器地址 addr 的 hash 值放置在环上。然后计算 key 的 hash 值放置在环上,顺时针查找,将数据放在找到的的第一个节点上。
在增加或删除节点时只有该节点附近的数据需要重新分布,从而解决了上述问题。
如果服务器节点较少则比较容易出现数据分布不均匀的问题,一般来说环上的节点越多数据分布越均匀。我们不需要真的增加一台服务器,只需要将实际的服务器节点映射为几个虚拟节点放在环上即可。
参考:https://www.cnblogs.com/Finley/p/14038398.html
lib/consistenthash/consistenthash.go
// hash函数定义,Go的hash函数就是这样定义的
type HashFunc func(data []byte) uint32
// 存储所有节点和节点的hash
type NodeMap struct {
hashFunc HashFunc
nodeHashs []int // 各个节点的hash值,顺序的
nodehashMap map[int]string // <hash, 节点>
}
func NewNodeMap(fn HashFunc) *NodeMap {
m := &NodeMap{
hashFunc: fn,
nodehashMap: make(map[int]string),
}
if m.hashFunc == nil {
m.hashFunc = crc32.ChecksumIEEE
}
return m
}
func (m *NodeMap) IsEmpty() bool {
return len(m.nodeHashs) == 0
}
// 添加节点到一致性哈希中
func (m *NodeMap) AddNode(keys ...string) {
for _, key := range keys {
if key == "" {
continue
}
hash := int(m.hashFunc([]byte(key)))
m.nodeHashs = append(m.nodeHashs, hash)
m.nodehashMap[hash] = key
}
sort.Ints(m.nodeHashs)
}
// 选择节点。使用二分查找,如果hash比nodeHashs中最大的hash还要大,idx=0
func (m *NodeMap) PickNode(key string) string {
if m.IsEmpty() {
return ""
}
hash := int(m.hashFunc([]byte(key)))
idx := sort.Search(len(m.nodeHashs), func(i int) bool {
return m.nodeHashs[i] >= hash
})
if idx == len(m.nodeHashs) {
idx = 0
}
return m.nodehashMap[m.nodeHashs[idx]]
}
HashFunc:hash函数定义,Go的hash函数就是这样定义的
NodeMap:存储所有节点和节点的hash
- nodeHashs:各个节点的hash值,顺序的
- nodehashMap<hash, 节点>
AddNode:添加节点到一致性哈希中
PickNode:选择节点。使用二分查找,如果hash比nodeHashs中最大的hash还要大,idx=0
database/standalone_database.go
type StandaloneDatabase struct {
dbSet []*DB
aofHandler *aof.AofHandler
}
func NewStandaloneDatabase() *StandaloneDatabase {
......
}
把database/database改名为database/standalone_database,再增加一个cluster_database用于对key的路由
resp/client/client.go
client:Redis客户端,具体看:https://www.cnblogs.com/Finley/p/14028402.html
// Client is a pipeline mode redis client
type Client struct {
conn net.Conn
pendingReqs chan *request // wait to send
waitingReqs chan *request // waiting response
ticker *time.Ticker
addr string
working *sync.WaitGroup // its counter presents unfinished requests(pending and waiting)
}
// request is a message sends to redis server
type request struct {
id uint64
args [][]byte
reply resp.Reply
heartbeat bool
waiting *wait.Wait
err error
}
const (
chanSize = 256
maxWait = 3 * time.Second
)
// MakeClient creates a new client
func MakeClient(addr string) (*Client, error) {
conn, err := net.Dial("tcp", addr)
if err != nil {
return nil, err
}
return