package com.sdtool.common.tool;


import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * Created by jlutt on 2020-08-28
 * 一致性hash实现
 * https://juejin.im/post/6844903862600482830
 * https://www.jianshu.com/p/da5a9f884aea
 *
 * @author jlutt
 */
public class ConsistentHash {

    private ConsistentHashFunction hashFunction;

    /**
     * 存储虚拟节点的hash值到真实节点的映射
     * 用跳表模拟一致性hash环,即使在节点很多的情况下,也可以有不错的性能
     */
    private final ConcurrentSkipListMap<Long, String> circle;

    /**
     * 节点的复制因子,实际节点个数 * numberOfReplicas = 虚拟节点个数
     */
    private final int numberOfReplicas;

    /**
     * 构造函数
     *
     * @param hashFunction     hash方法
     * @param numberOfReplicas
     * @param nodes            实际节点
     */
    public ConsistentHash(ConsistentHashFunction hashFunction, int numberOfReplicas, Collection<String> nodes) {
        this.circle = new ConcurrentSkipListMap<>();
        this.hashFunction = hashFunction;
        this.numberOfReplicas = numberOfReplicas;
        for (String node : nodes)
            add(node);
    }

    /**
     * 增加实际节点
     * <p>
     * 对于一个实际机器节点 node, 对应 numberOfReplicas 个虚拟节点
     * 不同的虚拟节点(i不同)有不同的hash值,但都对应同一个实际机器node
     * 虚拟node一般是均衡分布在环上的,数据存储在顺时针方向的虚拟node上
     *
     * @param node
     */
    public void add(String node) {
        if (node == null) return;

        //对于一个实际机器节点 node, 对应 numberOfReplicas 个虚拟节点
        //不同的虚拟节点(i不同)有不同的hash值,但都对应同一个实际机器node
        //虚拟node一般是均衡分布在环上的,数据存储在顺时针方向的虚拟node上
        for (int i = 0; i < numberOfReplicas; i++) {
            String visual = node.toString() + "-N" + i;
            Long hash = hashFunction.hash(visual);
            System.out.println("key=" + visual + " hash=" + hash);
            circle.put(hash, node);
        }
    }

    /**
     * 移除节点
     *
     * @param node
     */
    public void remove(String node) {
        if (node == null) return;

        for (int i = 0; i < numberOfReplicas; i++) {
            String visual = node.toString() + i;
            Long hash = hashFunction.hash(visual);
            circle.remove(hash, visual);
        }
    }

    /**
     * 获取key对应的节点
     * <p>
     * 获得一个最近的顺时针节点,根据给定的key 取Hash
     * 然后再取得顺时针方向上最近的一个虚拟节点对应的实际节点
     *
     * @param key
     * @return
     */
    public String get(Object key) {
        if (circle.isEmpty()) {
            return null;
        }

        // node 用String来表示,获得node在哈希环中的hashCode
        long hash = hashFunction.hash((String) key);

        if (!circle.containsKey(hash)) {
            //数据映射在两台虚拟机器所在环之间,就需要按顺时针方向寻找机器
            //tailMap函数返回Map中所有大于该key的元素组成的子map
            ConcurrentNavigableMap<Long, String> tailMap = circle.tailMap(hash);
            hash = tailMap.isEmpty() ? circle.firstKey() : tailMap.firstKey();
        }
        //返回存储key的真实的节点
        return circle.get(hash);
    }

    public static void main(String[] args) {

        ConsistentHash consistentHash = new ConsistentHash((s)-> (long) s.hashCode(), 3, Arrays.asList("1", "2", "3", "4", "5", "6", "7", "9", "10"));


        for (int i = 10; i < 100; i++) {
            String key2 = "Paper" + String.valueOf(i);

            String node2 = consistentHash.get(key2);
            System.out.println("key=" + key2 + " keyHashCode=" + key2.hashCode() + " node=" + node2);

        }

    }
}
