本文就一致性哈希算法的基本原理及实现方式做相关介绍

背景

试想这样一个场景，我们拥有3个Redis服务实例（分别为0号~2号实例）。我们期望缓存数据可以均匀分布到这3台机器上。一个朴素的想法是对key先哈希再取模，即 hash(key) % N。示例如下所示

hash(key 1) = 30，30%3 = 0 → 0号Redis实例
hash(key 2) = 17，17%3 = 2 → 2号Redis实例
hash(key 3) = 11，11%3 = 2 → 2号Redis实例
hash(key 4) = 22，22%3 = 1 → 1号Redis实例

一切看上去符合预期。但当我们增加了一台Redis服务实例，映射关系会发生什么变化

hash(key 1) = 30，30%4 = 2 → 2号Redis实例
hash(key 2) = 17，17%4 = 1 → 1号Redis实例
hash(key 3) = 11，11%4 = 3 → 3号Redis实例
hash(key 4) = 22，22%4 = 2 → 2号Redis实例

可以看到key1对应的目标Redis服务，从原来的0号实例变为2号实例。同理，key2由原来的2号实例调整为1号实例，key3由原来的2号实例调整为3号实例，key4由原来的1号实例调整为2号实例。可以看到在该策略下，如果添加任意数量的服务实例。即会导致原有映射关系大量发生变化。对于缓存服务而言，此举会导致大量缓存生效，进而让请求直接打到DB层，严重时甚至会产生Redis雪崩。而在现代分布式系统中，服务节点数量的动态变化会很频繁。我们期望有一种映射策略，使得能够在目标节点发生变动时，原有的映射关系能够尽可能的保持不变，即尽可能小的改变已有的映射关系。为此一致性Hash算法应运而生

基本原理

该算法将32bit的哈希值映射到一个圆环中，即Hash环空间。其中起点为0，终点为2^32-1，且首尾相连。如下所示

可利用目标节点的IP、名称等信息通过哈希函数计算相关目标节点的哈希值。然后根据节点的哈希值映射到哈希环当中。下面是一个三节点的映射示例

那如何利用这样的一个哈希环建立映射关系呢？其实很简单，我们依然继续对Key进行哈希值的计算，得到其哈希值即可确定该Key在哈希环中的位置。然后按照顺时针方向，则其在哈希环上遇到的第一个节点，即为其最终的映射结果。数据映射的示例如下所示，可以看到Key 1映射到2号节点上，而Key 2、Key 3则映射到1号节点上

如果节点发生变化，比如这里我们添加一个新的节点——4号节点，并将其映射到哈希环中。如下所示。可以看到大部分原有的映射关系并不会发生变化，而只是Key 3数据被调整映射到新的4号节点上。删除节点同理，此处不再赘述

一切看上去已经很完美了，但这里其实还有一个小问题。当真实节点数量不够多时，容易导致其在哈希环上的分布不够均匀，进而产生大量Key映射到个别某些节点的情形，即所谓的数据倾斜。如下图左所示。为此通过可以引入虚拟节点机制来解决，比如这里我们将1个真实节点对应于3个虚拟节点，具体可通过在真实节点的IP、名称等后面添加后缀编号实现。然后将虚拟节点映射到哈希环中即可，这样即可有效避免数据倾斜的发生。如下图右边所示

Java实现

可以看到，一致性哈希算法的基本思想还是很简单的。那么这里讨论下该如何利用Java进行实现。首先对于Hash环来说，由于我们需要在其上查询合适的目标节点，故这里我们通过TreeMap来进行实现。其次，如何选择一个合适的Hash函数。这里我们选择MurmurHash 3算法来生成一个32bit的哈希值。与其它流行的哈希函数相比，其对于规律性较强的输入能够保证良好的随机分布特征，故该哈希函数常适用于查找检索领域。值得一提的是由于各节点的Hash值可能发生冲突，为避免在移除服务器实例时出现错误。这里简便起见选择重建整个Hash环。下面即是一个基于虚拟节点机制的一致性哈希算法的实现示例

/**
 * Consistent Hashing
 * @author Aaron Zhu
 */
@Data
@AllArgsConstructor
@NoArgsConstructor
@Builder
public class ConsistentHashing {
    /**
     * 虚拟节点名称后缀
     */
    private static final String virtualNodeSuffix = "_VN_";

    /**
     * 虚拟节点数量
     */
    private Integer virtualNodeSize;

    /**
     * 服务器集合
     */
    private Set<String> serverSet;

    /**
     * 哈希环, key: Hash值; value: 虚拟节点名称
     */
    private SortedMap<Long, String> hashRing;

    /**
     * 工厂方法, 获取实例
     * @param virtualNodeNum 虚拟节点数量
     * @return
     */
    public static ConsistentHashing getInstance(Integer virtualNodeNum) {
        return ConsistentHashing.builder()
            .virtualNodeSize( virtualNodeNum )
            .serverSet( new HashSet<>())
            .hashRing( new TreeMap<>() )
            .build();
    }

    /**
     * 添加服务器
     */
    public void addServer(String ... servers) {
        Set<String> newServers = Arrays.stream(servers)
            .collect(Collectors.toSet());

        // 添加指定服务器
        serverSet.addAll( newServers );

        // 添加新节点到Hash环中
        buildHashRing( newServers );
    }

    /**
     * 移除服务器
     */
    public void removeServer(String ... servers) {
        // 移除指定服务器
        boolean flag = serverSet.removeAll( Arrays.asList(servers) );
        // 不存在需要移除的服务器
        if( !flag ) {
            return;
        }

        // 清空以重建Hash环
        hashRing.clear();
        // 重建Hash环
        buildHashRing( serverSet );
    }

    /**
     * 查询指定Key的路由结果
     * @param key
     * @return
     */
    public String route(String key) {
        // 计算该Key的Hash值
        long hash32 = hash32(key);

        String virtualNode = null;
        // 获取Hash环中 大于等于 该Key Hash值 的子Map
        SortedMap<Long, String> subMap = hashRing.tailMap(hash32);
        // Hash环中 没有比该Key Hash值 大于等于的, 则从Hash环的第一个虚拟节点开始
        if( subMap.isEmpty() ) {
            virtualNode = hashRing.get( hashRing.firstKey() ) ;
        } else {
            // 选择第一个不小于 该Key Hash值的虚拟节点
            virtualNode = subMap.get( subMap.firstKey() ) ;
        }

        // 将 虚拟节点 转换为 服务器
        String server = virtualNode.substring(0, virtualNode.indexOf(virtualNodeSuffix));
        return server;
    }

    /**
     * 构建Hash环
     * @param set
     * @return
     */
    private void buildHashRing(Set<String> set) {
        // 1. 将服务器转换为虚拟节点
        List<String> virtualNodeList = set.stream()
            .flatMap( server ->
                IntStream.rangeClosed(1, virtualNodeSize)
                    .mapToObj(num -> server + virtualNodeSuffix + num)
            )
            .collect(Collectors.toList());

        // 2. 将虚拟节点添加到Hash环中
        virtualNodeList.forEach( virtualNode -> {
            long hash32 = hash32(virtualNode);
            hashRing.put(hash32, virtualNode);
        } );
    }

    /**
     * 计算32位Hash值
     * @param msg
     * @return
     * @apiNote 哈希算法: MurmurHash 3
     */
    private long hash32(String msg) {
    	// 利用Apache Commons Codec库计算Hash值
        int hash = MurmurHash3.hash32x86( msg.getBytes() );
        return Integer.toUnsignedLong(hash);
    }

    /**
     * 查看哈希环信息
     */
    public void getHashRingInfo() {
        System.out.println();
        hashRing.forEach( (k,v) -> {
            String info = "(Hash: "+k+") ---->>>> ["+ v +"]";
            System.out.println(info);
        } );
        System.out.println();
    }

}

单元测试，如下所示

public class TestConsistentHashing {

    private static ConsistentHashing consistentHashing;

    @BeforeClass
    public static void init() {
        System.out.println("\n----------------------- Init -----------------------");
        // 获取实例
        consistentHashing = ConsistentHashing.getInstance(6);
        // 添加服务器
        consistentHashing.addServer("192.168.1.1:11");
        consistentHashing.addServer("192.168.1.2:22", "192.168.1.3:33");
    }

    @Test
    public void testRoute1() {
        System.out.println("\n----------------------- TestRoute 1 -----------------------");

        // 查看Hash环
        consistentHashing.getHashRingInfo();

        List<String> list = Arrays.asList("医药", "半导体", "新能源");
        list.forEach( e -> {
            String server = consistentHashing.route(e);
            System.out.println( e + " ---->>>> ["+server+"]");
        } );
    }

    @Test
    public void testRoute2() {
        System.out.println("\n----------------------- TestRoute 2 -----------------------");

        // 移除服务器
        consistentHashing.removeServer("192.168.1.2:22");
        // 查看Hash环
        consistentHashing.getHashRingInfo();

        List<String> list = Arrays.asList("医药", "半导体", "新能源");
        list.forEach( e -> {
            String server = consistentHashing.route(e);
            System.out.println( e + " ---->>>> ["+server+"]");
        } );
    }

}

测试结果如下所示

----------------------- Init -----------------------

----------------------- TestRoute 1 -----------------------

(Hash: 305141336) ---->>>> [192.168.1.2:22_VN_3]
(Hash: 450060799) ---->>>> [192.168.1.1:11_VN_3]
(Hash: 471029737) ---->>>> [192.168.1.3:33_VN_5]
(Hash: 534903081) ---->>>> [192.168.1.1:11_VN_1]
(Hash: 662155305) ---->>>> [192.168.1.2:22_VN_5]
(Hash: 1413588880) ---->>>> [192.168.1.2:22_VN_1]
(Hash: 1658845364) ---->>>> [192.168.1.3:33_VN_1]
(Hash: 1863772719) ---->>>> [192.168.1.1:11_VN_2]
(Hash: 2010503113) ---->>>> [192.168.1.1:11_VN_5]
(Hash: 2045065413) ---->>>> [192.168.1.3:33_VN_2]
(Hash: 2352433305) ---->>>> [192.168.1.3:33_VN_3]
(Hash: 2665969315) ---->>>> [192.168.1.1:11_VN_6]
(Hash: 2691359941) ---->>>> [192.168.1.3:33_VN_4]
(Hash: 2930788506) ---->>>> [192.168.1.2:22_VN_2]
(Hash: 3339997586) ---->>>> [192.168.1.3:33_VN_6]
(Hash: 3548837446) ---->>>> [192.168.1.2:22_VN_6]
(Hash: 4003122325) ---->>>> [192.168.1.1:11_VN_4]
(Hash: 4011199543) ---->>>> [192.168.1.2:22_VN_4]

医药 ---->>>> [192.168.1.2:22]
半导体 ---->>>> [192.168.1.3:33]
新能源 ---->>>> [192.168.1.2:22]

----------------------- TestRoute 2 -----------------------

(Hash: 450060799) ---->>>> [192.168.1.1:11_VN_3]
(Hash: 471029737) ---->>>> [192.168.1.3:33_VN_5]
(Hash: 534903081) ---->>>> [192.168.1.1:11_VN_1]
(Hash: 1658845364) ---->>>> [192.168.1.3:33_VN_1]
(Hash: 1863772719) ---->>>> [192.168.1.1:11_VN_2]
(Hash: 2010503113) ---->>>> [192.168.1.1:11_VN_5]
(Hash: 2045065413) ---->>>> [192.168.1.3:33_VN_2]
(Hash: 2352433305) ---->>>> [192.168.1.3:33_VN_3]
(Hash: 2665969315) ---->>>> [192.168.1.1:11_VN_6]
(Hash: 2691359941) ---->>>> [192.168.1.3:33_VN_4]
(Hash: 3339997586) ---->>>> [192.168.1.3:33_VN_6]
(Hash: 4003122325) ---->>>> [192.168.1.1:11_VN_4]

医药 ---->>>> [192.168.1.3:33]
半导体 ---->>>> [192.168.1.3:33]
新能源 ---->>>> [192.168.1.1:11]

参考文献

数据密集型应用系统设计（DDIA） Martin Kleppmann著
码农翻身刘欣著