美丽的一致性Hash算法

时间 2019-11-30

原文原文链接

若是在大型高并发系统须要数据的分布式存储但愿数据均匀分布可扩展性强那么一致性hash算法就能够完美解决这个问题一致性hash算法的应用再不少领域 缓存 hadoop ES 分布式数据库node

一致性Hash算法原理

一致性Hash算法是使用取模的方法，一致性的Hash算法是对2的32方取模。即，一致性Hash算法将整个Hash空间组织成一个虚拟的圆环，Hash函数的值空间为0 ~ 2^32 - 1(一个32位无符号整型)，整个哈希环以下： hash值是个整数非负，对集群的某个属性好比节点名取hash值放到环上，对数据key取hash值也放到环上，按照顺时针方向找到离它最近的节点放到它上面，整个圆环以顺时针方向组织，圆环0点右侧的第一个点表明n1服务器，以此类推。咱们将各个服务器使用Hash进行一个哈希，具体能够选择服务器的IP或主机名做为关键字进行哈希，这样每台服务器就肯定在了哈希环的一个位置上，好比咱们有三台机器，使用IP地址哈希后在环空间的位置如图所示：算法

咱们使用如下算法定位数据访问到相应的服务器：

将数据Key使用相同的函数Hash计算出哈希值，并肯定此数据在环上的位置，今后位置沿环顺时针查找，遇到的服务器就是其应该定位到的服务器。数据库

以下图三个数据O1,O2,O3通过哈希计算后，在环空间上的位置以下： O1-->n1 O2-->n2 O3-->n3 缓存

一致性Hash算法的容错性和可扩展性

如今，假设咱们的n3宕机了，咱们从图中能够看到，n一、n2不会受到影响，只有O3对象被从新定位到n1。因此咱们发现，在一致性Hash算法中，若是一台服务器不可用，受影响的数据仅仅是此服务器到其环空间前一台服务器之间的数据其余不会受到影响。如图所示： bash

如今咱们系统增长了一台服务器n4，如图所示服务器

从图中能够看出增长服务器后数据O2,O3没有收到影响只有O1受到影响了从新定位到新的节点n4上了。

一致性Hash算法对于节点的增减都只需重定位环空间中的一小部分数据，有很好的容错性和可扩展性。在一致性Hash算法服务节点太少的状况下，容易由于节点分布不均匀面形成数据倾斜（被缓存的对象大部分缓存在某一台服务器上）问题，如图：并发

这时咱们发现有大量数据集中在节点A上，而节点B只有少许数据。为了解决数据倾斜问题，一致性Hash算法引入了虚拟节点机制，即对每个服务器节点计算多个哈希，每一个计算结果位置都放置一个此服务节点，称为虚拟节点。具体操做能够为服务器IP或主机名后加入编号来实现，实现如图所示：好比一个n1节点咱们虚拟100个虚拟节点，在环上的即是虚拟节点，同理n2,n3也有100个虚拟节点，dom

当数据过来之后如何判断放置到哪个服务器呢

当数据过来入环之后先找到对应的虚拟节点，再经过虚拟节点找到对应的服务器，这样经过增长虚拟节点就能够作到数据的均匀分布，虚拟节点越多数据越均匀，通常咱们一个服务器放置200个虚拟节点便可分布式

数据定位算法不变，只须要增长一步：虚拟节点到实际点的映射。因此加入虚拟节点以后，即便在服务节点不多的状况下，也能作到数据的均匀分布。上面几种状况都是数据理想的状况下均匀分布的，其实一致性Hash算法存在一个数据倾斜问题ide

算法接口类

public interface IHashService {
    Long hash(String key);
}
复制代码

算法接口实现类

public class HashService implements IHashService {

    /**
     * MurMurHash算法,性能高,碰撞率低
     *
     * @param key String
     * @return Long
     */
    public Long hash(String key) {
        ByteBuffer buf = ByteBuffer.wrap(key.getBytes());
        int seed = 0x1234ABCD;

        ByteOrder byteOrder = buf.order();
        buf.order(ByteOrder.LITTLE_ENDIAN);

        long m = 0xc6a4a7935bd1e995L;
        int r = 47;

        long h = seed ^ (buf.remaining() * m);

        long k;
        while (buf.remaining() >= 8) {
            k = buf.getLong();

            k *= m;
            k ^= k >>> r;
            k *= m;

            h ^= k;
            h *= m;
        }

        if (buf.remaining() > 0) {
            ByteBuffer finish = ByteBuffer.allocate(8).order(ByteOrder.LITTLE_ENDIAN);
            finish.put(buf).rewind();
            h ^= finish.getLong();
            h *= m;
        }

        h ^= h >>> r;
        h *= m;
        h ^= h >>> r;

        buf.order(byteOrder);
        return h;

    }
}
复制代码

模拟机器节点

public class Node<T> {
    private String ip;
    private String name;

    public Node(String ip, String name) {
        this.ip = ip;
        this.name = name;
    }

    public String getIp() {
        return ip;
    }

    public void setIp(String ip) {
        this.ip = ip;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    /**
     * 使用IP当作hash的Key
     *
     * @return String
     */
    @Override
    public String toString() {
        return ip;
    }
}
复制代码

一致性Hash操做

public class ConsistentHash<T> {
    // Hash函数接口
    private final IHashService iHashService;
    // 每一个机器节点关联的虚拟节点数量
    private final int          numberOfReplicas;
    // 环形虚拟节点
    private final SortedMap<Long, T> circle = new TreeMap<Long, T>();

    public ConsistentHash(IHashService iHashService, int numberOfReplicas, Collection<T> nodes) {
        this.iHashService = iHashService;
        this.numberOfReplicas = numberOfReplicas;
        for (T node : nodes) {
            add(node);
        }
    }

    /**
     * 增长真实机器节点
     *
     * @param node T
     */
    public void add(T node) {
        for (int i = 0; i < this.numberOfReplicas; i++) {
            circle.put(this.iHashService.hash(node.toString() + i), node);
        }
    }

    /**
     * 删除真实机器节点
     *
     * @param node T
     */
    public void remove(T node) {
        for (int i = 0; i < this.numberOfReplicas; i++) {
            circle.remove(this.iHashService.hash(node.toString() + i));
        }
    }

    public T get(String key) {
        if (circle.isEmpty()) return null;

        long hash = iHashService.hash(key);

        // 沿环的顺时针找到一个虚拟节点
        if (!circle.containsKey(hash)) {
            SortedMap<Long, T> tailMap = circle.tailMap(hash);
            hash = tailMap.isEmpty() ? circle.firstKey() : tailMap.firstKey();
        }
        return circle.get(hash);
    }
}
复制代码

测试类

public class TestHashCircle {
    // 机器节点IP前缀
    private static final String IP_PREFIX = "192.168.0.";

    public static void main(String[] args) {
        // 每台真实机器节点上保存的记录条数
        Map<String, Integer> map = new HashMap<String, Integer>();

        // 真实机器节点, 模拟10台
        List<Node<String>> nodes = new ArrayList<Node<String>>();
        for (int i = 1; i <= 10; i++) {
            map.put(IP_PREFIX + i, 0); // 初始化记录
            Node<String> node = new Node<String>(IP_PREFIX + i, "node" + i);
            nodes.add(node);
        }

        IHashService iHashService = new HashService();
        // 每台真实机器引入100个虚拟节点
        ConsistentHash<Node<String>> consistentHash = new ConsistentHash<Node<String>>(iHashService, 500, nodes);

        // 将5000条记录尽量均匀的存储到10台机器节点上
        for (int i = 0; i < 5000; i++) {
            // 产生随机一个字符串当作一条记录，能够是其它更复杂的业务对象,好比随机字符串至关于对象的业务惟一标识
            String data = UUID.randomUUID().toString() + i;
            // 经过记录找到真实机器节点
            Node<String> node = consistentHash.get(data);
            // 再这里能够能过其它工具将记录存储真实机器节点上，好比MemoryCache等
            // ...
            // 每台真实机器节点上保存的记录条数加1
            map.put(node.getIp(), map.get(node.getIp()) + 1);
        }

        // 打印每台真实机器节点保存的记录条数
        for (int i = 1; i <= 10; i++) {
            System.out.println(IP_PREFIX + i + "节点记录条数：" + map.get(IP_PREFIX + i));
        }
    }
}

 
复制代码

运行结果以下：