import java.util.Arrays;
import java.util.Collection;
import java.util.SortedMap;
import java.util.TreeMap;

/**
 * Created by 01369537 on 2019/8/16.
 */
public class ConsistentHash<T> {

    private final HashFunction hashFunction;
    private final int numberOfReplicas;// 节点的复制因子,实际节点个数 * numberOfReplicas = 虚拟节点个数
    private final SortedMap<Long, T> circle = new TreeMap<>();// 存储虚拟节点的hash值到真实节点的映射

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

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

        }
    }

    public void remove(T node) {
        for (int i = 0; i < numberOfReplicas; i++)
            circle.remove(hashFunction.hash(node.toString() + i));
    }

    /*
     * 获得一个最近的顺时针节点,根据给定的key 取Hash
     * 然后再取得顺时针方向上最近的一个虚拟节点对应的实际节点
     */
    public T get(Object key) {
        if (circle.isEmpty())
            return null;
        long hash = hashFunction.hash((String) key);// node 用String来表示,获得node在哈希环中的hashCode
        if (!circle.containsKey(hash)) {//数据映射在两台虚拟机器所在环之间,就需要按顺时针方向寻找机器
            //SortedMap的tailMap函数返回Map中所有大于该key的元素组成的子map
            SortedMap<Long, T> tailMap = circle.tailMap(hash);
            hash = tailMap.isEmpty() ? circle.firstKey() : tailMap.firstKey();
        }
        //返回存储key的真实的节点
        return circle.get(hash);
    }

    public long getSize() {
        return circle.size();
    }

    public static void main(String[] args) {

        ConsistentHash<String> consistentHash = new ConsistentHash<>(new HashFunction() {
            @Override
            public Long hash(Object key) {
                return (long) key.hashCode();
            }
        }, 2, Arrays.asList("1", "2", "3"));

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

    }

}