package cn.qiangjun.hash;
import lombok.Data;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
/**
 * @Author: 梅子黄时雨 qiangjun@aliyun.com
 * @Date: 2020/7/7 17:33
 */
@Data
public class ConsistentHash {
    /**
     * hash环
     */
    private TreeMap<Long, Node> hashCricle = new TreeMap<>();
    /**
     * 每个节点缓存的数量
     */
    private ConcurrentHashMap<String, AtomicInteger> countEveryNode = new ConcurrentHashMap<>(10);
    /**
     * 虚拟节点
     */
    private int virtualCopies = 32 * 10000;
    /**
     * 添加一个节点
     * 创建虚拟节点
     *
     * @param node
     */
    public void addNode(Node node) {
        String machineIp = node.getMachineIp();
        for (int i = 0; i < virtualCopies; i++) {
            long key = FNVHash(machineIp +"#"+i);
            hashCricle.put(key, node);
        }
        countEveryNode.put(node.getMachineIp(), new AtomicInteger(0));
        return;
    }
    /**
     * 批量新增节点
     *
     * @param nodes
     */
    public void addNodes(Node... nodes) {
        for (int i = 0; i < nodes.length; i++) {
            addNode(nodes[i]);
        }
    }
    /**
     * 删除一个节点
     *
     * @param node
     */
    public void remove(Node node) {
        String machineIp = node.getMachineIp();
        for (int i = 0; i < virtualCopies; i++) {
            long key = FNVHash(machineIp +"#"+i);
            hashCircle.remove(key);
        }
        countEveryNode.remove(node.getMachineIp());
    }
    /**
     * 获取节点
     *
     * @param value 缓存对象
     */
    public Node getNode(String value) {
        long key = FNVHash(value);
        SortedMap<Long, Node> tailMap = hashCircle.tailMap(key);
        long hashKey = tailMap.isEmpty() ? hashCircle.firstKey() : tailMap.firstKey();
        Node node = hashCircle.get(hashKey);
        String machineIp = node.getMachineIp();
        countEveryNode.get(machineIp).incrementAndGet();
        return node;
    }
  
    // 32位的 Fowler-Noll-Vo 哈希算法
    // 比使用java自带的Object.hashCode()散列的均匀
    private static Long FNVHash(String key) {
        final int p = 16777619;
        Long hash = 2166136261L;
        for (int idx = 0, num = key.length(); idx < num; ++idx) {
            hash = (hash ^ key.charAt(idx)) * 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;
    }
}