GO实现Redis：GO实现Redis集群（5）

您所在的位置：网站首页 › redis集群hash槽 › GO实现Redis：GO实现Redis集群（5）

GO实现Redis：GO实现Redis集群（5）

#GO实现Redis：GO实现Redis集群（5）| 来源: 网络整理| 查看: 265

采用一致性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 map[int]string } 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) } 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

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 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 &Client{ addr: addr, conn: conn, pendingReqs: make(chan *request, chanSize), waitingReqs: make(chan *request, chanSize), working: &sync.WaitGroup{}, }, nil } // Start starts asynchronous goroutines func (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 connection func (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 server func (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

【本文地址】

GO实现Redis：GO实现Redis集群（5）

GO实现Redis：GO实现Redis集群（5）

今日新闻

推荐新闻