GO实现Redis：GO实现Redis集群（5）
[*]采用一致性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
type HashFunc func(data []byte) uint32
type NodeMap struct {
hashFunc HashFunc
nodeHashs []int
nodehashMap mapstring
}
func NewNodeMap(fn HashFunc) *NodeMap {
m := &NodeMap{
hashFunc: fn,
nodehashMap: make(mapstring),
}
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 = key
}
sort.Ints(m.nodeHashs)
}
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 >= hash
})
if idx == len(m.nodeHashs) {
idx = 0
}
return m.nodehashMap]
}HashFunc：hash函数定义，Go的hash函数就是这样定义的
NodeMap：存储所有节点和节点的hash
[*]nodeHashs：各个节点的hash值，顺序的
[*]nodehashMap
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 is a pipeline mode redis clienttype 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 servertype 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 clientfunc MakeClient(addr string) (*Client, error) { conn, err := net.Dial("tcp", addr) if err != nil { return nil, err } return &Client{ addr: addr, conn: conn, pendingReqs: make(chan *request, chanSize), waitingReqs: make(chan *request, chanSize), working: &sync.WaitGroup{}, }, nil}// Start starts asynchronous goroutinesfunc (client *Client) Start() { client.ticker = time.NewTicker(10 * time.Second) go client.handleWrite() go func() { err := client.handleRead() if err != nil { logger.Error(err) } }() go client.heartbeat()}// Close stops asynchronous goroutines and close connectionfunc (client *Client) Close() { client.ticker.Stop() // stop new request close(client.pendingReqs) // wait stop process client.working.Wait() // clean _ = client.conn.Close() close(client.waitingReqs)}func (client *Client) handleConnectionError(err error) error { err1 := client.conn.Close() if err1 != nil { if opErr, ok := err1.(*net.OpError); ok { if opErr.Err.Error() != "use of closed network connection" { return err1 } } else { return err1 } } conn, err1 := net.Dial("tcp", client.addr) if err1 != nil { logger.Error(err1) return err1 } client.conn = conn go func() { _ = client.handleRead() }() return nil}func (client *Client) heartbeat() { for range client.ticker.C { client.doHeartbeat() }}func (client *Client) handleWrite() { for req := range client.pendingReqs { client.doRequest(req) }}// Send sends a request to redis serverfunc (client *Client) Send(args [][]byte) resp.Reply { request := &request{ args: args, heartbeat: false, waiting: &wait.Wait{}, } request.waiting.Add(1) client.working.Add(1) defer client.working.Done() client.pendingReqs
页:
[1]