一致性哈希算法在分佈緩存中的應用

時間 2019-11-10

標籤一致性哈希算法分佈緩存應用欄目 Zookeeper 简体版

原文原文鏈接

1、簡介node

關於一致性哈希算法介紹有許多相似文章，須要把一些理論轉爲爲本身的知識，因此有了這篇文章，本文部分實現也參照了原有的一些方法。算法

該算法在分佈緩存的主機選擇中很經常使用，詳見 http://en.wikipedia.org/wiki/Consistent_hashing 。數據庫

2、算法誕生原因緩存

如今許多大型系統都離不開緩存(K/V)（因爲高併發等因素照成的數據庫壓力(或磁盤IO等)超負荷，須要緩存緩解壓力），爲了得到良好的水平擴展性，併發

緩存主機互相不通訊（如Mencached），經過客戶端計算Key而獲得數據存放的主機節點，最簡單的方式是取模，假如：負載均衡

----------------------------------------------------------------dom

如今有3臺緩存主機，如今有一個key 爲 cks 的數據須要存儲：ide

key = "cks"函數

hash(key) = 10高併發

10 % 3 = 1 ---> 則表明選擇第一臺主機存儲這個key和對應的value。

缺陷：

假若有一臺主機宕機或增長一臺主機（必須考慮的狀況），取模的算法將致使大量的緩存失效（計算到其餘沒有緩存該數據的主機），數據庫等忽然承受巨大負荷，很大可能致使DB服務不可用等。

----------------------------------------------------------------

3、一致性哈希算法原理

該算法須要解決取模方法當增長主機或者宕機時帶來的大量緩存抖動問題，要在生產環境中使用，算法需具有如下幾個特色：

1. 平衡性：指緩存數據儘可能平衡分佈到全部緩存主機上，有效利用每臺主機的空間。

2. 單調性：當出現宕機或有新主機加入時，儘可能保護原有的緩存能繼續命中，即少許緩存須要從新分配。

3. 負載均衡：每臺緩存主機儘可能平衡分擔壓力，即Key的分配比例在這些主機中應趨於平衡。

假如咱們把鍵hash爲int類型（32字節），取值範圍爲 -2^31 到 (2^31-1) ，咱們把這些值首尾相連造成一個圓環，以下圖：

假設如今有3臺緩存主機： C0一、C0二、C03 ，把它們放在環上(經過IP hash，後面實現會介紹)，以下圖：

假如如今有5個key須要緩存，它們分別爲 A,B,C,D,E，假設它們通過hash後分布以下，順時針找到它們最近的主機，並存儲在上面：

假如當新加入節點C04的時候，數據分配以下：

能夠發現，只有少許緩存被從新分配的新主機，減小抖動帶來的壓力。

但同時出現一個問題，數據分佈並不儘可能均勻（當有大量緩存的時候能夠看出來），這時候須要把真實的緩存節點虛擬爲多個節點，分佈在環上，

當順時針找到虛擬節點的時候再映射到真實節點，則能夠知道數據緩存在哪臺主機。

4、算法實現（Java版本）

算法的實現有許多，下面例子僅供參考，實際仍須要考慮其餘多個問題：

假設有4主機：

192.168.70.1-5

public class Node {
    
    private String ip;
    // 表明主機中存放的K/V
    private ConcurrentMap<Object, Object> map = new ConcurrentHashMap<Object, Object>();
    
    public Node(String ip) {
        this.ip = ip;
    }
    public String getIp() {
        return ip;
    }    
    public ConcurrentMap<Object, Object> getMap() {
        return map;
    }
    @Override
    public String toString() {
        return ip;
    }
}

下面是一個沒有虛擬節點的狀況實現方法：

public class ConsistentHash {
    
    // -2^31 - (2^31-1) 圓環, 用於存儲節點
    private final SortedMap<Integer, Node> circle = new TreeMap<Integer, Node>();
    
    private IHash hashIf;
    
    public ConsistentHash(IHash hash) {
        this.hashIf = hash;
    }
    
    public void addNode(Node node) {
        circle.put(hashIf.hash(node.getIp()), node);
    }
    
    public void removeNode(Node node) {
        circle.remove(hashIf.hash(node.getIp()));
    }
    
    public Node getNode(Object key) {
        int hashCode = hashIf.hash(key);
        if (!circle.containsKey(hashCode)) {
            // 相似順時針取得最近的存儲節點
            SortedMap<Integer, Node> tailMap = circle.tailMap(hashCode);
            hashCode = tailMap.isEmpty()? circle.firstKey() : tailMap.firstKey();
        }
        return circle.get(hashCode);
    }
}

其中IHash 爲散列方法接口，可實現不一樣的散列方式，下面是一個基於MD5算法獲得的int值（還有其餘算法）：

public interface IHash {
    
    int hash(Object key);
    
}

public class MD5HashImpl implements IHash {
    
    MessageDigest digest;
    
    public MD5HashImpl() throws NoSuchAlgorithmException {
        digest = MessageDigest.getInstance("MD5");
    }
    
    @Override
    public int hash(Object key) {
        if (key == null) return 0;
        
        int h = key.hashCode();
        byte[] bytes = new byte[4];  
        for(int i=3; i>-1; i--) {
            bytes[i] = (byte)( h>>(i*8) );
        } 
        
        byte[] hashBytes ;
        synchronized (digest) {
            hashBytes = digest.digest(bytes);
        }
        
        int result = 0;
        for (int i=0; i<4; i++) {
            int idx = i*4;
            result += (hashBytes[idx + 3]&0xFF << 24)
                    | (hashBytes[idx + 2]&0xFF << 16)
                    | (hashBytes[idx + 1]&0xFF << 8)
                    | (hashBytes[idx + 0]&0xFF);
        }
        return result;
    }
}

測試方法以下：

public class ConsistentHashTest {

    public static void main(String[] args) throws Exception {
        ConsistentHash cHash = new ConsistentHash(new MD5HashImpl());
        
        // Nodes
        List<Node> nodes = new ArrayList<Node>();
        for (int i=1; i<5; i++) {
            Node node = new Node("192.168.70." + i); // Fake
            nodes.add(node);
            cHash.addNode(node);
        }
        
        Map<String, Set<Integer>> counter = new HashMap<String, Set<Integer>>();
        for (Node n : nodes) {
            counter.put(n.getIp(), new HashSet<Integer>());
        }
        
        // 隨機KEY測試分佈狀況
        Set<Integer> allKeys = new HashSet<Integer>();
　　　　　Random random = new Random();
        int testTimes = 1000000;
        for (int i=0; i<testTimes; i++) {
            int randomInt = random.nextInt();
            Node node = cHash.getNode(randomInt);
            Set<Integer> count = counter.get(node.getIp());
            count.add(randomInt);
　　　　　　  allKeys.add(randomInt);
        }
        
        for (Map.Entry<String, Set<Integer>> entry : counter.entrySet()) {
            System.out.println(entry.getKey() + "\t" + entry.getValue().size() 
                    + "\t" + (entry.getValue().size()*100/(float)allKeys.size()) + "%");
        }
    }
}

測試結果（每次運行的實際結果不一樣）：

IP Count Percent

--------------------------------------
192.168.70.1 216845 21.6845%
192.168.70.4 7207 0.7207%
192.168.70.2 749929 74.9929%
192.168.70.3 25891 2.5891%

-------------------------------------

這結果表示每一個節點分配並不均勻，須要把每一個節點虛擬爲多個節點，ConsistentHash 算法更改以下：

public class ConsistentHash {
    
    // -2^31 - (2^31-1) 圓環, 用於存儲節點
    private final SortedMap<Integer, Node> circle = new TreeMap<Integer, Node>();
    
    private IHash hashIf;
    private int virtualNum; // 把實際節點虛擬爲多個節點
    
    public ConsistentHash(IHash hash, int virtualNum) {
        this.hashIf = hash;
        this.virtualNum = virtualNum;
    }
    
    public void addNode(Node node) {
        for (int i=0; i<virtualNum; i++) {
            circle.put(hashIf.hash(i + node.getIp()), node);
        }
    }
    
    public void removeNode(Node node) {
        for (int i=0; i<virtualNum; i++) {
            circle.remove(hashIf.hash(i + node.getIp()));
        }
    }
    
    public Node getNode(Object key) {
        int hashCode = hashIf.hash(key);
        if (!circle.containsKey(hashCode)) {
            // 相似順時針取得最近的存儲節點
            SortedMap<Integer, Node> tailMap = circle.tailMap(hashCode);
            hashCode = tailMap.isEmpty()? circle.firstKey() : tailMap.firstKey();
        }
        return circle.get(hashCode);
    }
}

只須要在測試函數裏面修改：

// 這裏每一個節點虛擬爲120個，根據實際狀況考慮修改合理的值，虛擬數量少則致使部分不均勻，數量大則致使樹的查找效率下降，二者須要權衡。

ConsistentHash cHash = new ConsistentHash(new MD5HashImpl(), 120);

--------------------------- 添加虛擬節點後的分佈狀況 ----------------------------------------

IP Count Percent
192.168.70.1 277916 27.7916%
192.168.70.4 251437 25.1437%
192.168.70.2 226645 22.6645%
192.168.70.3 243871 24.3871%

------------------------------------------------------------------------------------------------

下面再模擬宕機和新增主機狀況下面的緩存失效率：

public class HitFailureTest {

    public static void main(String[] args) throws Exception {
        ConsistentHash cHash = new ConsistentHash(new MD5HashImpl(), 120);
        
        List<Node> nodes = new ArrayList<Node>();
        for (int i=1; i<5; i++) {
            Node node = new Node("192.168.70." + i); // Fake
            nodes.add(node);
            cHash.addNode(node);
        }
        
        Set<Integer> allKeys = new HashSet<Integer>();
        
        Random random = new Random();
        int testTimes = 1000000;
        for (int i=0; i<testTimes; i++) {
            int randomInt = random.nextInt();
            cHash.getNode(randomInt).getMap().put(randomInt, 0);
            allKeys.add(randomInt);
        }
        
        // 移除主機序號
        int removeIdx = 1;
        cHash.removeNode(nodes.get(removeIdx));
        
        int failureCount = 0;
        for (Integer key : allKeys) {
            if(!cHash.getNode(key).getMap().containsKey(key)) {
                failureCount ++;
            }
        }
        System.out.println("FailureCount \t Percent");
        System.out.println(failureCount + "\t" + (failureCount*100/(float)allKeys.size()) + "%");
    }

}