一致性哈希算法及其实现（Consistent Hashing）
一，一致性哈希算法的原理
1，一致性哈希算法诞生的背景
技术和业务是相互推动，共同前进的。一致性哈希算法的产生也源于业务的需求。随着业务的增长，一台单机
已经不能满足业务的需要，分布式架构应运而生。分布式环境下，多台机器需要协同作业，如果保证数据在分布式
环境下的一致性，就成为了亟待解决的问题。一致性哈希算法，就是为了解决多台机器，在动态增删的情况下，能够
最大限度地保证信息的一致性。
一致性哈希算法是一种分布式哈希算法，设计目标是为了解决互联网中的热点(Hot spot)问题。一致性哈希算法
设计初衷和CARP十分类似。CARP，即Composition/Aggregation Principle，组合/聚合原则。CARP的目标之一，是为
了改善服务的可用性。在多台服务器环境下，进行故障转移，提高系统的可用性。一致性哈希修正了CARP使用的简单
哈希算法带来的问题，使得分布式哈希（DHT）可以在P2P环境中真正得到应用。
2，一致性哈希算法实现参考标准
设计一致性hash算法，一般需要遵循以下几个标准：
2.1、平衡性(Balance)：平衡性不仅仅指的是平均分配，可以理解为一种加权的平均，根据每台服务器的能力，把任务
分配下去，充分利用每台机器的资源。
2.2、单调性(Monotonicity)：如何理解单调性？是一个难点。网上的说法基本上都是“拿来主义”，并没有把单调性的
概念说清楚。如果仅仅从概念上来说单调性，很难说清楚。我们可以换一个角度来解读“单调性”，从单调性要解决的
问题入手，解读“单调性”，或者说从单调性的功能入手，解读“单调性”。在动态变化的分布式环境中，增加服务器节点
和移除服务器节点是最常见的操作。如果我们采用简单的哈希算法，比如使用节点的IP的哈希值hash%节点数N，做为哈
希值，映射到节点上。那么，一旦节点数发生了变化，所以的哈希值都会失效。单调性就是为了解决这个问题的。
2.3、分散性(Spread)：在分布式环境中，节点A有可能看不到其他所有的N-1个节点，只看到其中的一部分节点。当节
点A将数据映射到其他节点时，由于不同节点所见的集群范围有可能不同，从而导致哈希的结果不一致，最终相同的
数据被不同的节点映射到不同的内存中。这种情况显然是应该避免的，因为它导致相同的数据被存储到不同的节点，降低
了系统存储的效率。分散性目前不是很理解，为什么相同的数据会映射到不同的节点呢？留着以后研究。
2.4、负载(Load)：负载的概念看了网上的说法，也不是很明白，留着以后研究。
二，一致性哈希算法的创新点
一致性哈希算法的实现过程，其实就是为了解决上述问题的过程。我们这里不按照网上的方式进行枯燥的讲解，而是
通过对比，进行有针对性的讲解。一致性哈希算法，在不同的系统环境下，具有不同的实现方式。但是，实现的大致过程
还是一致的。
1，静态映射 --> 动态映射
普通的哈希算法，比如上面提到的（hash % N），由于数据和节点是静态绑定的。也就是说，进行哈希运算后，数据
和节点之间的关系就确定了。一旦节点数发生变化，所有的哈希都失效了。一致性哈希算法，是如何解决这个问题的呢？
一致性哈希算法引入了环的概念，并且最关键的创新点是：将节点的分配和数据的分配，拆分成了2个独立的过程。数据
和节点的关联，不是通过哈希算法直接建立起来的。这样数据和节点就相对独立了，节点A的变化，并不会影响到整个分布
式系统，因为此时不需要对所有数据进行哈希运算。
一致性哈希算法的进步之处在于，把数据和节点的关联，从“静态”变成了“动态”。
2，顺时针就近查找节点
一致性哈希算法是怎么把数据和节点关联起来的呢？在节点和数据都哈希到圆环上以后，数据通过顺时针方向查找的
方式，与节点建立关联。数据把顺时针找到的第一个节点作为自己的存储位置，这样一来，数据和节点就完美的关联起来
了。
三，一致性哈希算法面对的问题
一致性哈希算法解决了普通哈希算法不能解决的问题，但是一致性哈希算法也存在一定的缺陷。在节点A挂掉的情况下，
映射到节点A上的数据，会受到影响。因为之前映射到A节点的数据，现在按照顺时针查找，映射到了节点A的下一个节点。
同样的，在增加一个节点时，也会影响一部分数据。
一致性哈希算法的另一个缺陷是，当集群中的节点数量很少时，会造成数据倾斜。数据倾斜的问题，可以通过虚拟节点
的方式来解决。在虚拟节点和实际节点之间再增加一次映射。
总之，相比于普通的哈希算法，一致性哈希算法对于节点的动态增删，具有一定的容错性和可扩展性。
/**
* MurMurHash算法，是非加密HASH算法，性能很高，
* 比传统的CRC32,MD5，SHA-1（这两个算法都是加密HASH算法，复杂度本身就很高，带来的性能上的损害也不可避免）
* 等HASH算法要快很多，而且据说这个算法的碰撞率很低.
*
*/
private 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实现
package redis.cn;
import java.nio.charset.Charset;
import java.util.List;
import java.util.SortedMap;
import java.util.TreeMap;
import com.google.common.hash.HashFunction;
import com.google.common.hash.Hashing;
public class ConsistentHash {
// ------------------ 一致性哈希算法的java实现 ------------------
private SortedMap<Long,String> ketamaNodes = new TreeMap<Long,String>();
private int numberOfReplicas = 1024;
// 这里使用了谷歌的jar包 -- guava-18.0.jar
private HashFunction hashFunction = Hashing.md5();
private List<String> nodes;
private volatile boolean init = false; //标志是否初始化完成
// 有参数构造函数
public ConsistentHash(int numberOfReplicas,List<String> nodes){
this.numberOfReplicas = numberOfReplicas;
this.nodes = nodes;
init();
}
// 根据key的哈希值，找到最近的一个节点（服务器）
public String getNodeByKey(String key){
if(!init)
throw new RuntimeException("init uncomplete...");
// 注意，这里是NIO包 java.nio.charset.Charset
byte[] digest = hashFunction.hashString(key, Charset.forName("UTF-8")).asBytes();
long hash = hash(digest,0);
//如果找到这个节点，直接取节点，返回
if(!ketamaNodes.containsKey(hash)){
//得到大于当前key的那个子Map，然后从中取出第一个key，就是大于且离它最近的那个key
SortedMap<Long,String> tailMap = ketamaNodes.tailMap(hash);
if(tailMap.isEmpty()){
hash = ketamaNodes.firstKey();
}else{
hash = tailMap.firstKey();
}
}
return ketamaNodes.get(hash);
}
// 新增节点
public synchronized void addNode(String node){
init = false;
nodes.add(node);
init();
}
private void init(){
//对所有节点，生成numberOfReplicas个虚拟节点
for(String node:nodes){
//每四个虚拟节点为1组
for(int i=0;i<numberOfReplicas/4;i++){
//为这组虚拟结点得到惟一名称
byte[] digest = hashFunction.hashString(node+i, Charset.forName("UTF-8")).asBytes();
//Md5是一个16字节长度的数组，将16字节的数组每四个字节一组，分别对应一个虚拟结点，这就是为什么上面把虚拟结点四个划分一组的原因
for(int h=0;h<4;h++){
Long k = hash(digest,h);
ketamaNodes.put(k,node);
}
}
}
init = true;
}
public void printNodes(){
for(Long key:ketamaNodes.keySet()){
System.out.println(ketamaNodes.get(key));
}
}
// 哈希算法
public static 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)
| ((long)digest[0 + nTime * 4] & 0xFF);
return rv;
}
}
五，一致性哈希算法在redis中的应用
Redis本身不支持集群，所以需要借助API或者其他第三方产品，来实现集群部署。当然，也可以借助一致性哈希算法来
实现Redis集群。Memcached对大家应该不陌生，通过把Key映射到Memcached Server上，实现快速读取。我们可以动态对其节点
增加，并未影响之前已经映射到内存的Key与memcached Server之间的关系，这就是因为使用了一致性哈希算法。Memcached的
哈希策略是在客户端实现的，因此不同的客户端实现有区别，以Spymemcache、Xmemcache为例，都是使用了KETAMA作为其实现。
实现redis分布式集群，可以参考下面几种思路:
* 使用jedis
* 自己实现一致性哈希算法；
1，jedis
jedis是redis客户端API。Redis-server端并没有sharding方法，但是我们可以使用jedis来实现分布式。jedis使用了一种
叫做sharding的思想。
什么是sharding呢？简单的来说，就是数据库“分片”。sharding的核心理念就是将数据按照一定的策略"分散"存储在集群
中不同的物理机器上，从根本上来讲，实现了"大数据"分布式存储，体现了"集群"的概念。比如1亿条数据，我们可以根据数据
的hashcode，把数据散列存储在5个物理机器上。
sharding的实现，也是基于一致性哈希算法。我们先来看一下sharding实现的关键源代码。
1.1 hashcode取值：源码来自redis.clients.util.Hashing。Jedis中默认的hash算法是MD5，即我们熟悉的第五代信息摘要
算法：Message Digest Algorithm 5 。
//少量优化性能
public ThreadLocal<MessageDigest> md5Holder = new ThreadLocal<MessageDigest>();
public static final Hashing MD5 = new Hashing() {
public long hash(String key) {
return hash(SafeEncoder.encode(key));
}
// sharding使用的哈希算法是MD5
public long hash(byte[] key) {
try {
if (md5Holder.get() == null) {
md5Holder.set(MessageDigest.getInstance("MD5"));
}
}
catch (NoSuchAlgorithmException e) {
throw new IllegalStateException("++++ no md5 algorythm found");
}
MessageDigest md5 = md5Holder.get();
md5.reset();
md5.update(key);
//获得MD5字节序列
byte[] bKey = md5.digest();
//前四个字节作为计算参数，最终获得一个32位int值.
//此种计算方式，能够确保key的hash值更加“随机”/“离散”
//如果hash值过于密集，不利于一致性hash的实现(特别是有“虚拟节点”设计时)
long res = ((long) (bKey[3] & 0xFF) << 24)
| ((long) (bKey[2] & 0xFF) << 16)
| ((long) (bKey[1] & 0xFF) << 8)
| (long) (bKey[0] & 0xFF);
return res;
}
};
1.2 node构建过程(redis.clients.util.Sharded)：
//shards列表为客户端提供了所有redis-server配置信息，包括：ip，port，weight，name
//其中weight为权重，将直接决定“虚拟节点”的“比例”(密度)，权重越高，在存储是被hash命中的概率越高
//--其上存储的数据越多。
//其中name为“节点名称”，jedis使用name作为“节点hash值”的一个计算参数。
//---
//一致性hash算法，要求每个“虚拟节点”必须具备“hash值”，每个实际的server可以有多个“虚拟节点”(API级别)
//其中虚拟节点的个数= “逻辑区间长度” * weight，每个server的“虚拟节点”将会以“hash”的方式分布在全局区域中
//全局区域总长为2^32.每个“虚拟节点”以hash值的方式映射在全局区域中。
// 环形：0-->vnode1(:1230)-->vnode2(:2800)-->vnode3(400000)---2^32-->0
//所有的“虚拟节点”将按照其”节点hash“顺序排列(正序/反序均可)，因此相邻两个“虚拟节点”之间必有hash值差，
//那么此差值，即为前一个(或者后一个，根据实现而定)“虚拟节点”所负载的数据hash值区间。
//比如hash值为“2000”的数据将会被vnode1所接受。
private void initialize(List<S> shards){
//虚拟节点，采取TreeMap存储:排序，二叉树
nodes = new TreeMap<Long, S>();
for (int i = 0; i != shards.size(); ++i) {
final S shardInfo = shards.get(i);
if (shardInfo.getName() == null)
//当没有设置“name”是，将“SHARD-NODE”作为“虚拟节点”hash值计算的参数
//"逻辑区间步长"为160,为什么呢??
//最终多个server的“虚拟节点”将会交错布局，不一定非常均匀。
for (int n = 0; n < 160 * shardInfo.getWeight(); n++) {
nodes.put(this.algo.hash("SHARD-" + i + "-NODE-" + n), shardInfo);
}
else
for (int n = 0; n < 160 * shardInfo.getWeight(); n++) {
nodes.put(this.algo.hash(shardInfo.getName() + "*" + shardInfo.getWeight() + n), shardInfo);
}
resources.put(shardInfo, shardInfo.createResource());
}
}
1.3，node选择方式：
public R getShard(String key) {
return resources.get(getShardInfo(key));
}
public S getShardInfo(byte[] key) {
//获取>=key的“虚拟节点”的列表
SortedMap<Long, S> tail = nodes.tailMap(algo.hash(key));
//如果不存在“虚拟节点”，则将返回首节点。
if (tail.size() == 0) {
return nodes.get(nodes.firstKey());
}
//如果存在，则返回符合(>=key)条件的“虚拟节点”的第一个节点
return tail.get(tail.firstKey());
}
Jedis sharding模式下，如果某个server失效，客户端并不会删除此sharding，所以如果访问此sharding将会抛出异常。
这是为了保持所有的客户端数据视图一致性。你可能希望动态的一致性hash拓扑结构（即如果某个shard失效，sharding结构
则重新调整，失效的sharding上的数据则被hash到其他sharding上），但是很遗憾，SharedJedis客户端无法支持，如果非要
支持，则需要巨大的代码调整，而且还需要引入额外的拓扑自动发现机制。（参看：redis cluster架构，已提供此问题的完
善解决方案）。不过，在持久存储的情况下,我们可以使用"强hash"分片,则需要重写其Hash算法。强hash算法下,如果某个虚
拟节点所在的物理server故障，将导致数据无法访问(读取/存储)，即不会从虚拟节点列表中删除那些失效的server。
对于jedis如果重写了一致性哈希算法，你需要考虑以下几个方面：
1) 虚拟节点hash是否相对均匀
2) 数据的hash值分布是否均匀
3) 虚拟节点在“全局”是否散列均匀。
如果设计不良，很有可能导致数据在server上分布不均，而失去了sharding的本身意义。
2，java中使用jedis的demo
package redis.cn;
import java.util.ArrayList;
import java.util.List;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import redis.clients.jedis.JedisShardInfo;
import redis.clients.jedis.ShardedJedis;
import redis.clients.jedis.ShardedJedisPool;
/**
* @author yangcq
* @category jedis也是一致性哈希算法的一种实现。搭建redis分布式集群，可以使用jedis。
*/
public class ShardedRedis {
// 除了jdk自带的工具包以后，还需要导入下面2个jar包
// commons-pool2-2.0.jar
// jedis-2.4.2.jar
public static void main(String[] args){
// jedis配置参数
GenericObjectPoolConfig genericObjectPoolConfig = new GenericObjectPoolConfig();
genericObjectPoolConfig.setMaxTotal(1000);
genericObjectPoolConfig.setMaxIdle(500);
List<JedisShardInfo> jedisShardInfoList = new ArrayList<JedisShardInfo>();
JedisShardInfo jedisShardInfo1 = new JedisShardInfo("127.0.0.1",1234);
JedisShardInfo jedisShardInfo2 = new JedisShardInfo("127.0.0.1",1235);
JedisShardInfo jedisShardInfo3 = new JedisShardInfo("127.0.0.1",1236);
jedisShardInfoList.add(jedisShardInfo1);
jedisShardInfoList.add(jedisShardInfo2);
jedisShardInfoList.add(jedisShardInfo3);
ShardedJedisPool shardedJedisPool = new ShardedJedisPool(genericObjectPoolConfig,jedisShardInfoList);
set("key1","value1",shardedJedisPool);
set("key2","value2",shardedJedisPool);
set("key3","value3",shardedJedisPool);
set("key4","value4",shardedJedisPool);
set("key5","value5",shardedJedisPool);
// jedis隐藏了实现一致性哈希算法的细节，只是给我们提供了简单的接口调用，就可以实现redis分布式集群的搭建
// 那么jedis到底是如何实现一致性哈希算法的呢？
}
public static void set(String key,String value,ShardedJedisPool pool){
// 从共享资源池中获取redis实例
ShardedJedis shardedJedis = pool.getResource();
// 赋值
shardedJedis.set(key,value);
pool.returnResource(shardedJedis);
}
}
------------------------------------------------------
参考源码：
Jedis是通过ShardedJedis向redis集群写入的数据，ShardedJedis中的关键方法：
public Sharded(List<S> shards, Hashing algo, Pattern tagPattern) {
this.algo = algo;
this.tagPattern = tagPattern;
initialize(shards);
}
//初始化哈希环
private void initialize(List<S> shards) {
nodes = new TreeMap<Long, S>();
for (int i = 0; i != shards.size(); ++i) {
final S shardInfo = shards.get(i);
if (shardInfo.getName() == null)
for (int n = 0; n < 160 * shardInfo.getWeight(); n++) {
nodes.put(this.algo.hash("SHARD-" + i + "-NODE-" + n),
shardInfo);
}
else
for (int n = 0; n < 160 * shardInfo.getWeight(); n++) {
nodes.put(
this.algo.hash(shardInfo.getName() + "*"
+ shardInfo.getWeight() + n), shardInfo);
}
resources.put(shardInfo, shardInfo.createResource());
}
}
//将key，value存储到相应的shard
public String set(String key, String value) {
Jedis j = getShard(key);
return j.set(key, value);
}
public R getShard(String key) {
return resources.get(getShardInfo(key));
}
//根据key获取shard
public S getShardInfo(byte[] key) {
SortedMap<Long, S> tail = nodes.tailMap(algo.hash(key));
if (tail.isEmpty()) {
return nodes.get(nodes.firstKey());
}
return tail.get(tail.firstKey());
}
