Golang 實現 Redis(7): Redis 集羣與一致性 Hash

本文是使用 golang 實現 redis 系列的第七篇， 將介紹如何將單點的緩存服務器擴展爲分佈式緩存。godis 集羣的源碼在Github:Godis/cluster

單臺服務器的CPU和內存等資源老是有限的，隨着數據量和訪問量的增長單臺服務器很容易遇到瓶頸。利用多臺機器創建分佈式系統，分工處理是提升系統容量和吞吐量的經常使用方法。

使用更多機器來提升系統容量的方式稱爲系統橫向擴容。與之相對的，提升單臺機器性能被稱爲縱向擴容。因爲沒法在單臺機器上無限提升硬件配置且硬件價格與性能的關係並不是線性的，因此創建分佈式系統進行橫向擴容是更爲經濟實用的選擇。

咱們採用一致性 hash 算法 key 分散到不一樣的服務器，客戶端能夠鏈接到服務集羣中任意一個節點。當節點須要訪問的數據不在本身本地時，須要經過一致性 hash 算法計算出數據所在的節點並將指令轉發給它。

與分佈式系統理論中的分區容錯性不一樣，咱們僅將數據存在一個節點沒有保存副本。這種設計提升了系統吞吐量和容量，可是並無提升系統可用性，當有一個節點崩潰時它保存的數據將沒法訪問。

生產環境實用的 redis 集羣一般也採起相似的分片存儲策略，併爲每一個節點配置從節點做爲熱備節點，並使用 sentinel 機制監控 master 節點狀態。在 master 節點崩潰後，sentinel 將備份節點提高爲 master 節點以保證可用性。

一致性 hash 算法

爲何須要一致性 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 值放置在環上，順時針查找，將數據放在找到的的第一個節點上。

key1, key2 和 key5 在 node2 上，key 3 在 node4 上，key4 在 node6 上

在增長或刪除節點時只有該節點附近的數據須要從新分佈，從而解決了上述問題。

新增 node8 後，key 5 從 node2 轉移到 node8。其它 key 不變

若是服務器節點較少則比較容易出現數據分佈不均勻的問題，通常來講環上的節點越多數據分佈越均勻。咱們不須要真的增長一臺服務器，只須要將實際的服務器節點映射爲幾個虛擬節點放在環上便可。

Golang 實現一致性 Hash

咱們使用 Golang 實現一致性 hash 算法, 源碼在 Github: HDT3213/Godis, 大約 80 行代碼。

type HashFunc func(data []byte) uint32

type Map struct {
    hashFunc HashFunc
    replicas int
    keys     []int // sorted
    hashMap  map[int]string
}

func New(replicas int, fn HashFunc) *Map {
    m := &Map{
        replicas: replicas, // 每一個物理節點會產生 replicas 個虛擬節點
        hashFunc: fn,
        hashMap:  make(map[int]string), // 虛擬節點 hash 值到物理節點地址的映射
    }
    if m.hashFunc == nil {
        m.hashFunc = crc32.ChecksumIEEE
    }
    return m
}

func (m *Map) IsEmpty() bool {
    return len(m.keys) == 0
}

接下來實現添加物理節點的 Add 方法:

func (m *Map) Add(keys ...string) {
    for _, key := range keys {
        if key == "" {
            continue
        }
        for i := 0; i < m.replicas; i++ {
            // 使用 i + key 做爲一個虛擬節點，計算虛擬節點的 hash 值
            hash := int(m.hashFunc([]byte(strconv.Itoa(i) + key))) 
            // 將虛擬節點添加到環上
            m.keys = append(m.keys, hash) 
            // 註冊虛擬節點到物理節點的映射
            m.hashMap[hash] = key
        }
    }
    sort.Ints(m.keys)
}

接下來實現查找算法:

func (m *Map) Get(key string) string {
    if m.IsEmpty() {
        return ""
    }

    // 支持根據 key 的 hashtag 來肯定分佈 
    partitionKey := getPartitionKey(key)
    hash := int(m.hashFunc([]byte(partitionKey)))

    // sort.Search 會使用二分查找法搜索 keys 中知足 m.keys[i] >= hash 的最小 i 值
    idx := sort.Search(len(m.keys), func(i int) bool { return m.keys[i] >= hash })

    // 若 key 的 hash 值大於最後一個虛擬節點的 hash 值，則 sort.Search 找不到目標
    // 這種狀況下選擇第一個虛擬節點
    if idx == len(m.keys) {
        idx = 0
    }

    // 將虛擬節點映射爲實際地址
    return m.hashMap[m.keys[idx]]
}

實現集羣

實現了一致性 hash 算法後咱們能夠着手實現集羣模式了，Godis 集羣的代碼在 Github:Godis/cluster。

集羣最核心的邏輯是找到 key 所在節點並將指令轉發過去:

// 集羣模式下，除了 MSet、DEL 等特殊指令外，其它指令會交由 defaultFunc 處理
func defaultFunc(cluster *Cluster, c redis.Connection, args [][]byte) redis.Reply {
    key := string(args[1])
    peer := cluster.peerPicker.Get(key) // 經過一致性 hash 找到節點
    return cluster.Relay(peer, c, args)
}

func (cluster *Cluster) Relay(peer string, c redis.Connection, args [][]byte) redis.Reply {
    if peer == cluster.self { // 若數據在本地則直接調用數據庫引擎
        // to self db
        return cluster.db.Exec(c, args)
    } else {
        // 從鏈接池取一個與目標節點的鏈接
        // 鏈接池使用 github.com/jolestar/go-commons-pool/v2 實現
        peerClient, err := cluster.getPeerClient(peer) 
        if err != nil {
            return reply.MakeErrReply(err.Error())
        }
        defer func() {
            _ = cluster.returnPeerClient(peer, peerClient) // 處理完成後將鏈接放回鏈接池
        }()
        // 將指令發送到目標節點
        return peerClient.Send(args) 
    }
}

func (cluster *Cluster) getPeerClient(peer string) (*client.Client, error) {
    connectionFactory, ok := cluster.peerConnection[peer]
    if !ok {
        return nil, errors.New("connection factory not found")
    }
    raw, err := connectionFactory.BorrowObject(context.Background())
    if err != nil {
        return nil, err
    }
    conn, ok := raw.(*client.Client)
    if !ok {
        return nil, errors.New("connection factory make wrong type")
    }
    return conn, nil
}

func (cluster *Cluster) returnPeerClient(peer string, peerClient *client.Client) error {
    connectionFactory, ok := cluster.peerConnection[peer]
    if !ok {
        return errors.New("connection factory not found")
    }
    return connectionFactory.ReturnObject(context.Background(), peerClient)
}