import com.zliot.zltom.common.Constants;

import java.util.ArrayList;
import java.util.List;
import java.util.SortedMap;
import java.util.TreeMap;


/**
 * 
 * 一致性HASH演算法
 * 
 * @author zhuhy 
 * @version zltom1.0.0,2021-7-5
 * @since zltom1.0.0
 */
public class ConsistentHash
{
   //一臺物理伺服器虛擬為多少個虛擬伺服器節點
   private final static int VIRTUAL_NODE_NUM = 150;
   
   //一致性HASH環
   private final SortedMap<Integer, String> hashCircleNodes = 
         new TreeMap<Integer, String>();
   
   /**
    * 構造方法
    * 根據節點的HASH值，將伺服器節點放置到HASH環上
    * 
    * @param nodeNames 真實服務節點名稱列表
    */
    public ConsistentHash(List<String> nodeNames) 
    {
        for (String nodeName : nodeNames)
        {
            addNode(nodeName);
        }
    }
     
    
    /**
     * 增加一臺物理伺服器節點
     * @param nodeName
     */
    public void addNode(String nodeName) 
    {
       //往環中增加對應的虛擬節點
        for (int i = Constants.NUM_ZERO; i < VIRTUAL_NODE_NUM; i++)
        {
            hashCircleNodes.put(getHashCode(nodeName + i), nodeName);
        }
    }
     
    
    /**
     * 刪除一臺物理伺服器節點
     * @param nodeName
     */
    public void removeNode(String nodeName) 
    {
       //在環中刪除對應的虛擬節點
        for (int i = Constants.NUM_ZERO; i < VIRTUAL_NODE_NUM; i++) 
        {
            hashCircleNodes.remove(getHashCode(nodeName + i));
        }
    }
     
    
    /**
     * 根據KEY獲取路由節點名稱
     * @param key 關鍵字
     * @return
     */
    public String getNode(String key) 
    {
        if(hashCircleNodes.isEmpty())
        {
            return null;
        }

        int hashCode = getHashCode(key);
        System.out.println("hashcode:" + hashCode);

        //如果沒有命中節點，則從環中獲取最近的節點KEY
        if (!hashCircleNodes.containsKey(hashCode))
        {
            hashCode = getNearestKey(hashCode);
        }

        return hashCircleNodes.get(hashCode);
    }


    /**
     * 從HASH環中獲取最接近的KEY
     * @param key
     * @return
     */
    private Integer getNearestKey(Integer key)
    {
        Integer nearestKey;
        if (hashCircleNodes.containsKey(key))
        {
            nearestKey = key;
        }
        else
        {
            SortedMap<Integer, String> tailMap = hashCircleNodes.tailMap(key);
            SortedMap<Integer, String> headMap = hashCircleNodes.headMap(key);
            if(null == tailMap)
            {
                nearestKey = hashCircleNodes.firstKey();
            }
            else if(null == headMap)
            {
                nearestKey = hashCircleNodes.lastKey();
            }
            else
            {
                if((headMap.lastKey() - key) < (key -tailMap.firstKey()))
                {
                    nearestKey = headMap.lastKey();
                }
                else
                {
                    nearestKey = tailMap.firstKey();
                }
            }
        }

        return nearestKey;
    }
    
    
    /**
     * 獲取輸入資料的HASH值
     * @param data
     * @return
     */
    public int getHashCode(String data)
    {
        //使用FNV HASH演算法，使得接近的資料內容可以得到相差較大的HASH值
        //return data.hashCode();
        return fnvHash(data);
    }


    /**
     * 改進的32位FNV演算法1
     * https://en.wikipedia.org/wiki/Fowler–Noll–Vo_hash_function
     * @param data 字串
     * @return hash結果
     */
    public int fnvHash(String data) {
        final int p = 16777619;
        int hash = (int) 2166136261L;
        for (int i = 0; i < data.length(); i++) {
            hash = (hash ^ data.charAt(i)) * p;
        }
        hash += hash << 13;
        hash ^= hash >> 7;
        hash += hash << 3;
        hash ^= hash >> 17;
        hash += hash << 5;
        return Math.abs(hash);
    }
    


   /**
    * 主方法測試
    * @param args
    */
   public static void main(String[] args)
   {
      List<String> nodeNames = new ArrayList<String>();
      nodeNames.add("伺服器組1");
      nodeNames.add("伺服器組2");
      nodeNames.add("伺服器組3");
      nodeNames.add("伺服器組4");
   
      ConsistentHash consistentHash = new ConsistentHash(nodeNames);

      System.out.println("伺服器組1：" + consistentHash.fnvHash("伺服器組1"));
      System.out.println("伺服器組2：" + consistentHash.fnvHash("伺服器組2"));
      System.out.println("伺服器組3：" + consistentHash.fnvHash("伺服器組3"));
      System.out.println("伺服器組4：" + consistentHash.fnvHash("伺服器組4"));
       
      String key = "dc48a8b8-4569-4743-9b40-3475f960bf29";
      System.out.println(consistentHash.getNode(key));

      String key1 = "dc48a8b8-c301-4743-9b40-3475f960bf29";
      System.out.println(consistentHash.getNode(key1));
       
      String key2 = "dc48a8b8-c301-4743-9865-3475f960be26";
      System.out.println(consistentHash.getNode(key2));
       
      String key3 = "cde9ceee-483e-4b23-a2b1-e63bf6976d20";
      System.out.println(consistentHash.getNode(key3));

      String key4 = "e048a8b8-c301-4743-9b40-3475f960bf29";
      System.out.println(consistentHash.getNode(key4));
       
      String key5 = "5558a8b8-c301-4743-9b40-3475f960bf29";
      System.out.println(consistentHash.getNode(key5));
      System.out.println("============================");

      consistentHash.removeNode("伺服器組4");

      System.out.println(consistentHash.getNode(key));
      System.out.println(consistentHash.getNode(key1));
      System.out.println(consistentHash.getNode(key2));
      System.out.println(consistentHash.getNode(key3));
      System.out.println(consistentHash.getNode(key4));
      System.out.println(consistentHash.getNode(key5));
      System.out.println("============================");
      consistentHash.addNode("伺服器組4");
      System.out.println(consistentHash.getNode(key));
      System.out.println(consistentHash.getNode(key1));
      System.out.println(consistentHash.getNode(key2));
      System.out.println(consistentHash.getNode(key3));
      System.out.println(consistentHash.getNode(key4));
      System.out.println(consistentHash.getNode(key5));
      System.out.println("============================");
      consistentHash.addNode("伺服器組5");
      System.out.println(consistentHash.getNode(key));
      System.out.println(consistentHash.getNode(key1));
      System.out.println(consistentHash.getNode(key2));
      System.out.println(consistentHash.getNode(key3));
      System.out.println(consistentHash.getNode(key4));
      System.out.println(consistentHash.getNode(key5));
      System.out.println("============================");
   }
}