一致性Hash算法与Java实现

1.简介

1.1 普通hash算法

普通hash算法是通过key将数据映射到具体节点上，如key%N，key为数据的hash值，N为节点数量，如果有机器加入或者退出集群，则key映射失效了，导致数据丢失。

1.2 一致性hash算法

相比普通hash算法，一致性hash就可以解决这种问题。一致性hash是分布式系统常用的一种算法，常用于负载均衡。

2.原理分析

2.1 环形Hash空间

一致性hash算法，将value映射为一个32的key（常用MD5函数），即key分配到一个0~2^32次方的空间里。我们可以想象这些数字头尾相连，形成一个闭合的环形。如下图：

2.2 将数据映射到环形空间上

现在我们假设有4个对象，object1，object2，object3，object4，我们按照一致性Hash算法，将它们的key分散到环形空间上。

Hash(Object1) = key1

Hash(Object2) = key2

Hash(Object3) = key3

Hash(Object4) = key4

如下图

2.2 将机器映射到环形空间上

在采用一致性哈希算法的分布式集群中将新的机器加入，其原理是通过使用与对象存储一样的Hash算法将机器也映射到环中（一般情况下对机器的hash计算是采用机器的IP或者机器唯一的别名作为输入值），然后以顺时针的方向计算，将所有对象存储到离自己最近的机器中。

假设现在有node1，node2，node3三台机器，通过Hash算法得到对应的key值，映射到环中，其示意图如下：

Hash(node1) = key1

Hash(node2) = key2

Hash(node3) = key3

如下图



通过上图可以看出对象与机器处于同一哈希空间中，这样按顺时针转动object1存储到了NODE1中，object3存储到了NODE2中，object2、object4存储到了NODE3中。在这样的部署环境中，hash环是不会变更的，因此，通过算出对象的hash值就能快速的定位到对应的机器中，这样就能找到对象真正的存储位置了。

2.3 节点的加入和删除

2.3.1 节点的加入

如果往集群中添加一个新的节点NODE4，通过对应的哈希算法得到KEY4，并映射到环中，如下图：



通过按顺时针迁移的规则，那么object2被迁移到了NODE4中，其它对象还保持这原有的存储位置。通过对节点的添加和删除的分析，一致性哈希算法在保持了单调性的同时，还是数据的迁移达到了最小，这样的算法对分布式集群来说是非常合适的，避免了大量数据迁移，减小了服务器的的压力。

2.3.2 节点的删除

以上面的分布为例，如果NODE2出现故障被删除了，那么按照顺时针迁移的方法，object3将会被迁移到NODE3中，这样仅仅是object3的映射位置发生了变化，其它的对象没有任何的改动。如下图：



2.3.3 平衡性

根据上面的图解分析，一致性哈希算法满足了单调性和负载均衡的特性以及一般hash算法的分散性，但这还并不能当做其被广泛应用的原由，因为还缺少了平衡性。下面将分析一致性哈希算法是如何满足平衡性的。hash算法是不保证平衡的，如上面只部署了NODE1和NODE3的情况（NODE2被删除的图），object1存储到了NODE1中，而object2、object3、object4都存储到了NODE3中，这样就照成了非常不平衡的状态。在一致性哈希算法中，为了尽可能的满足平衡性，其引入了虚拟节点。

“虚拟节点”（ virtual node ）是实际节点（机器）在 hash 空间的复制品（ replica ），一实际个节点（机器）对应了若干个“虚拟节点”，这个对应个数也成为“复制个数”，“虚拟节点”在 hash 空间中以hash值排列。

以上面只部署了NODE1和NODE3的情况（NODE2被删除的图）为例，之前的对象在机器上的分布很不均衡，现在我们以2个副本（复制个数）为例，这样整个hash环中就存在了4个虚拟节点，最后对象映射的关系图如下：



根据上图可知对象的映射关系：object1->NODE1-1，object2->NODE1-2，object3->NODE3-2，object4->NODE3-1。通过虚拟节点的引入，对象的分布就比较均衡了。那么在实际操作中，正真的对象查询是如何工作的呢？对象从hash到虚拟节点到实际节点的转换如下图：



虚拟节点”的hash计算可以采用对应节点的IP地址加数字后缀的方式。例如假设NODE1的IP地址为192.168.1.100。引入“虚拟节点”前，计算 cache A 的 hash 值：

Hash(“192.168.1.100”);

引入“虚拟节点”后，计算“虚拟节”点NODE1-1和NODE1-2的hash值：

Hash(“192.168.1.100#1”); // NODE1-1

Hash(“192.168.1.100#2”); // NODE1-2

3.Java代码实现

import java.io.UnsupportedEncodingException;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.SortedMap;
import java.util.TreeMap;

/**
*
*/

/**
* 一致性Hash算法-Java实现
*
* @author tangthis
*
* <p>https://github.com/tangthis
*/
public class ConsistencyHash {
private TreeMap<Long,Object> nodes = null;
//真实服务器节点信息
private List<Object> shards = new ArrayList();
//设置虚拟节点数目
private int VIRTUAL_NUM = 4;

/**
* 初始化一致环
*/
public void init() {
shards.add("192.168.0.0-服务器0");
shards.add("192.168.0.1-服务器1");
shards.add("192.168.0.2-服务器2");
shards.add("192.168.0.3-服务器3");
shards.add("192.168.0.4-服务器4");

nodes = new TreeMap<Long,Object>();
for(int i=0; i<shards.size(); i++) {
Object shardInfo = shards.get(i);
for(int j=0; j<VIRTUAL_NUM; j++) {
nodes.put(hash(computeMd5("SHARD-" + i + "-NODE-" + j),j), shardInfo);
}
}
}

/**
* 根据key的hash值取得服务器节点信息
* @param hash
* @return
*/
public Object getShardInfo(long hash) {
Long key = hash;
SortedMap<Long, Object> tailMap=nodes.tailMap(key);
if(tailMap.isEmpty()) {
key = nodes.firstKey();
} else {
key = tailMap.firstKey();
}
return nodes.get(key);
}

/**
* 打印圆环节点数据
*/
public void printMap() {
System.out.println(nodes);
}

/**
* 根据2^32把节点分布到圆环上面。
* @param digest
* @param nTime
* @return
*/
public long hash(byte[] digest, int nTime) {
long rv = ((long) (digest[3+nTime*4] & 0xFF) << 24)
| ((long) (digest[2+nTime*4] & 0xFF) << 16)
| ((long) (digest[1+nTime*4] & 0xFF) << 8)
| (digest[0+nTime*4] & 0xFF);

return rv & 0xffffffffL; /* Truncate to 32-bits */
}

/**
* Get the md5 of the given key.
* 计算MD5值
*/
public byte[] computeMd5(String k) {
MessageDigest md5;
try {
md5 = MessageDigest.getInstance("MD5");
} catch (NoSuchAlgorithmException e) {
throw new RuntimeException("MD5 not supported", e);
}
md5.reset();
byte[] keyBytes = null;
try {
keyBytes = k.getBytes("UTF-8");
} catch (UnsupportedEncodingException e) {
throw new RuntimeException("Unknown string :" + k, e);
}

md5.update(keyBytes);
return md5.digest();
}

public static void main(String[] args) {
Random ran = new Random();
ConsistencyHash hash = new ConsistencyHash();
hash.init();
hash.printMap();
//循环50次，是为了取50个数来测试效果，当然也可以用其他任何的数据来测试
for(int i=0; i<50; i++) {
System.out.println(hash.getShardInfo(hash.hash(hash.computeMd5(String.valueOf(i)),ran.nextInt(hash.VIRTUAL_NUM))));
}
}
}