一致性hash算法的实现（用作负载均衡场景节点的选取）

/**
* Copyright (c) AVIT LTD (2016). All Rights Reserved.
* Welcome to www.avit.com.cn
*/
package com.avit.ipmp.common.utils;


import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.util.ArrayList;
import java.util.List;
import java.util.SortedMap;
import java.util.Timer;
import java.util.TimerTask;
import java.util.TreeMap;

import com.avit.ipmp.nodeManager.beans.NodeInfo;

public class SharedMqttNode<T> {

private TreeMap<Long, T> nodes = new TreeMap<Long, T>();     // 虚拟节点  
    private List<T> shards; // 真实机器节点  
    private final int NODE_NUM = 5; // 每个机器节点关联的虚拟节点个数  
 
    public SharedMqttNode(List<T> shards) { 
        super(); 
        this.shards = shards; 
        init(); 
    } 
 
    private void init() { // 初始化一致性hash环  
        for (int i = 0; i != shards.size(); ++i) { // 每个真实机器节点都需要关联虚拟节点  
            for (int n = 0; n < NODE_NUM; n++) 
                // 一个真实机器节点关联NODE_NUM个虚拟节点  
                nodes.put(hash(shards.get(i).hashCode()+n+""), shards.get(i)); 
 
        } 
    }
   
   
    public void add(T node) {   
    this.shards.add(node);
        for (int i = 0; i < NODE_NUM; i++) {   
        // 一个真实机器节点关联NODE_NUM个虚拟节点  
            nodes.put(hash(node.hashCode()+i+""), node); 
       }   
}   

  public void remove(T node) {   
        for (int i = 0; i < NODE_NUM; i++) {   
              nodes .remove(hash(node.hashCode()+i+""));   
       }
        this.shards.remove(node);
}   
   
 
    public T getShardInfo(String key) { 
        SortedMap<Long, T> tail = nodes.tailMap(hash(key)); // 沿环的顺时针找到一个虚拟节点  
        if (tail.size() == 0) { 
            return nodes.get(nodes.firstKey()); 
        } 
        return tail.get(tail.firstKey()); // 返回该虚拟节点对应的真实机器节点的信息  
       
//        Entry en = nodes.ceilingEntry(hash(key)); 
//        if (en == null) { 
//           return (T) nodes.firstEntry().getValue(); 
//        } 
//        return (T) en.getValue(); 
    } 
   
   
    public List<T> getReal() { 
         return shards; 
    } 

 
    /**
     *  MurMurHash算法，是非加密HASH算法，性能很高，
     *  比传统的CRC32,MD5，SHA-1（这两个算法都是加密HASH算法，复杂度本身就很高，带来的性能上的损害也不可避免）
     *  等HASH算法要快很多，而且据说这个算法的碰撞率很低.
     *  http://murmurhash.googlepages.com/
     */ 
    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); 
            // for big-endian version, do this first:  
            // finish.position(8-buf.remaining());  
            finish.put(buf).rewind(); 
            h ^= finish.getLong(); 
            h *= m; 
        } 
 
        h ^= h >>> r; 
        h *= m; 
        h ^= h >>> r; 
 
        buf.order(byteOrder); 
        return h; 
    } 

   
    public static void main(String[] args) throws InterruptedException {

    List<NodeInfo> list = new ArrayList<NodeInfo>();
    NodeInfo ni =null;;
   
   
    ni = new NodeInfo();
    ni.setNodeIP("192.168.32.17");
    ni.setMqttPort(8088);
    list.add(ni);
   
    ni = new NodeInfo();
    ni.setNodeIP("192.168.32.18");
    ni.setMqttPort(8088);
    list.add(ni);
   
    ni = new NodeInfo();
    ni.setNodeIP("192.168.32.19");
    ni.setMqttPort(8088);
    list.add(ni);
   
    ni = new NodeInfo();
    ni.setNodeIP("192.168.2.20");
    ni.setMqttPort(8088);
    list.add(ni);
   
    ni.setNodeIP("192.168.33.20");
    ni.setMqttPort(8088);
    list.add(ni);
   
    ni = new NodeInfo();
    ni.setNodeIP("192.168.35.20");
    ni.setMqttPort(8088);
    list.add(ni);
   
    ni = new NodeInfo();
    ni.setNodeIP("192.168.36.23");
    ni.setMqttPort(8088);
    list.add(ni);
   
    ni = new NodeInfo();
    ni.setNodeIP("192.168.3.20");
    ni.setMqttPort(8088);
    list.add(ni);
   
        final SharedMqttNode<NodeInfo> sn = new SharedMqttNode<NodeInfo>(list);
   
       TimerTask task = new TimerTask() {
@Override
public void run() {

        String[] nodes = {"9922000000230198", "9320010864623470", "9922000002481525"};
    for (int i = 0; i < nodes.length; i++){
                System.out.println("[" + nodes[i] + "]的hash值为" +
                sn.hash(nodes[i]) + ", 被路由到结点[" + sn.getShardInfo(nodes[i]).getNodeIP() + "]");

    }
    System.out.println("<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<");

    System.out.println("node size======="+sn.getReal().size());
   
    NodeInfo ni = new NodeInfo();
    ni.setNodeIP("192.168.5.17");
    ni.setMqttPort(8080);
    sn.add(ni);
    System.out.println("node size >>>>>>>>>"+sn.getReal().size());
}

};
Timer timer = new Timer(); 
timer.schedule(task, 1000, 5000); 
   
//        final SharedMqttNode<NodeInfo> sn = new SharedMqttNode<NodeInfo>(list);
//        ExecutorService es = Executors.newCachedThreadPool(); 
//        final CountDownLatch cdl = new CountDownLatch(1000); 
//        // 1000个线程 
//        for (int j = 0; j < 1000; j++) { 
//            es.execute(new Runnable() { 
// 
//                @Override 
//                public void run() { 
//                    // Random rd = new Random(1100); 
//                    for (int k = 0; k < 10000; k++) { 
//                        sn.getShardInfo(String.valueOf(Math.random())).inc(); 
//                    } 
//                    cdl.countDown(); 
//                } 
//            }); 
//        } 
// 
//        // 等待所有线程结束 
//        cdl.await(); 
//        List<NodeInfo> nodeList = sn.getReal(); 
//        for (NodeInfo node : nodeList) { 
//            System.out.println("node" + node.getNodeIP()+ ":" + node.getCount()); 
//        } 
   
}
}