一致性哈希算法

[b]分布式缓存问题[/b]

假设我们有一个网站，最近发现随着流量的增加，服务器压力越来越大，之前直接读写数据库的方式不太给力了，于是我们想引入Memcached作为缓存机制。现在我们一共有三台机器可以作为Memcached服务器，如下图所示。
[img]http://dl2.iteye.com/upload/attachment/0101/3749/f996cf02-df05-376a-b915-ec3caa9e6f85.png[/img]

很显然，最简单的策略是将每一次Memcached请求随机发送到一台Memcached服务器，但是这种策略可能会带来两个问题：一是同一份数据可能被存在不同的机器上而造成数据冗余，二是有可能某数据已经被缓存但是访问却没有命中，因为无法保证对相同key的所有访问都被发送到相同的服务器。因此，随机策略无论是时间效率还是空间效率都非常不好。

要解决上述问题只需做到如下一点：保证对相同key的访问会被发送到相同的服务器。很多方法可以实现这一点，最常用的方法是计算key的哈希值，再对哈希值取模。公式为：
h = Hash(key) % N

其中h是计算出来的每个key的请求应该被发送到哪台服务器的服务器编号；Hash()是一个从字符串到正整数的哈希映射函数，常见的哈希算法有MurMurHash, MD5等；N是服务器的个数，服务器按照0 ~ (N-1)编号。

这样，如果我们将Memcached Server分别编号为0、1、2就可以解决上面提到的两个问题，但是还存在着一些其它问题。

这个算法的问题在于容错性和扩展性不好。所谓容错性是指当系统中某一个或几个服务器变得不可用时，整个系统是否可以正确高效运行；而扩展性是指当加入新的服务器后，整个系统是否可以正确高效运行。

现假设有一台服务器宕机了，那么为了填补空缺，要将宕机的服务器从编号列表中移除，后面的服务器按顺序前移一位并将其编号值减一，此时每个key所在的缓存服务器就要按h = Hash(key) % (N-1)重新计算；同样，如果新增了一台服务器，虽然原有服务器编号不用改变，但是要按h = Hash(key) % (N+1)重新计算，大量的key会被重定位到不同的服务器从而造成大量的缓存不命中。而这种情况在分布式系统中是非常糟糕的。

一个设计良好的分布式哈希方案应该具有良好的单调性，即服务节点的增减不会造成大量哈希重定位。一致性哈希算法就是这样一种哈希方案。

[b]一致性哈希算法[/b]

[b]算法简述[/b]
一致性哈希算法（Consistent Hashing）最早在论文[url=http://www.akamai.com/dl/technical_publications/ConsistenHashingandRandomTreesDistributedCachingprotocolsforrelievingHotSpotsontheworldwideweb.pdf]《Consistent Hashing and Random Trees: Distributed Caching Protocols for Relieving Hot Spots on the World Wide Web》[/url]中被提出。简单来说，一致性哈希将整个哈希值空间组织成一个虚拟的圆环，如假设某哈希函数H的值空间为0 ~ 232-1（即哈希值是一个32位无符号整型），整个哈希空间环如下：

[img]http://dl2.iteye.com/upload/attachment/0101/3675/3c0494b6-cf64-3425-a4ed-9a1e39d3a4d7.png[/img]
整个空间按顺时针方向组织。0和232-1在零点钟方向重合。

下一步可以选择服务器的IP地址或主机名作为关键字，使用Hash()函数将各个服务器进行哈希，这样每台机器就能确定其在哈希环上的位置，这里假设将上述三台服务器使用IP地址哈希后在环空间的位置如下：

[img]http://dl2.iteye.com/upload/attachment/0101/3677/d92fdb89-d50f-347f-8015-426d61581d76.png[/img]
接下来使用如下算法定位数据访问到相应服务器：
将数据key使用相同的哈希函数Hash()计算出哈希值h，根据h确定此数据在环上的位置，从该位置沿着环以顺时针方向“行走”，第一台遇到的服务器就是其应该定位到的服务器。如果哈希值h超过232-1，则从0开始定位到顺时针方向最近的服务器。


例如我们有A、B、C、D四个数据对象，经过哈希计算后，在环空间上的位置如下：

[img]http://dl2.iteye.com/upload/attachment/0101/3679/2c9ed844-2b2b-33d4-a15d-4470422ff290.png[/img]
根据一致性哈希算法，数据A会被定为到Server 1上，D被定为到Server 3上，而B、C分别被定为到Server 2上。

[b]容错性与可扩展性分析[/b]
下面分析一致性哈希算法的容错性和可扩展性。现假设Server 3宕机了：

[img]http://dl2.iteye.com/upload/attachment/0101/3681/56a9c635-5ad9-3265-8703-2f342b3d0adc.png[/img]
可以看到此时A、C、B不会受到影响，只有D节点被重定位到Server 2。一般的，在一致性哈希算法中，如果一台服务器不可用，则受影响的数据仅仅是此服务器到其环空间中前一台服务器（即顺着逆时针方向行走遇到的第一台服务器）之间的数据，其它不会受到影响。

下面考虑另外一种情况，如果我们在系统中增加一台服务器Memcached Server 4：

[img]http://dl2.iteye.com/upload/attachment/0101/3683/e6598be9-a7db-3b06-b624-40b72edba9f2.png[/img]
此时A、D、C不受影响，只有B需要重定位到新的Server 4。一般的，在一致性哈希算法中，如果增加一台服务器，则受影响的数据仅仅是新服务器到其环空间中前一台服务器（即顺着逆时针方向行走遇到的第一台服务器）之间的数据，其它不会受到影响。

综上所述，一致性哈希算法对于节点的增减都只需重定位环空间中的一小部分数据，具有较好的容错性和可扩展性。

[b]虚拟节点[/b]
一致性哈希算法在服务节点太少时，容易因为节点分布不均匀而造成数据倾斜问题。例如我们的系统中有两台服务器，其环分布如下：

[img]http://dl2.iteye.com/upload/attachment/0101/3685/4e71e7e2-e50b-3795-b15f-d15e357046ea.png[/img]
此时必然造成大量数据集中到Server 1上，而只有极少量会定位到Server 2上。为了解决这种数据倾斜问题，一致性哈希算法引入了虚拟节点机制，即对每一个服务节点计算多个哈希，每个计算结果都放置一个此服务节点，称为虚拟节点。具体做法可以在服务器IP地址或主机名的后面增加编号来实现。例如上面的情况，我们决定为每台服务器计算三个虚拟节点，于是可以分别计算“Memcached Server 1#1”、“Memcached Server 1#2”、“Memcached Server 1#3”、“Memcached Server 2#1”、“Memcached Server 2#2”、“Memcached Server 2#3”的哈希值，于是形成六个虚拟节点：

[img]http://dl2.iteye.com/upload/attachment/0101/3687/526ccd2a-1e3e-351e-9205-cf850470a4a0.png[/img]
同时数据定位算法不变，只是多了一步虚拟节点到实际节点的映射，例如定位到“Memcached Server 1#1”、“Memcached Server 1#2”、“Memcached Server 1#3”三个虚拟节点的数据均定位到Server 1上。这样就解决了服务节点少时数据倾斜的问题。在实际应用中，通常将虚拟节点数设置为32甚至更大，因此即使很少的服务节点也能做到相对均匀的数据分布。

import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.Arrays;
import java.util.List;
import java.util.SortedMap;
import java.util.TreeMap;

/**
 * 一致性哈希算法 简单示例
 */
public class ConsistentHash<T> {

    private final Hashing algo; // 哈希函数
    private final int virtualNodeNum = 32; // 虚拟节点数量
    private final SortedMap<Long, T> circleMap = new TreeMap<Long, T>(); // 哈希环

    public ConsistentHash(Hashing algo, List<T> nodes) {
        this.algo = algo;
        for (T node : nodes) {
            add(node);
        }
    }

    public void add(T node) {
        for (int i = 0; i < virtualNodeNum; i++) {
            circleMap.put(algo.hash(node.toString() + i), node);
        }
    }

    public void remove(T node) {
        for (int i = 0; i < virtualNodeNum; i++) {
            circleMap.remove(algo.hash(node.toString() + i));
        }
    }

    public T get(Object key) {
        if (circleMap.isEmpty()) {
            return null;
        }
        SortedMap<Long, T> tailMap = circleMap.tailMap(algo.hash(key.toString()));
        if (tailMap.size() == 0) {
            return circleMap.get(circleMap.firstKey());
        } else {
            return circleMap.get(tailMap.firstKey());
        }
    }

    public static void main(String[] args) {
        Node node1 = new Node("node1", "192.168.0.1");
        Node node2 = new Node("node2", "192.168.0.3");
        Node node3 = new Node("node3", "192.168.0.5");
        List<Node> nodeList = Arrays.asList(node1, node2, node3);
        ConsistentHash<Node> consistentHash = new ConsistentHash<Node>(new Hashing(), nodeList);
        System.out.println(consistentHash.get("134"));
        System.out.println(consistentHash.get("863"));
        System.out.println(consistentHash.get("048"));
        System.out.println(consistentHash.get("652"));
        System.out.println(consistentHash.get("862"));
    }
}

class Node {
    private String name;
    private String ip;
    public Node(String name, String ip) {
        this.name = name;
        this.ip = ip;
    }
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
    public String getIp() {
        return ip;
    }
    public void setIp(String ip) {
        this.ip = ip;
    }
    @Override
    public String toString() {
        return ip;
    }
}

class Hashing {
    public ThreadLocal<MessageDigest> md5Holder = new ThreadLocal<MessageDigest>();
    public long hash(String key) {
        try {
            return hash(key.getBytes("UTF-8"));
        } catch (Exception e) {
            throw new RuntimeException();
        }
    }

    public long hash(byte[] key) {
        try {
            if (md5Holder.get() == null) {
                md5Holder.set(MessageDigest.getInstance("MD5"));
            }
        } catch (NoSuchAlgorithmException e) {
            throw new IllegalStateException("No MD5 algorithm found");
        }
        MessageDigest md5 = md5Holder.get();

        md5.reset();
        md5.update(key);
        byte[] bKey = md5.digest();
        long res = ((long) (bKey[3] & 0xFF) << 24) | ((long) (bKey[2] & 0xFF) << 16) | ((long) (bKey[1] & 0xFF) << 8)
                | (long) (bKey[0] & 0xFF);
        return res;
    }
}

参考资料：
http://blog.codinglabs.org/articles/consistent-hashing.html