一致性hash

一致性Hash的应用场景

我们先假设一种场景，有一个应用的请求量非常大，在服务端有若干台缓存服务器，这些缓存服务器记录了各用户请求的缓存数据（当然这样做的目的我想大家应该清楚）。怎样将这么多的请求按照一定的规则分发到不同的缓存服务器就显得非常关键。

一种简单的调度规则是通过求余的方式将请求关联到某一台服务器，比如说有10台服务器，那么通过请求的hash值对10求余就关联到了其中的一台服务器。看起来这种方法还ok，但是运行中的服务器难免有挂掉的风险，或者任务分发后服务器的负载仍然很大致使我们不得不增加服务器，遇到这样类似的情况，如果我们还是通过上述方法进行调度的话就会打乱原来的请求映射，比如请求A之前关联的是服务器1001，当服务器数量发生改变之后，通过求余，很可能A就关联到了其他的服务器，而这是我们不能接受的，我们希望的是服务器的增减尽量不影响用户之前的缓存映射。怎么办？一致性Hash就能够较好的解决这个问题。

一致性Hash的原理

整体来看，有两个基本空间，用户空间和服务空间，我们的目的是将这两个空间关联起来，这里我们引入hash空间，将用户空间和服务空间都映射到hash空间，这样两者就关联起来了。将服务器映射到hash空间，按照hash值由小到大顺时针排列，并且首尾相连，形成一个环形。当用户请求过来的时候，也将其映射到hash空间上，然后将其关联到顺时针最近的服务器即可。但有一个问题是，比如说当其中一个服务器节挂掉时，按照这个规则，尽管之前关联到其他服务器的请求的关联性不会发生改变，但关联到该服务器的请求就全部关联到了另一台服务器。这样势必会增大后者的负载，不甚合理。
换一种方式，现在我们不直接将服务器映射到hash空间，而是基于每一台服务器虚拟出诸多虚拟服务器（100台或者更多），我们将这些虚拟服务器映射到hash空间，这样归属不同服务器的各虚拟服务器就交织在hash空间中，当请求过来时先关联到某个虚拟服务器，然后再将虚拟服务器关联到其所属的真实服务器。当其中的一台服务器挂掉时，之前归属到这台服务器的请求就可能关联到其他任意服务器，这样就实现了负载均衡，即所谓的一致性hash。

代码示例

首先定义服务器节点：

    public static class Node{
        String name;
        String ip;
        public Node(String name, String ip){
            this.name = name;
            this.ip = ip;
        }
        public String toString(){
            return this.name+"-"+this.ip;
        }
    }

然后是测试类：（这里有两个关键点，一个是hash环的管理，另一个是hash函数的选取。本实例用的是murmur哈希）

    public class Shard<Node> {
        private static TreeMap<Long, Node> nodes;//虚拟主机hash环
        private static TreeMap<Long, Node> treeKey;//客户机主机映射
        private List<Node> mNodes = new ArrayList<Node>();//实体主机列表
        private final Integer vNodes = 100;//一个实体对应虚拟主机个数
        public Shard(List<Node> mNodes){
            this.mNodes = mNodes;
            init();
        }
        //main函数
        public static void main(String[] args){
            Node n1 = new Node("n1", "172.16.226.111");
            Node n2 = new Node("n2", "172.16.226.112");
            Node n3 = new Node("n3", "172.16.226.113");
            Node n4 = new Node("n4", "172.16.226.114");
            List<Node> nodes = new ArrayList<Node>();
            nodes.add(n1);
            nodes.add(n2);
            nodes.add(n3);
            nodes.add(n4);
            Shard<Node> shardN = new Shard<Node>(nodes);
            System.out.println("添加客户端，一开始有4个主机，分别为n1,n2,n3,n4,每个主机有100个虚拟主机：");
            shardN.keyToNode("1111号客户端");
            shardN.keyToNode("1112号客户端");
            shardN.keyToNode("1113号客户端");
            shardN.keyToNode("1114号客户端");
            shardN.keyToNode("1115号客户端");
            shardN.keyToNode("1116号客户端");
            shardN.keyToNode("1117号客户端");
            shardN.keyToNode("1118号客户端");
            shardN.keyToNode("1119号客户端");
            shardN.deleteNode(n2);//删除节点n2
            Node n5 = new Node("n3", "172.16.226.115");
            shardN.addNode(n5);//增加节点n5
        }
        //构建hash环
        public void init (){
            nodes = new TreeMap<Long, Node>();
            treeKey = new TreeMap<Long, Node>();
            for(Node n : mNodes){
                for(Integer i=0;i<vNodes;i++){
                    nodes.put(hash("SHARD-"+n.name+"-NODE-"+i), n);
                }
            }
        }
        //删除节点
        public  void deleteNode(Node node){
            System.out.println("删除节点："+node);
            for(int i=0;i<vNodes;i++){
                Long key = hash("SHARD-"+node.name+"-NODE-"+i);
                //nodes.remove(key);
                SortedMap<Long,Node> headNodes = nodes.headMap(key);
                SortedMap<Long,Node> tailNodes = nodes.tailMap(key);
                SortedMap<Long,Node> betweenNodes;
                if(headNodes.size() == 0){
                    betweenNodes = treeKey.tailMap(tailNodes.lastKey());
                }else{
                    betweenNodes = treeKey.subMap(headNodes.lastKey(), key);
                }
                nodes.remove(tailNodes.firstKey());
                tailNodes.remove(tailNodes.firstKey());
                Iterator iterator = betweenNodes.keySet().iterator();
                while(iterator.hasNext()){
                    Long keytemp = (Long)iterator.next();
                    if(tailNodes.size() == 0){
                        treeKey.put(keytemp, headNodes.get(headNodes.firstKey()));
                        System.out.println(keytemp + "连接转移到->" + headNodes.get(headNodes.firstKey()));
                    }else{
                        treeKey.put(keytemp, tailNodes.get(tailNodes.firstKey()));
                        System.out.println(keytemp + "连接转移到->" + tailNodes.get(tailNodes.firstKey()));
                    }
                }
            }
        }
        //增加节点
        public void addNode(Node node){
            System.out.println("增加节点："+node);
            for(int i=0;i<vNodes;i++){
                Long key = hash("SHARD-"+node.name+"-NODE-"+i);
                SortedMap<Long,Node> headNodes = nodes.headMap(key);
                SortedMap<Long,Node> tailNodes = nodes.tailMap(key);
                SortedMap<Long,Node> betweenNodes;
                if(headNodes.size() == 0){
                    betweenNodes = treeKey.tailMap(nodes.lastKey());
                }else{
                    betweenNodes = treeKey.subMap(headNodes.lastKey(), key);
                }
                nodes.put(key, node);
                Iterator iterator = betweenNodes.keySet().iterator();
                while(iterator.hasNext()){
                    Long keytemp = (Long)iterator.next();
                    System.out.println(keytemp + "连接转移到->" + node);
                }
            }
        }
        //将请求关联到服务器
        public void keyToNode(String key){
            SortedMap<Long, Node> tail = nodes.tailMap(hash(key));
            if(tail.size() == 0){
                treeKey.put(hash(key), nodes.get(nodes.firstKey()));
                System.out.println(key+"（hash："+hash(key)+"）连接到主机->"+nodes.get(nodes.firstKey()));
                return;
            }
            treeKey.put(hash(key), tail.get(tail.firstKey()));
            System.out.println(key+"（hash："+hash(key)+"）连接到主机->"+tail.get(tail.firstKey()));
        }
        //murmur  Hash
        private static 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;  
        }
    }

参考文章

应用于负载均衡的一致性哈希及java实现——haitao111313
一致性哈希算法与Java实现
Hash函数概览
一致性哈希算法（用于解决服务器均衡问题）——caigen1988