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

1. type HashFunc func(data []byte) uint32
3. type NodeMap struct {
4.    hashFunc    HashFunc
5.    nodeHashs   []int         
6.    nodehashMap map[int]string
7. }
9. func NewNodeMap(fn HashFunc) *NodeMap {
10.    m := &NodeMap{
11.       hashFunc:    fn,
12.       nodehashMap: make(map[int]string),
13.    }
14.    if m.hashFunc == nil {
15.       m.hashFunc = crc32.ChecksumIEEE
16.    }
17.    return m
18. }
20. func (m *NodeMap) IsEmpty() bool {
21.    return len(m.nodeHashs) == 0
22. }
24. func (m *NodeMap) AddNode(keys ...string) {
25.    for _, key := range keys {
26.       if key == "" {
27.          continue
28.       }
29.       hash := int(m.hashFunc([]byte(key)))
30.       m.nodeHashs = append(m.nodeHashs, hash)
31.       m.nodehashMap[hash] = key
32.    }
33.    sort.Ints(m.nodeHashs)
34. }
36. func (m *NodeMap) PickNode(key string) string {
37.    if m.IsEmpty() {
38.       return ""
39.    }
41.    hash := int(m.hashFunc([]byte(key)))
43.    
44.    idx := sort.Search(len(m.nodeHashs), func(i int) bool {
45.       return m.nodeHashs[i] >= hash
46.    })
48.    
49.    if idx == len(m.nodeHashs) {
50.       idx = 0
51.    }
53.    return m.nodehashMap[m.nodeHashs[idx]]
54. }

复制代码

HashFunc：hash函数定义，Go的hash函数就是这样定义的
NodeMap：存储所有节点和节点的hash

• nodeHashs：各个节点的hash值，顺序的
  
• nodehashMap
  

AddNode：添加节点到一致性哈希中
PickNode：选择节点。使用二分查找，如果hash比nodeHashs中最大的hash还要大，idx=0


database/standalone_database.go

1. type StandaloneDatabase struct {
2.    dbSet []*DB
3.    aofHandler *aof.AofHandler
4. }
6. func NewStandaloneDatabase() *StandaloneDatabase {
7.   ......
8. }

复制代码

把database/database改名为database/standalone_database，再增加一个cluster_database用于对key的路由


resp/client/client.go
[code]// 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