一致性 hash 算法

关注
发布于: 2020 年 07 月 08 日
﻿
题目：
用你熟悉的编程语言实现一致性 hash 算法。
编写测试用例测试这个算法，测试 100 万 KV 数据，10 个服务器节点的情况下，计算这些 KV 数据在服务器上分布数量的标准差，以评估算法的存储负载不均衡性。
﻿
主要思想：
首先，理解题意。
一致性算法是什么？
简单来说，一致性哈希将整个哈希值空间组织成一个虚拟的圆环，如假设某哈希函数H的值空间为0-2^32-1（即哈希值是一个32位无符号整形），整个哈希空间环如下，整个空间按顺时针方向组织。0和232-1在零点中方向重合。下一步将各个服务器使用Hash进行一个哈希，具体可以选择服务器的ip或主机名作为关键字进行哈希，这样每台机器就能确定其在哈希环上的位置。
﻿
﻿
标准差是什么？
标准差（Standard Deviation） ，是离均差平方的算术平均数的平方根，用σ表示。在概率统计中最常使用作为统计分布程度上的测量。标准差是方差的算术平方根。标准差能反映一个数据集的离散程度。平均数相同的两组数据，标准差未必相同。
﻿
其次，要达到以上目的，要选择合适的数据结构，计算hash值，从而具有：均衡性（在整个环中分布尽量均匀），不变性（每次更新等操作原来的值不变）。
﻿
具体实现如下：
﻿
﻿
1、选取计算hash的算法和计算标准差的工具类 ConsistentUtil
﻿
public class ConsistentUtil {
    /**
     * 计算Hash值, 使用FNV1_32_HASH算法,memcached官方使用了基于md5的KETAMA算法
     * @param str
     * @return
     */
    public static int getHash(String str) {
        final int p = 16777619;
        int hash = (int)2166136261L;
        for (int i = 0; i < str.length(); i++) {
            hash =( hash ^ str.charAt(i) ) * p;
        }
        hash += hash << 13;
        hash ^= hash >> 7;
        hash += hash << 3;
        hash ^= hash >> 17;
        hash += hash << 5;
﻿
        if (hash < 0) {
            hash = Math.abs(hash);
        }
        return hash;
    }
﻿
    /**
     * 标准差（Standard Deviation） ，是离均差平方的算术平均数的平方根，用σ表示。
     * @param statMap
     * @return
     */
    public static double computeStd(Map<String,Integer> statMap) {
        Integer[] hitData = new Integer[statMap.size()];
        statMap.values().toArray(hitData);
        double avg = Arrays.stream(hitData).mapToInt(Integer::intValue).average().orElse(0d);
        double avgStd = Arrays.stream(hitData).map(count -> Math.pow(count - avg, 2)).mapToDouble(Double::doubleValue).average().orElse(0d);
        double std = Math.sqrt(avgStd);
        return std;
    }
    
  }
﻿
2、对一致性hash算法的接口IConsistHash
public interface IConsistHash {
    /**
     * 构建环
     */
    SortedMap<Integer,String> SERVER_SORTED_MAP = new TreeMap<>();
    /**
     * 虚拟节点
     */
    int VIRTUAL_NODE_NUM = 200;
    /**
     * 统计命中服务的次数
     */
    Map<String,Integer> SERVER_HIT_RET = Maps.newHashMap();
﻿
    /**
     * 加入统计
     * @param node
     */
    default void addStat(String node){
        if (!SERVER_HIT_RET.containsKey(node)){
            SERVER_HIT_RET.put(node,1);
        }else {
            SERVER_HIT_RET.put(node, SERVER_HIT_RET.get(node)+1);
        }
    }
﻿
    SortedMap<Integer,String> putNodetoHashring(List<Server> serverList);
﻿
    /**
     * 根据访问key，得到所在服务node
     * @param key
     * @return
     */
    String getServer(String key);
﻿
    public void setVnode(int vnode);
﻿
}
﻿
3、实现IConsistHash接口，一个不带虚拟节点，一个带有虚拟节点
   其中带有虚拟节点的如下：
@Setter @Getter
public class ConsistHashWithVirtualNodeimpl extends AbstractConsistHash implements IConsistHash{
    //虚拟节点，方便测试用
    private int vnode = 0;
﻿
    @Override
    public SortedMap<Integer, String> putNodetoHashring(List<Server> serverList) {
        serverList.forEach(server -> {
            int hash = ConsistentUtil.getHash(server.getDomainName());
            SERVER_SORTED_MAP.put(hash,server.getDomainName());
            //int hash = 0;
            if (this.getVnode() !=0){
                for (int i=0;i< vnode ;i++){
                    String vnName = getVirtualNode(server.getDomainName(),i);
                    hash = ConsistentUtil.getHash(vnName);
                    SERVER_SORTED_MAP.put(hash,server.getDomainName());
                }
            }else {
                for (int i=0;i< VIRTUAL_NODE_NUM;i++){
                    String vnName = getVirtualNode(server.getDomainName(),i);
                    hash = ConsistentUtil.getHash(vnName);
                    SERVER_SORTED_MAP.put(hash,server.getDomainName());
                }
            }
﻿
        });
        return SERVER_SORTED_MAP;
    }
﻿
﻿
    public static String getVirtualNode(String realName, int num) {
        return realName + "#" + String.valueOf(num);
    }
}
﻿
4、ServerConfigImpl 实现IServerConfig 接口，实现对服务节点添加和删除。操作的时候同时刷新hash环
public class ServerConfigImpl implements IServerConfig {
﻿
    private List<Server> groups = Lists.newArrayList(new Server("198.168.10.1")
﻿
            );
﻿
    private IConsistHash consistHash;
    public ServerConfigImpl (IConsistHash consistHash){
        this.consistHash = consistHash;
        refreshHashring();
    }
    @Override
    public List<Server> getAllServer(){
        return groups;
    }
﻿
﻿
    @Override
    public void add(Server server) {
        if (groups.contains(server)){
            return;
        }
        groups.add(server);
        refreshHashring();
    }
﻿
    @Override
    public void remove(Server server) {
        groups.remove(server);
        refreshHashring();
    }
﻿
    private void refreshHashring(){
        IConsistHash.SERVER_SORTED_MAP.clear();
        // 统计结果清除
        IConsistHash.SERVER_HIT_RET.clear();
        consistHash.putNodetoHashring(groups);
    }
}
﻿
﻿
5、抽象类AbstractConsistHash实现对hash的读取所在的服务器
public abstract class AbstractConsistHash implements IConsistHash{
﻿
    @Override
    public String getServer(String key){
        int keyHash = ConsistentUtil.getHash(key);
        // 只取出大于等于该hash值的部分，从中获取第一个server
        SortedMap<Integer,String> sortedMap = SERVER_SORTED_MAP.tailMap(keyHash);
        // hash值在最尾部，应该在第一个server上
        String node ;
        if (CollectionUtil.isEmpty(sortedMap)){
            node = SERVER_SORTED_MAP.get(SERVER_SORTED_MAP.firstKey());
        }else {
            node = sortedMap.get(sortedMap.firstKey());
        }
        // 加入统计
        addStat(node);
        return node;
    }
﻿
    @Override
    public void setVnode(int vnode) {
﻿
    }
}
﻿
6、测试结果
测试类：
public class ConsisHashingWithVirtualNodeTest {
﻿
    private static double printStd(IConsistHash consistHash){
        int requestCount = 100*10000;
        for (int i = 0; i < requestCount; i++) {
            consistHash.getServer(""+i);
        }
        return ConsistentUtil.computeStd(IConsistHash.SERVER_HIT_RET);
﻿
    }
﻿
    public static void main(String[] args) {
        //serverConfig.remove(new Server("memcache1"));
        //printStd(consistHash);
//        int vnode = 200;
//        for (int i=1;i<vnode;i++){
//            IConsistHash consistHash = new ConsistHashWithVirtualNodeimpl();
//            consistHash.setVnode(i);
//            IServerConfig serverConfig = new ServerConfigImpl(consistHash);
//            log.info("vnode num:{},std:{}",i,printStd(consistHash));
//        }
        IConsistHash consistHash = new ConsistHashWithVirtualNodeimpl();
        IServerConfig serverConfig = new ServerConfigImpl(consistHash);
        for (int i=1;i<11;i++){
            serverConfig.add(new Server("198.168.10."+i));
            System.out.println("服务器数量:"+i+",标准差:"+printStd(consistHash)+", 每个服务器上的key数量:"+ IConsistHash.SERVER_HIT_RET.values());
        }
﻿
    }
}
﻿
结果如下：
服务器数量:1,标准差:0.0, 每个服务器上的key数量:[1000000]
服务器数量:2,标准差:16627.0, 每个服务器上的key数量:[516627, 483373]
服务器数量:3,标准差:14155.14088787376, 每个服务器上的key数量:[313638, 346283, 340079]
服务器数量:4,标准差:14865.209500710038, 每个服务器上的key数量:[229803, 245650, 271099, 253448]
服务器数量:5,标准差:15007.06248404397, 每个服务器上的key数量:[192299, 185260, 219258, 216979, 186204]
服务器数量:6,标准差:14202.605320464583, 每个服务器上的key数量:[157092, 158440, 160552, 186063, 186634, 151219]
服务器数量:7,标准差:10015.700680553957, 每个服务器上的key数量:[136393, 134230, 139076, 143512, 155872, 159393, 131524]
服务器数量:8,标准差:9190.120687455634, 每个服务器上的key数量:[126396, 116342, 110036, 130296, 127076, 136745, 136547, 116562]
服务器数量:9,标准差:7092.908108251422, 每个服务器上的key数量:[114146, 99332, 114742, 98971, 113789, 114702, 121486, 114277, 108555]
服务器数量:10,标准差:7164.8596078360115, 每个服务器上的key数量:[98090, 91868, 99125, 89392, 96058, 113547, 100062, 111314, 99035, 101509]
﻿