一致性哈希算法

一致哈希 是一种特殊的哈希算法。在使用一致哈希算法后，哈希表槽位数（大小）的改变平均只需要对K/n个关键字重新映射，其中K是关键字的数量，n是槽位数量。然而在传统的哈希表中，添加或删除一个槽位的几乎需要对所有关键字进行重新映射。

分布式缓存问题

一个网站随着流量增加，服务器压力越来越大，之前直接读写数据库的方式遭遇I/O和连接数的瓶颈，于是我们引入Memcached作为缓存服务器减轻DB压力。
如果我们有三台Memcached缓存服务器，如何路由数据到不同的缓存服务器呢？
我们很容易想到通过hash的方法来路由请求，通过target=hash(key) %n，其中n是服务器数量。
但是以上方法会有一个问题，就是当某台服务器宕机或者新增服务器时，大量的缓存内容会变更目标地址，所以会导致大量的缓存失效和重设。
这个时候，就需要一种一致性哈希算法来尽可能使同一个资源映射到同一台缓存服务器。
一致哈希算法的主要思想是将每个缓存服务器与一个或多个哈希值域区间关联起来，其中区间边界通过计算缓存服务器对应的哈希值来决定。如果一个缓存服务器被移除，则它所对应的区间会被并入到邻近的区间，其他的缓存服务器不需要任何改变。

算法背景

一致性哈希算法在1997年由麻省理工学院的Karger等人在解决分布式Cache中提出的，设计目标是为了解决因特网中的热点(Hot spot)问题。原文在http://dl.acm.org/citation.cfm?id=258660

实现

一致哈希将每个对象映射到圆环边上的一个点，系统再将可用的节点机器映射到圆环的不同位置。
查找某个对象对应的机器时，用一致哈希算法计算得到对象对应圆环边上位置，沿着圆环边上查找直到遇到某个节点机器，这台机器即为对象应该保存的位置。
当删除一台节点机器时，这台机器上保存的所有对象都要移动到下一台机器。
添加一台机器到圆环边上某个点时，这个点的下一台机器需要将这个节点前对应的对象移动到新机器上。
更改对象在节点机器上的分布可以通过调整节点机器的位置来实现。

一致性hash算法

1. 首先求出memcached服务器（节点）的哈希值，并将其配置到0～2^32的圆环上
2. 然后采用同样的方法求出存储数据的键的哈希值，并映射到相同的圆上
3. 然后从数据映射到的位置开始顺时针查找，将数据保存到找到的第一个服务器上
4. 如果超过2^32仍然找不到服务器，就会保存到第一台memcached服务器上

当添加一个新的服务器节点时，如下图所示：

添加一个新的服务器节点

node5是新增的节点，此时原本原本全部在node4上的内容，会有一部分转移到node5上，其他节点的数据内容不受影响。

同理，删除一个节点，受影响的也只是与被删除节点相邻的服务器。

带虚拟节点的一致性哈希算法

以上算法存在一个问题，就是在节点数目很少，且节点的hash值较为集中的时候，会造成很明显的数据分配不均的问题。

数据分配不均问题

如上图所示，当只存在两个节点时，大量数据会分配在nodeA上，只有少量数据会分配在nodeB上，为了解决上述问题，引入了虚拟节点的概念。即对每一个服务节点计算多个哈希，每个计算结果位置都放置一个此服务节点，称为虚拟节点。具体做法可以在服务器ip或主机名的后面增加编号来实现。例如上面的情况，可以为每台服务器计算三个虚拟节点，于是可以分别计算 “NodeA#1”、“NodeA#2”、“Node A#3”、“NodeB#1”、“NodeB#2”、“NodeB#3”的哈希值，于是形成六个虚拟节点。最后再根据虚拟节点-真实节点的映射规则，求出真实节点。

虚拟节点的一致性哈希算法

代码实现（Java）

package com.zhuke.daily;

import java.util.Iterator;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.UUID;
import java.util.zip.CRC32;

/**
 * 一致性hash算法
 * <p>
 * Created by ZHUKE on 2017/8/16.
 */
public class ConsistentHashing {

    /**
     * 服务器信息
     */
    private static final String[] SERVERS = new String[]{"node1", "node2", "node3"};

    /**
     * 一致性哈希环
     */
    private static SortedMap<Long, String> hashRing = new TreeMap<>();

    /**
     * 虚拟节点数目
     */
    private static final int VIRTUAL_NODE_COUNT = 32;

    /**
     * hash环的最大位置2^32
     */
    private static final long MAX_HASH_LOCATION = 1L << 32;

    static {
        //initHashRingWithoutVirtualNode();
        initHashRingWithVirtualNode();
    }

    /**
     * 初始化哈希环，没有虚拟节点
     */
    public static void initHashRingWithoutVirtualNode() {
        for (String server : SERVERS) {
            hashRing.put(hash(server), server);
        }
    }

    /**
     * 初始化哈希环，带虚拟节点
     */
    public static void initHashRingWithVirtualNode() {
        for (String server : SERVERS) {
            for (int i = 0; i < VIRTUAL_NODE_COUNT; i++) {
                String virtualServer = server + "#virtual" + i;
                hashRing.put(hash(virtualServer), virtualServer);
            }
        }
    }

    /**
     * 为key分配server
     *
     * @param key key
     * @return 分配的server
     */
    public static String allocateServer(String key) {
        long hashValue = hash(key);
        /*
        * TreeMap保证key的有序排列，可以当作是一个首尾相连的环,
        * 所以为新的key分配server地址时，按照相同的hash算法得到hashValue，
        * 在TreeMap的环状结构中，查找第一个比该hashValue大的key，其对应的server即为分配的server
        * */
        SortedMap<Long, String> tailHashRing = hashRing.tailMap(hashValue);
        String virtualNode;
        if (tailHashRing.isEmpty()) {
            virtualNode = hashRing.get(hashRing.firstKey());
        } else {
            virtualNode = hashRing.get(tailHashRing.firstKey());
        }
        return getRealNode(virtualNode);
    }


    private static long hash(String key) {
        CRC32 crc32 = new CRC32();
        crc32.update(key.getBytes());
        return crc32.getValue() % MAX_HASH_LOCATION;
    }

    private static String getRealNode(String virtualNode) {
        if (!virtualNode.contains("#")) {
            return virtualNode;
        }
        return virtualNode.split("#")[0];
    }

    public static void main(String[] args) {
        Iterator<Long> iterator = hashRing.keySet().iterator();
        while (iterator.hasNext()) {
            Long key = iterator.next();
            System.out.println(key + "," + hashRing.get(key));
        }
        int node1Count = 0;
        int node2Count = 0;
        int node3Count = 0;

        for (int i = 0; i < 1000; i++) {
            String key = UUID.randomUUID().toString();
            String server = allocateServer(key);
            if (server.equals("node1")) node1Count++;
            if (server.equals("node2")) node2Count++;
            if (server.equals("node3")) node3Count++;
        }
        System.out.println("node1Count=" + node1Count);
        System.out.println("node2Count=" + node2Count);
        System.out.println("node3Count=" + node3Count);
    }

}