一致性哈希算法原理分析及實現

        一致性哈希算法常用於負載均衡中要求資源被均勻的分布到所有節點上，並且對資源的請求能快速路由到對應的節點上。具體的舉兩個場景的例子：

        1、MemCache集群，要求存儲各種數據均勻的存到集群中的各個節點上，訪問這些數據時能快速的路由到集群中對應存放該數據的節點上；並且要求增刪節點對整個集群的影響很小，不至於有大的動盪造成整體負載的不穩定；

        2、RPC過程中服務提供者做N個節點的集群部署，為了能在服務上維護一些業務狀態，希望同一種請求每次都落到同一台服務上。

        比如有{N₀, N₁, N₂}三個節點，陸續有多個資源要分配到這三個節點上，如何盡可能均勻的分配到這些節點上？

        一致性哈希算法的思路為：先構造出一個長度為2³²整數環，根據N_0-3的節點名稱的hash值（分布為[0,2³²-1]）放到這個環上。現在要存放資源，根據資源的Key的Hash值（也是分布為[0,2³²-1]）值H_aaa，在環上順時針的找到離H_aaa最近（第一個大於或等於Haaa）的一個節點，就建立了資源和節點的映射關系。

        

        以上圖片引自http://www.cnblogs.com/xrq730/p/4948707.html。

        為什么要用環存儲節點，並用hashKey順時針尋找對應節點？

        我們分配節點最簡單的辦法是取余算法，即有3個節點，資源key=5, 5%3=2，選取N₂，key=3，3%3=0，選取N₀。雖然簡單，但有個缺點，如果節點數增加或減少，就會有大量的key不命中，造成請求壓力轉移，可能對系統整體有很大的影響，甚至發生宕機危險。

        而一致性哈希算法增加或減少節點，只會引起少部分key不命中，如下圖，增加一個Node4節點，只會將加粗部分的key值從Node1(10.0.0.0:91002)移到Node4(10.0.0.0:91003)，對集群影響很小。

        以上圖片引自http://www.cnblogs.com/xrq730/p/4948707.html

        Java實現中用什么表示Hash環好呢？經對比，用TreeMap的時間復雜度是O(logN)，相對效率比較高，因為TreeMap使用了紅黑樹結構存儲實體對象。

        Hash算法的選擇上，首先我們考慮簡單的String.HashCode()方法，這個算法的缺點是，相似的字符串如N1(10.0.0.0:91001)，N2(10.0.0.0:91002)，N3(10.0.0.0:91003)，哈希值也很相近，造成的結果是節點在Hash環上分布很緊密，導致大部分Key值落到了N0上，節點資源分布不均。一般我們采用FNV1_32_HASH、KETAMA_HASH等算法，KETAMA_HASH是MemCache集群默認的實現方法，這些算法效果要好得多，會使N0，N1，N2的Hash值更均勻的分布在環上。

        我們用KETAMA_HASH算法實現一致性哈希（無虛擬節點方式），如下代碼所示：        

package com.example.demo.arithmetic;

import java.io.UnsupportedEncodingException;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.Arrays;
import java.util.Map;
import java.util.TreeMap;

/**
 * Created by markcd on 2018/2/28.
 */
public class ConsistentHashLoadBalanceNoVirtualNode {

    private TreeMap<Long, String> realNodes = new TreeMap<>();
    private String[] nodes;

    public ConsistentHashLoadBalanceNoVirtualNode(String[] nodes){
        this.nodes = Arrays.copyOf(nodes, nodes.length);
        initalization();
    }

    /**
     * 初始化哈希環
     * 循環計算每個node名稱的哈希值，將其放入treeMap
     */
    private void initalization(){
        for (String nodeName: nodes) {
            realNodes.put(hash(nodeName, 0), nodeName);
        }
    }

    /**
     * 根據資源key選擇返回相應的節點名稱
     * @param key
     * @return 節點名稱
     */
    public String selectNode(String key){
        Long hashOfKey = hash(key, 0);
        if (! realNodes.containsKey(hashOfKey)) {
               //ceilingEntry()的作用是得到比hashOfKey大的第一個Entry 41             Map.Entry<Long, String> entry = realNodes.ceilingEntry(hashOfKey);
            if (entry != null)
                return entry.getValue();
            else
                return nodes[0];
        }else
            return realNodes.get(hashOfKey);
    }

    private Long hash(String nodeName, int number) {
        byte[] digest = md5(nodeName);
        return (((long) (digest[3 + number * 4] & 0xFF) << 24)
                | ((long) (digest[2 + number * 4] & 0xFF) << 16)
                | ((long) (digest[1 + number * 4] & 0xFF) << 8)
                | (digest[number * 4] & 0xFF))
                & 0xFFFFFFFFL;
    }

    /**
     * md5加密
     *
     * @param str
     * @return
     */
    public byte[] md5(String str) {
        try {
            MessageDigest md = MessageDigest.getInstance("MD5");
            md.reset();
            md.update(str.getBytes("UTF-8"));
            return md.digest();
        } catch (NoSuchAlgorithmException e) {
            e.printStackTrace();
            return null;
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
            return null;
        }
    }

    private void printTreeNode(){
        if (realNodes != null && ! realNodes.isEmpty()){
            realNodes.forEach((hashKey, node) ->
                    System.out.println(
                            new StringBuffer(node)
                            .append(" ==> ")
                            .append(hashKey)
                    )
            );
        }else
            System.out.println("Cycle is Empty");
    }

    public static void main(String[] args){
        String[] nodes = new String[]{"192.168.2.1:8080", "192.168.2.2:8080", "192.168.2.3:8080", "192.168.2.4:8080"};
        ConsistentHashLoadBalanceNoVirtualNode consistentHash = new ConsistentHashLoadBalanceNoVirtualNode(nodes);
        consistentHash.printTreeNode();
    }
}

 

        main()方法執行結果如下，可以看到，hash值分布的距離比較開闊。

        192.168.2.3:8080 ==> 1182102228

        192.168.2.4:8080 ==> 1563927337

        192.168.2.1:8080 ==> 2686712470

        192.168.2.2:8080 ==> 3540412423

        KETAMA_HASH解決了hash值分布不均的問題，但還存在一個問題，如下圖，在沒有Node3節點時，資源相對均勻的分布在{Node0,Node1,Node2}上。增加了Node3節點后，Node1到Node3節點中間的所有資源從Node2遷移到了Node3上。這樣，Node0,Node1存儲的資源多，Node2,Node3存儲的資源少，資源分布不均勻。

        以上圖片引自http://www.cnblogs.com/xrq730/p/4948707.html

        如何解決這個問題呢？我們引入虛擬節點概念，如將一個真實節點Node0映射成100個虛擬節點分布在Hash環上，與這100個虛擬節點根據KETAMA_HASH哈希環匹配的資源都存到真實節點Node0上。{Node0,Node1,Node2}以相同的方式拆分虛擬節點映射到Hash環上。當集群增加節點Node3時，在Hash環上增加Node3拆分的100個虛擬節點，這新增的100個虛擬節點更均勻的分布在了哈希環上，可能承擔了{Node0,Node1,Node2}每個節點的部分資源，資源分布仍然保持均勻。

        每個真實節點應該拆分成多少個虛擬節點？數量要合適才能保證負載分布的均勻，有一個大致的規律，如下圖所示，Y軸表示真實節點的數目，X軸表示需拆分的虛擬節點數目：

        

        真實節點越少，所需闡發的虛擬節點越多，比如有10個真實節點，每個節點所需拆分的虛擬節點個數可能是100~200個，才能達到真正的負載均衡。

        下面貼出使用了虛擬節點的算法實現：        

package com.example.demo.arithmetic;

import java.io.UnsupportedEncodingException;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.LinkedList;
import java.util.Map;
import java.util.TreeMap;

/**
 * Created by markcd on 2018/2/28.
 */
public class ConsistentHashLoadBalance {

    private TreeMap<Long, String> virtualNodes = new TreeMap<>();
    private LinkedList<String> nodes;
        //每個真實節點對應的虛擬節點數  17     private final int replicCnt;

    public ConsistentHashLoadBalance(LinkedList<String> nodes, int replicCnt){
        this.nodes = nodes;
        this.replicCnt = replicCnt;
        initalization();
    }

    /**
     * 初始化哈希環
     * 循環計算每個node名稱的哈希值，將其放入treeMap
     */
    private void initalization(){
        for (String nodeName: nodes) {
            for (int i = 0; i < replicCnt/4; i++) {
                String virtualNodeName = getNodeNameByIndex(nodeName, i);
                for (int j = 0; j < 4; j++) {
                    virtualNodes.put(hash(virtualNodeName, j), nodeName);
                }
            }
        }
    }

    private String getNodeNameByIndex(String nodeName, int index){
        return new StringBuffer(nodeName)
                .append("&&")
                .append(index)
                .toString();
    }

    /**
     * 根據資源key選擇返回相應的節點名稱
     * @param key
     * @return 節點名稱
     */
    public String selectNode(String key){
        Long hashOfKey = hash(key, 0);
        if (! virtualNodes.containsKey(hashOfKey)) {
            Map.Entry<Long, String> entry = virtualNodes.ceilingEntry(hashOfKey);
            if (entry != null)
                return entry.getValue();
            else
                return nodes.getFirst();
        }else
            return virtualNodes.get(hashOfKey);
    }

    private Long hash(String nodeName, int number) {
        byte[] digest = md5(nodeName);
        return (((long) (digest[3 + number * 4] & 0xFF) << 24)
                | ((long) (digest[2 + number * 4] & 0xFF) << 16)
                | ((long) (digest[1 + number * 4] & 0xFF) << 8)
                | (digest[number * 4] & 0xFF))
                & 0xFFFFFFFFL;
    }

    /**
     * md5加密
     *
     * @param str
     * @return
     */
    public byte[] md5(String str) {
        try {
            MessageDigest md = MessageDigest.getInstance("MD5");
            md.reset();
            md.update(str.getBytes("UTF-8"));
            return md.digest();
        } catch (NoSuchAlgorithmException e) {
            e.printStackTrace();
            return null;
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
            return null;
        }
    }

    public void addNode(String node){
        nodes.add(node);
        String virtualNodeName = getNodeNameByIndex(node, 0);
        for (int i = 0; i < replicCnt/4; i++) {
            for (int j = 0; j < 4; j++) {
                virtualNodes.put(hash(virtualNodeName, j), node);
            }
        }
    }

    public void removeNode(String node){
        nodes.remove(node);
        String virtualNodeName = getNodeNameByIndex(node, 0);
        for (int i = 0; i < replicCnt/4; i++) {
            for (int j = 0; j < 4; j++) {
                virtualNodes.remove(hash(virtualNodeName, j), node);
            }
        }
    }

    private void printTreeNode(){
        if (virtualNodes != null && ! virtualNodes.isEmpty()){
            virtualNodes.forEach((hashKey, node) ->
                    System.out.println(
                            new StringBuffer(node)
                                    .append(" ==> ")
                                    .append(hashKey)
                    )
            );
        }else
            System.out.println("Cycle is Empty");
    }

    public static void main(String[] args){
        LinkedList<String> nodes = new LinkedList<>();
        nodes.add("192.168.2.1:8080");
        nodes.add("192.168.2.2:8080");
        nodes.add("192.168.2.3:8080");
        nodes.add("192.168.2.4:8080");
        ConsistentHashLoadBalance consistentHash = new ConsistentHashLoadBalance(nodes, 160);
        consistentHash.printTreeNode();
    }
}

         以上main方法執行的結果如下：

         192.168.2.4:8080 ==> 18075595
         192.168.2.1:8080 ==> 18286704
         192.168.2.1:8080 ==> 35659769
         192.168.2.2:8080 ==> 43448858
         192.168.2.1:8080 ==> 44075453
         192.168.2.3:8080 ==> 47625378
         ........(由於內容過多，不做全部展示)

 

        至此哈希一致性算法的原理和實現描述完畢，歡迎大家討論，如有不當的地方歡迎大家提出異議。