分布式缓存--一致性hash原理和hash槽，以及算法实现
分布式缓存--一致性hash原理和hash槽，以及算法实现
背景
我们在使用n台存储设备存储数据的时候，常规做法有将数据根据key%n这样计算放在哪台服务器，但是在扩容的时候就会遇到数据迁移的问题，比如扩容m台服务器，以前是key%n，现在是key%(n+m)，导致数据存储的位置需要变化，数据迁移的成本比较大，这个时候我们就引用了一种叫一致性hash的算法 。 一致性哈希算法在1997年由麻省理工学院提出，设计目标是为了解决因特网中的热点(Hot spot)问题，初衷和CARP十分类似。一致性哈希修正了CARP使用的简单哈希算法带来的问题，使得DHT可以在P2P环境中真正得到应用。
hash槽
redis cluster里面使用的方法，一个 Redis Cluster包含16384（0~16383）个哈希槽，存储在Redis Cluster中的所有键都会被映射到这些slot中 集群中的每个键都属于这16384个哈希槽中的一个，集群使用公式slot=CRC16（key）/16384来计算key属于哪个槽，其中CRC16(key)语句用于计算key的CRC16校验和。
一致性hash的原理
一致性Hash算法也使用取模的方法，刚才描述的取模法是对服务器的数量进行取模，而一致性Hash算法是对2^32取模 首先，我们把2^32 想象成一个圆，圆上一共有2^32 个点，编号0-2^32-1，这个圆称为hash环，如下：
将服务器的ip或者编号进行hash算法计算取模hash(sever)%2^32 ，计算出服务器处于环上的位置，如下：
在存取数据的时候，根据hash算法计算数据在环中的位置hash(data)%2^32 ，再计算离数据最近顺时针方向最近的节点getNodeIndex(data) ，查找到数据存放的节点，如下：
遇到故障的时候
假如有服务节点遇到故障，则只需要将该服务节点的数据移入路径上的下一个节点，对原有的节点没有影响 如下图，假设C遇到故障，则C的数据重定到D即可：
扩容的时候
假如服务需要扩容的时候，则根据扩容的节点的位置，只需要将该位置路径下一个节点的部分数据移入新节点即可 如下图，新增一个node x，则只需要将C节点中，从b到x段的数据移入x即可，极大的减轻了扩容时对整个系统的影响
虚拟节点
在服务节点较少的时候，容易出现hash环倾斜的情况，即大量数据分布在少部分节点上，如图：
为了解决这种情况，引入了虚拟节点，一个实际节点对于多个虚拟节点，hash环上的节点越多，数据被均匀分布的概率就越大，如图：
算法实现
手写了一版简单的算法实现仅供理解
缓存节点CacheNode
@Data
public class CacheNode {
private String cacheName;
private String cacheIP;
private Long hashValue;
}一致性hash算法
@Data
@Slf4j
public class CacheManage {
private List<CacheNode> cacheNodeList;
private long MAX_CIRCLE = (1L << 32) - 1;
private Long hash(String nodeName) {
return MAX_CIRCLE & nodeName.hashCode();
}
/**
* 获取节点所在的下标
* @param hash
* @return
*/
private int getNodeIndex(Long hash) {
int index = cacheNodeList.size();
for (int i = 0; i < cacheNodeList.size(); i++) {
if (hash <= hash(cacheNodeList.get(i).getCacheName())) {
index = i;
break;
}
}
return index;
}
private void printList() {
for (CacheNode cacheNode : cacheNodeList) {
log.info("cachenode: {}", cacheNode);
}
}
/**
* 初始化带有虚拟节点的节点链表
* @param size
* @param virtualSize
*/
public void initVitualNode(int size, int virtualSize) {
cacheNodeList = new ArrayList<>();
log.info("mx: {}", MAX_CIRCLE);
for (int i = 0; i < size; i++) {
for (int j = 0; j < virtualSize; j++) {
CacheNode cacheNode = new CacheNode();
cacheNode.setCacheName("" + i + "_" + j + i + j + "_node_" + i + "_" + j);
cacheNode.setCacheIP("192.168.1.10" + i);
Long hashValue = hash(cacheNode.getCacheName());
cacheNode.setHashValue(hashValue);
int index = getNodeIndex(hashValue);
if (index == cacheNodeList.size()) {
cacheNodeList.add(cacheNode);
} else {
cacheNodeList.add(index, cacheNode);
}
}
}
printList();
}
/**
* 初始化节点列表
* @param size
*/
public void initCacheNode(int size) {
cacheNodeList = new ArrayList<>();
for (int i = 0; i < size; i++) {
CacheNode cacheNode = new CacheNode();
cacheNode.setCacheName("node_" + i);
cacheNode.setCacheIP("192.168.1.10" + i);
Long hashValue = hash(cacheNode.getCacheName());
cacheNode.setHashValue(hashValue);
int index = getNodeIndex(hashValue);
if (index == 0) {
cacheNodeList.add(cacheNode);
} else {
cacheNodeList.add(index, cacheNode);
}
printList();
}
printList();
}
/**
* 存数据
* @param data
*/
public void putData(String data) {
Long hashValue = hash(data);
int index = getNodeIndex(hashValue);
if(index == cacheNodeList.size()){
index = 0;
}
log.info("data:{}[{}] put into ====>{}", data, hashValue, cacheNodeList.get(index));
}
}