第十一章 自己实现一致性hash算法

关于一致性hash算法的意义以及其相对于简单求余法（除数求余法）的好处，查看第六章 memcached剖析

注意：真实的hash环的数据结构是二叉树，这里为了简便使用了列表List

1、一致性hash算法的使用地方

• memcached服务器
• Jedis分片机制

2、真实服务器节点没有虚拟化的一致性hash算法实现

ServerNode：真实服务器节点

package hash;

/**
 * server节点
 */
public class ServerNode {
    private String serverName;
    private long serverHash;

    public String getServerName() {
        return serverName;
    }

    public void setServerName(String serverName) {
        this.serverName = serverName;
    }

    public long getServerHash() {
        return serverHash;
    }

    public void setServerHash(long serverHash) {
        this.serverHash = serverHash;
    }

    /**
     * 下边重写hashcode()和equals()是为了在删除节点的时候只根据传入的serverName删除即可
     */
    @Override
    public int hashCode() {
        final int prime = 31;
        int result = 1;
        result = prime * result
                + ((serverName == null) ? 0 : serverName.hashCode());
        return result;
    }

    @Override
    public boolean equals(Object obj) {
        if (this == obj)
            return true;
        if (obj == null)
            return false;
        if (getClass() != obj.getClass())
            return false;
        ServerNode other = (ServerNode) obj;
        if (serverName == null) {
            if (other.serverName != null)
                return false;
        } else if (!serverName.equals(other.serverName))
            return false;
        return true;
    }
    
    
}

注意：

• serverName可以自己取名，这里取名为"ip:port"
• 对于hashCode()和equals()方法的重写仅仅是为了删除服务器节点的时候，只根据serverName就可以删除，而不需要再计算服务器节点的hash值

ServerComparator：真实服务器比较器

package hash;

import java.util.Comparator;

/**
 * 服务器排序比较器
 */
public class ServerComparator implements Comparator<ServerNode> {

    public int compare(ServerNode node1, ServerNode node2) {
        if(node1.getServerHash() <= node2.getServerHash()) {
            return -1;//node1<node2
        }
        return 1;//node1>node2
    }

}

注意：

• 关于java的比较器，有两种：（后者用的多一些）
  • javabean实现comparable接口，实现compareTo()方法
  • 另外建一个类实现comparator接口，实现其中的compare()方法

ConsistentHash：一致性hash实现类

package hash;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.zip.CRC32;

/**
 * 一致性hash实现（数据结构：list）（服务器没有虚拟化）
 * 一致性hash的真正数据结构是二叉树
 */
public class ConsistentHash {
    private List<ServerNode> servers = new ArrayList<ServerNode>();//存放服务器
    
    /** 计算服务器和存储的键的hash值 */
    public long hash(String str){
        CRC32 crc32 = new CRC32();
        crc32.update(str.getBytes());
        return crc32.getValue();
    }
    
    /**
     * 添加server到环上
     * @param serverName ip:port
     */
    public void addServer(String serverName){

        ServerNode node = new ServerNode();
        node.setServerName(serverName);
        node.setServerHash(hash(serverName));
        
        servers.add(node);
        Collections.sort(servers, new ServerComparator());
    }
    
    /**
     * 从环上删除server节点
     */
    public void deleteServer(String serverName){

        ServerNode node = new ServerNode();
        node.setServerName(serverName);
        
        servers.remove(node);
    }
    
    /**
     * 获取一个缓存key应该存放的位置
     * @param cachekey 缓存的key
     * @return 缓存的服务器节点
     */
    public ServerNode getServer(String cachekey){
        long keyHash = hash(cachekey);
        
        for(ServerNode node : servers){
            if(keyHash<=node.getServerHash()){
                return node;
            }
        }
        
        return servers.get(0);//如果node没有合适放置位置，放在第一台服务器上去
    }
    
    /****************测试*******************/
    public void printServers(){
        for(ServerNode server : servers){
            System.out.println(server.getServerName()+"-->"+server.getServerHash());
        }
    }
    
    public static void main(String[] args) {
        ConsistentHash ch = new ConsistentHash();
        ch.addServer("127.0.0.1:11211");
        ch.addServer("127.0.0.1:11212");
        ch.addServer("127.0.0.2:11211");
        ch.addServer("127.0.0.2:11212");
        
        ch.printServers();
        
        ServerNode node = ch.getServer("hello");
        System.out.println(ch.hash("hello")+"-->"+node.getServerName()+"-->"+node.getServerHash());
        
        ServerNode node2 = ch.getServer("name");
        System.out.println(ch.hash("name")+"-->"+node2.getServerName()+"-->"+node2.getServerHash());
        
        ServerNode node3 = ch.getServer("a");
        System.out.println(ch.hash("a")+"-->"+node3.getServerName()+"-->"+node3.getServerHash());
        
        /********************删除节点*********************/
        ch.deleteServer("127.0.0.1:11212");
        ch.printServers();
        
        ServerNode node0 = ch.getServer("hello");
        System.out.println(ch.hash("hello")+"-->"+node0.getServerName()+"-->"+node0.getServerHash());
        
        ServerNode node02 = ch.getServer("name");
        System.out.println(ch.hash("name")+"-->"+node02.getServerName()+"-->"+node02.getServerHash());
        
        ServerNode node03 = ch.getServer("a");
        System.out.println(ch.hash("a")+"-->"+node03.getServerName()+"-->"+node03.getServerHash());
        
    }
}

注意：

• 在计算服务器节点和存储的key的hash值的时候，不仅仅可以使用crc32算法，还可以使用MD5算法等等，只要是最后得出的结果是一个>=0&&<=2³²的数就好
• 在这个实现中，并没有将真实服务器节点进行虚拟化

3、真实服务器节点虚拟化后的一致性hash算法实现

为什么要虚拟化，查看第六章 memcached剖析 ，这里只列出几条原因：

• 在memcached服务器较少的情况下，很难平均的分布到hash环上，这样就会造成负载不均衡--引入虚拟化节点，可以解决这个问题
• 当一台memcached宕机时，其原先所承受的压力全部给了其下一个节点，为了将其原先所承受的压力尽可能的分布给所有剩余的memcached节点，引入虚拟化节点可以达到这个目的
• 当新添加了一台memcached服务器server1时，server1只会缓解其中的一台服务器（即server1插入环后，server1的下一个节点）的压力，为了可以让server1尽可能的缓解所有的memcached服务器的压力，引入虚拟节点可以达到这个目的

VirtualServerNode：虚拟节点

package hash2;

/**
 * 虚拟节点
 */
public class VirtualServerNode {
    private String serverName;//真实节点名称
    private long virtualServerHash;//虚拟节点hash

    public String getServerName() {
        return serverName;
    }

    public void setServerName(String serverName) {
        this.serverName = serverName;
    }

    public long getVirtualServerHash() {
        return virtualServerHash;
    }

    public void setVirtualServerHash(long virtualServerHash) {
        this.virtualServerHash = virtualServerHash;
    }

}

注意：

• 该类中的serverName是该虚拟节点对应的真实节点的名称，这里就是"ip:port"
• 真正的虚拟节点的名称是serverName-i（其中，i是0~virtualCount的整数值），这一块儿请查看ConsistentHashWithVirtualNode的addServer(String serverName)

VirtualServerComparator：虚拟节点比较器

package hash2;

import java.util.Comparator;

/**
 * 虚拟节点比较器
 */
public class VirtualServerComparator implements Comparator<VirtualServerNode> {

    public int compare(VirtualServerNode node1, VirtualServerNode node2) {
        if(node1.getVirtualServerHash() <= node2.getVirtualServerHash()) {
            return -1;
        }
        return 1;
    }

}

ConsistentHashWithVirtualNode：真实节点虚拟化后的一致性hash算法

package hash2;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.zip.CRC32;

/**
 * 具有虚拟节点的一致性hash实现（数据结构：list）
 * 一致性hash的真正数据结构是二叉树
 */
public class ConsistentHashWithVirtualNode {
    private List<VirtualServerNode> virtualServers = new ArrayList<VirtualServerNode>();//存放虚拟节点
    private static final int virtualCount = 8;//每个真实节点虚拟成8个虚拟节点
    
    /** 计算服务器和存储的键的hash值 */
    public long hash(String str){
        CRC32 crc32 = new CRC32();
        crc32.update(str.getBytes());
        return crc32.getValue();
    }
    
    /**
     * 添加server的虚拟节点到环上
     * @param serverName ip:port
     */
    public void addServer(String serverName){

        for(int count=0;count<virtualCount;count++){
            VirtualServerNode node = new VirtualServerNode();
            node.setServerName(serverName);
            node.setVirtualServerHash(hash(serverName+"-"+count));//虚拟节点的名字：serverName+"-"+count
            virtualServers.add(node);
        }
        
        Collections.sort(virtualServers, new VirtualServerComparator());
    }
    
    /**
     * 从环上删除server节点（需要删除所有的该server节点对应的虚拟节点）
     */
    public void deleteServer(String serverName){
        
        /*
         * 在这种删除的时候，会出现java.util.ConcurrentModificationException
         * 这是因为此处的遍历方式为使用ArrayList内部类Itr进行遍历，
         * 在遍历的过程中发生了remove、add等操作，导致modCount发生了变化，
         * 产生并发修改异常，
         * 可以使用下边的那一种方式来进行遍历（遍历方式不是Itr），
         * 再这样的遍历过程中，add和remove都是没有问题的
         */
        /*for(VirtualServerNode node : virtualServers){
            if(node.getServerName().equals(serverName)){
                virtualServers.remove(node);
            }
        }*/
        for(int i=0;i<virtualServers.size();i++) {
            VirtualServerNode node = virtualServers.get(i);
            if(node.getServerName().equals(serverName)) {
                virtualServers.remove(node);
            }
        }
        
    }
    
    /**
     * 获取一个缓存key应该存放的位置
     * @param cachekey 缓存的key
     * @return 缓存的服务器节点
     */
    public VirtualServerNode getServer(String cachekey){
        long keyHash = hash(cachekey);
        
        for(VirtualServerNode node : virtualServers){
            if(keyHash<=node.getVirtualServerHash()){
                return node;
            }
        }
        
        return virtualServers.get(0);//如果node没有合适放置位置，放在第一台服务器上去
    }
    
    /****************测试*******************/
    public void printServers(){
        for(VirtualServerNode server : virtualServers){
            System.out.println(server.getServerName()+"-->"+server.getVirtualServerHash());
        }
    }
    
    public static void main(String[] args) {
        ConsistentHashWithVirtualNode ch = new ConsistentHashWithVirtualNode();
        ch.addServer("127.0.0.1:11211");
        ch.addServer("127.0.0.1:11212");
        ch.addServer("127.0.0.2:11211");
        ch.addServer("127.0.0.2:11212");
        
        ch.printServers();
        
        VirtualServerNode node = ch.getServer("hello");
        System.out.println(ch.hash("hello")+"-->"+node.getServerName()+"-->"+node.getVirtualServerHash());
        
        VirtualServerNode node2 = ch.getServer("name");
        System.out.println(ch.hash("name")+"-->"+node2.getServerName()+"-->"+node2.getVirtualServerHash());
        
        VirtualServerNode node3 = ch.getServer("a");
        System.out.println(ch.hash("a")+"-->"+node3.getServerName()+"-->"+node3.getVirtualServerHash());
        
        /*********************删除节点之后**********************/
        ch.deleteServer("127.0.0.1:11212");
        ch.printServers();
        
        VirtualServerNode node0 = ch.getServer("hello");
        System.out.println(ch.hash("hello")+"-->"+node0.getServerName()+"-->"+node0.getVirtualServerHash());
        
        VirtualServerNode node02 = ch.getServer("name");
        System.out.println(ch.hash("name")+"-->"+node02.getServerName()+"-->"+node02.getVirtualServerHash());
        
        VirtualServerNode node03 = ch.getServer("a");
        System.out.println(ch.hash("a")+"-->"+node03.getServerName()+"-->"+node03.getVirtualServerHash());
        
    }
}

注意：

• 在实际操作中，一台memcached服务器虚拟成150台比较合适（100~200）
• 从环上删除节点的算法写的较差，但是考虑到删除节点的操作在实际使用中用的比较少（宕机比较少，人为的删除节点也较少），也无所谓
• 删除节点的时候，注意使用foreach语法糖去遍历的时候，在遍历的过程中不可以做删除、增加操作，否则会抛出并发修改异常，具体的原因见注释和第二章 ArrayList源码解析；想要实现在遍历的过程中进行删除、增加操作，使用简单for循环，见如上代码