一致性hash原理及实现（python版）

使用背景

用户的每一次动态数据的请求，都涉及数据库的访问。而一个系统中，数据库往往是最脆弱的缓解，为了缓解数据库的读压力，通常会将热点数据放入缓存。当用户进行数据请求时，先访问缓存，若有数据直接返回，没有，才进行数据库的查询，并将访问结果写回缓存。从而，挡住读访问的大部分流量。
缓存存储的数据量很大，所以，一般需要搭建缓存服务器集群，这样就需要把不同的数据放入不同的缓存服务器，当用户请求到来时，根据用户的请求的内容，从不同的缓存服务器获取数据。
从缓存中获取数据就涉及到缓存的路由选择。

一致性hash原理

最简单的路由方法就是对key取hash，并取余，通过余数，确定分发到哪一个缓存服务器。但是，这样，当增加或缓存服务时，会有大量的key，变换访问是缓存服务器。从而，导致缓存击穿、雪崩等一系列问题。
为了解决这一情况，提出了一致性hash。

一致性hash原理

1. 将缓存服务器的哈希值映射到0～2^32的圆上。
2. 当请求到来时，将请求也映射到0～2^32的圆。并开始顺时针查找，从找到的第一个缓存服务器上，获取数据。

基于虚拟节点的一致性hash算法

一致性hash解决了增加/下线 缓存服务器，可能引起的缓存击穿等问题。但是当缓存服务器数量较少时，出现数据倾斜等问题。
为了解决这个问题，产生了基于虚拟节点的一致性hash算法。即，一个缓存服务器映射到hash环上的多个地址。从而，减少数据倾斜的情况。

代码实现

一致性hash实现

#/bin/python 
# -*- coding: utf-8 -*-
import bisect
import hashlib

def get_hash(raw_str):
    """将字符串映射到2^32的数字中"""
    md5_str = hashlib.md5(raw_str).hexdigest()
    return long(md5_str, 16)


class CacheNode(object):
    """缓存节点类，负责记录缓存服务器信息，以及发送缓存服务器方法"""
    def __init__(self, ip):
        self.ip = ip
    
    def send(self, request):
        """发送到对应的cache
        Args:
            request：需要转发的request信息
        """
        # 假装有内容的亚子
        pass


class HashSeverMgr(object):
    """server管理类，给定ip，返回对应的server"""
    def __init__(self):
        self.cache_list = []
        self.cache_node = dict()

    def add_server(self, node):
        """添加缓存节点
        Args:
            node: 缓存节点类CacheNode，记录缓存server的香港信息。
        """
        node_hash = get_hash(node.ip)
        bisect.insort(self.cache_list, node_hash)
        self.cache_node[node_hash] = node
    
    def del_server(self, node):
        """删除缓存节点"""
        node_hash = get_hash(node.ip)
        self.cache_list.remove(node_hash)
        del self.cache_node[node_hash]

    def get_server(self, source_key):
        """获取目标缓存节点，确定待转发的目标缓存server"""
        key_hash = get_hash(source_key)
        index = bisect.bisect_left(self.cache_list, key_hash)
        index = index % len(self.cache_list)  # 若比最大的node hash还大，分发给第一个node
        return self.cache_node[self.cache_list[index]]
    

if __name__ == "__main__":
    cache_ips = [
        "1.2.3.4",
        "2.3.4.5",
        "3.4.5.6"
    ]
    source_key = "1234567890"
    hash_mgr = HashSeverMgr()
    for ip in cache_ips:
        cache_node_temp = CacheNode(ip)
        hash_mgr.add_server(cache_node_temp)
    
    sended_node = hash_mgr.get_server(source_key)
    print sended_node.ip

基于虚拟节点的一致性hash实现

只需修改HashSeverMgr的add_server和del_server，每添加一个节点时，添加多个虚拟hash节点即可。get_server函数不用修改。

class HashSeverMgr(object):
    """server管理类，给定ip，返回对应的server"""
    def __init__(self):
        self.cache_list = []
        self.cache_node = dict()
        self.virtual_num = 10

    def add_server(self, node):
        """添加缓存节点
        Args:
            node: 缓存节点类CacheNode，记录缓存server的香港信息。
        """
        for index in xrange(0, self.virtual_num):
            node_hash = get_hash("%s_%s" % (node.ip, index))
            bisect.insort(self.cache_list, node_hash)
            self.cache_node[node_hash] = node
    
    def del_server(self, node):
        """删除缓存节点"""
        for index in xrange(0, self.virtual_num):
            node_hash = get_hash("%s_%s" % (node.ip, index))
            self.cache_list.remove(node_hash)
            del self.cache_node[node_hash]

效果比较

为了验证基于虚拟节点的一致性hash，可以将数据更平均的分布在各个服务器上。 测试随机100KV在10节点服务器上分布的标准差。基于虚拟节点的一致性hash，将每个节点虚拟10个节点服务器。

#/bin/python 
# -*- coding: utf-8 -*-
import random
import consistent_hashing_simply
import consistent_hashing_virtual


IP_NUM = 1000000

cache_ips = [   # 随机定义10节点服务器ip
    "1.2.3.4",
    "2.3.4.5",
    "3.4.5.6",
    "4.5.6.7",
    "5.6.7.8",
    "6.7.8.9",
    "7.8.9.10",
    "8.9.10.11",
    "9.10.11.12",
    "10.11.12.13"
]


def get_source_key():
    """随机返回source_key"""
    return str(random.randint(0, 99999999999999))


def get_standard_deviation(cache_info):
    """获取标准差"""
    # 获取期望
    sum_num = 0
    for _, num in cache_info.items():
        sum_num += num
    except_num = float(sum_num) / len(cache_info)

    # 获取标准差
    temp = 0
    for _, num in cache_info.items():
        temp += (num - except_num) * (num - except_num)
    standard_deviation = (float(temp) / len(cache_info)) ** 0.5
    return standard_deviation


if __name__ == "__main__":
    simply_distribution = dict()
    virtual_distribution = dict()
    for ip in cache_ips:
        simply_distribution[ip] = 0
        virtual_distribution[ip] = 0
    
    simply_hash_mgr = consistent_hashing_simply.HashSeverMgr()
    virtual_hash_mgr = consistent_hashing_virtual.HashSeverMgr()
    for ip in cache_ips:
        cache_node_temp = consistent_hashing_simply.CacheNode(ip)
        simply_hash_mgr.add_server(cache_node_temp)
        virtual_hash_mgr.add_server(cache_node_temp)

    for _ in xrange(0, IP_NUM):
        source_key = get_source_key()
        simply_node = simply_hash_mgr.get_server(source_key)
        virtual_node = virtual_hash_mgr.get_server(source_key)

        simply_distribution[simply_node.ip] += 1
        virtual_distribution[virtual_node.ip] += 1

    simply_standard_deviation = get_standard_deviation(simply_distribution)
    virtual_standard_deviation = get_standard_deviation(virtual_distribution)
    
    print "consistent_hashing_simply's standard deviation: ", str(simply_standard_deviation)
    print "consistent_hashing_virtual's standard deviation: ", str(virtual_standard_deviation)

某次运行结果：

consistent_hashing_simply's standard deviation: 86800.2052774
consistent_hashing_virtual's standard deviation: 33318.0594513

由于随机产生key，每次运行结果会有差异，但是，可以从结果看出，基于虚拟节点的一致性hash，可以使数据分布的更平均。