一致性hash原理与实现

一致性hash算法原理：

把数据用hash函数（如MD5），映射到一个很大的空间里，如图所示。数据的存储时，先得到一个hash值，对应到这个环中的每个位置，如k1对应到了图中所示的位置，然后沿顺时针找到一个机器节点B，将k1存储到B这个节点中。

如果B节点宕机了，则B上的数据就会落到C节点上，如下图所示：

这样，只会影响C节点，对其他的节点A，D的数据不会造成影响。然而，这又会造成一个“雪崩”的情况，即C节点由于承担了B节点的数据，所以C节点的负载会变高，C节点很容易也宕机，这样依次下去，这样造成整个集群都挂了。

       为此，引入了“虚拟节点”的概念：即把想象在这个环上有很多“虚拟节点”，数据的存储是沿着环的顺时针方向找一个虚拟节点，每个虚拟节点都会关联到一个真实节点，如下图所使用：

图中的A1、A2、B1、B2、C1、C2、D1、D2都是虚拟节点，机器A负载存储A1、A2的数据，机器B负载存储B1、B2的数据，机器C负载存储C1、C2的数据。由于这些虚拟节点数量很多，均匀分布，因此不会造成“雪崩”现象。

应用举例：

    1、在memcache集群中使用一致性hash算法可以任意添加和删除服务节点，使其只有一个节点受影响，而且可以将缓存平均的存在各个节点上，不会导致雪崩现象。

    2、在提供socket服务集群中有多台服务器，客户端在建立连接时可以使用一致性hash来选择一个服务器进行链接，可以有效地平衡各个服务器的压力。

代码实现：

import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;


public class ConsistentHash<T> {

    private final int numberOfReplicas;
    private volatile TreeMap<Integer, List<T>> circle = new TreeMap<>();

    private static final int circleSize = 188833;

    /***
     * 
     * @param numberOfReplicas
     * 	每个节点的虚拟节点的个数
     * @param nodes
     */
    public ConsistentHash(int numberOfReplicas, Collection<T> nodes) {
        this.numberOfReplicas = numberOfReplicas;
        for (T node : nodes) {
            addNode(circle, node);
        }
    }

    public synchronized void add(T node) {
        TreeMap<Integer, List<T>> newCircle = copyCircle();
        addNode(newCircle, node);
        this.circle = newCircle;
    }

    public synchronized void remove(T node)	{
        TreeMap<Integer, List<T>> newCircle = copyCircle();
        remove(newCircle, node);
        this.circle = newCircle;
    }

    private TreeMap<Integer, List<T>> copyCircle() {
        TreeMap<Integer, List<T>> newTree = new TreeMap<>();

        for (Map.Entry<Integer, List<T>> entry : circle.entrySet())	{
            List<T> list = new ArrayList<T>();
            list.addAll(entry.getValue());
            newTree.put(entry.getKey(), list);
        }
        return newTree;
    }

    private void addNode(TreeMap<Integer, List<T>> circle, T node) {
        for (int i = 0; i < numberOfReplicas; i++) {
            int key = hashMd5(node.toString() + i);
            List<T> list = circle.get(key);
            if (list == null) {
                list = new ArrayList<T>();
                circle.put(key, list);
            }
            if (!containsNode(list, node)) {
                list.add(node);
            }
        }
    }

    private void removeNodeToList(List<T> list, T node)	{
        Iterator<T> it = list.iterator();
        while (it.hasNext()) {
            if (node.equals(it.next())) {
                it.remove();
            }
        }
    }

    private boolean containsNode(List<T> list, T node) {
        for (T t : list) {
            if (t.equals(node))	{
                return true;
            }
        }
        return false;
    }
    
    private void remove(TreeMap<Integer, List<T>> circle, T node) {
        for (int i = 0; i < numberOfReplicas; i++) {
            int key = hashMd5(node.toString() + i);
            List<T> list = circle.get(key);
            if (list != null) {
                if (list.contains(node)) {
                    removeNodeToList(list, node);
                }
                if (list.isEmpty())	{
                    circle.remove(key);
                }
            }
        }
    }

    public T get(Object key) {
        if (circle.isEmpty()) {
            return null;
        }
        int hash = hashMd5(key);
        Map.Entry<Integer, List<T>> entry = circle.ceilingEntry(hash);    //返回键值对，该键至少大于或等于给定键，如果不存在这样的键的键 - 值映射，则返回null相关联。
        List<T> node = null;
        if (entry == null) {
            node = circle.firstEntry().getValue();
        }
        else {
            node = entry.getValue();
        }
        if (node != null && !node.isEmpty()) {
            return node.get(0);
        }
        return null;
    }

    private static int hashCode(byte[] bytes) {
        int hash = 0;
        for (byte b : bytes) {
            hash = hash * 31 + ((int) b & 0xFF);
            if (hash > 0x4000000) {
                hash = hash % 0x4000000;
            }
        }
        return hash;
    }

    private  int hashMd5(Object o) {
        MessageDigest md;
        try {
            md = MessageDigest.getInstance("MD5");
            byte[] bytes = md.digest(o.toString().getBytes());
            int hashCode = hashCode(bytes);
            return hashCode % circleSize;
        }
        catch (NoSuchAlgorithmException e) {
            e.printStackTrace();
        }
        return 0;
    }
}