一致性哈希

一致性哈希算法

 

 

使用场景

现在我们假设有100台redis data服务器，一份数据101进来的时候，以散列公式hash(i)&100，计算所存放的服务器，假设hash(i) = i,那么数据被散列到标号为1的服务器,然后这个时候服务器新增了一台，然后散列公式为hash(i)%101，这个时候请求访问数据101的时候，被分配至0号服务器，但是其实这个时候数据是在1号服务器的。

所以这个时候大量的数据失效了（访问不到了）。

所以这个时候，我们假设是新增了服务器，如果是持久化存储的，我们可以让服务器集群对数据进行重新散列，进行数据迁移，然后进行恢复，但是这个时候就意味着每次增减服务器的时候，集群就需要大量的通信，进行数据迁移，这个开销是非常大的。如果只是缓存，那么缓存就都失效了。所以这个时候怎么办？

我们可以看到，关键问题在于，服务器数量变动的时候，要能够保证旧的数据能够按照老的算法，计算到数据所在的服务器，而新的数据能够按照新的散列算法，计算出数据所在的服务器。

如上图，我们有ABCD四台服务器，这四台服务器被分配至0~232 的一个环上，比如0~230的存储在A服务器，230 +1~231 存储到B服务器上.....CD按照这样的进行均分。将我们的散列空间也划为0~232 ，然后数据进来后对232 取模，得到一个值K1，我们根据K1在环上所处的位置，得到所分配到的服务器，如图，K1被分配到B服务器。 这个时候，我们有一台服务器B失效了。

我们可以看到，如果是B失效了，那么如果有持久化存储的，需要做数据恢复，将B的数据迁移至C即可，对于原本散列在A和D的数据，不需要做任何改变。 同理，如果我们是新增了服务器，那么只需要对一台服务器的数据迁移一部分至新加的服务器即可。

一致性hash算法，减少了数据映射关系的变动，不会像hash(i)%N那样带来全局的变动

而且这样还有个好处，假设我们使用UID作为散列范围（即上面的232 ）,那么假设有部分UID的访问很频繁，而且这部分UID集中在B服务器上，那么就造成了B的负载远远高于其他服务器。这就是热点数据的问题。这个时候我们可以向B所在的UID空间添加服务器，减少B的压力。

其实还有个更好的解决办法：虚拟节点。

上面说的情况是，使用真实的服务器作为节点散列在232 上。 我们假设，只有4台服务器（如上图），然后A上面有热点数据，结果A挂掉了，然后做数据恢复，A的数据迁移至B，然后B需要承受A+B的数据，也承受不住，也挂了。。。。然后继续CD都挂了。这就造成了

雪崩效应。

上面会造成雪崩效应的原因分析：
如果不存在热点数据的时候，每台机器的承受的压力是M/2(假设每台机器的最高负载能力为M)，原本是不会有问题的，但是，这个时候A服务器由于有热点数据挂了，然后A的数据迁移至B，导致B所需要承受的压力变为M（还不考虑热点数据访问的压力），所以这个失败B是必挂的，然后C至少需要承受1.5M的压力。。。。然后大家一起挂。。。
所以我们通过上面可以看到，之所以会大家一起挂，原因在于如果一台机器挂了，那么它的压力全部被分配到一台机器上，导致雪崩。

如果我们A挂了以后，数据被平均分配到BCD上，每台机器多承受M/6的压力，然后大家就都不会挂啦（不考虑热点数据）。

这里引入虚拟节点，如图：

环上的空间被划分为8份，然后A存储A1和A2。。。
这个时候，如果A服务器挂了，访问压力会分配至C2和D1，也就是C和D服务器，而不是像前面，全部被分配到B上。

引入虚拟节点，主要在于，如果一台服务器挂了，能够将压力引流至不同的服务器。

总结：一致性hash算法（DHT）通过减少影响范围的方式解决了增减服务器导致的数据散列问题，从而解决了分布式环境下负载均衡问题，如果存在热点数据，那么通过增添节点的方式，对热点区间进行划分，将压力分配至其他服务器。重新达到负载均衡的状态。

tair的负载均衡就是采用的一致性hash算法啦~~~
一致性hash算法在分布式环境中应用的很广，只要是涉及到分布式存储的负载均衡问题，一致性hash都是很好的解决的方案。

 

 

python中的hashring模块便实现了一致性哈希

hashring安装

下载hashring https://pypi.python.org/pypi/hash_ring#downloads　　

找到hash_ring.py文件，如果是python2.x环境直接拿来可以用

如果安装环境python3.x，需要在hash_ring.py文件里做如下修改

# -*- coding: utf-8 -*-
"""
    hash_ring
    ~~~~~~~~~~~~~~
    Implements consistent hashing that can be used when
    the number of server nodes can increase or decrease (like in memcached).

    Consistent hashing is a scheme that provides a hash table functionality
    in a way that the adding or removing of one slot
    does not significantly change the mapping of keys to slots.

    More information about consistent hashing can be read in these articles:

        "Web Caching with Consistent Hashing":
            http://www8.org/w8-papers/2a-webserver/caching/paper2.html

        "Consistent hashing and random trees:
        Distributed caching protocols for relieving hot spots on the World Wide Web (1997)":
            http://citeseerx.ist.psu.edu/legacymapper?did=38148


    Example of usage::

        memcache_servers = ['192.168.0.246:11212',
                            '192.168.0.247:11212',
                            '192.168.0.249:11212']

        ring = HashRing(memcache_servers)
        server = ring.get_node('my_key')

    :copyright: 2008 by Amir Salihefendic.
    :license: BSD
"""

import math
import sys
from bisect import bisect

if sys.version_info >= (2, 5):
    import hashlib
    md5_constructor = hashlib.md5
else:
    import md5
    md5_constructor = md5.new

class HashRing(object):

    def __init__(self, nodes=None, weights=None):
        """`nodes` is a list of objects that have a proper __str__ representation.
        `weights` is dictionary that sets weights to the nodes.  The default
        weight is that all nodes are equal.
        """
        self.ring = dict()
        self._sorted_keys = []

        self.nodes = nodes

        if not weights:
            weights = {}
        self.weights = weights

        self._generate_circle()

    def _generate_circle(self):
        """Generates the circle.
        """
        total_weight = 0
        for node in self.nodes:
            total_weight += self.weights.get(node, 1)

        for node in self.nodes:
            weight = 1

            if node in self.weights:
                weight = self.weights.get(node)

            factor = math.floor((40*len(self.nodes)*weight) / total_weight);

            for j in range(0, int(factor)):
                b_key = self._hash_digest( '%s-%s' % (node, j) )

                for i in range(0, 3):
                    key = self._hash_val(b_key, lambda x: x+i*4)
                    self.ring[key] = node
                    self._sorted_keys.append(key)

        self._sorted_keys.sort()

    def get_node(self, string_key):
        """Given a string key a corresponding node in the hash ring is returned.

        If the hash ring is empty, `None` is returned.
        """
        pos = self.get_node_pos(string_key)
        if pos is None:
            return None
        return self.ring[ self._sorted_keys[pos] ]

    def get_node_pos(self, string_key):
        """Given a string key a corresponding node in the hash ring is returned
        along with it's position in the ring.

        If the hash ring is empty, (`None`, `None`) is returned.
        """
        if not self.ring:
            return None

        key = self.gen_key(string_key)

        nodes = self._sorted_keys
        pos = bisect(nodes, key)

        if pos == len(nodes):
            return 0
        else:
            return pos

    def iterate_nodes(self, string_key, distinct=True):
        """Given a string key it returns the nodes as a generator that can hold the key.

        The generator iterates one time through the ring
        starting at the correct position.

        if `distinct` is set, then the nodes returned will be unique,
        i.e. no virtual copies will be returned.
        """
        if not self.ring:
            yield None, None

        returned_values = set()
        def distinct_filter(value):
            if str(value) not in returned_values:
                returned_values.add(str(value))
                return value

        pos = self.get_node_pos(string_key)
        for key in self._sorted_keys[pos:]:
            val = distinct_filter(self.ring[key])
            if val:
                yield val

        for i, key in enumerate(self._sorted_keys):
            if i < pos:
                val = distinct_filter(self.ring[key])
                if val:
                    yield val

    def gen_key(self, key):
        """Given a string key it returns a long value,
        this long value represents a place on the hash ring.

        md5 is currently used because it mixes well.
        """
        b_key = self._hash_digest(key)
        return self._hash_val(b_key, lambda x: x)

    def _hash_val(self, b_key, entry_fn):
        return (( b_key[entry_fn(3)] << 24)
                |(b_key[entry_fn(2)] << 16)
                |(b_key[entry_fn(1)] << 8)
                | b_key[entry_fn(0)] )

    def _hash_digest(self, key):
        m = md5_constructor()
        m.update(bytes(key,encoding='utf-8'))
        #m.digest  2.7中返回字符串，3.x返回字节
        # return map(ord, m.digest())
        return list(m.digest())



if __name__ == '__main__':
    memcache_servers = ['192.168.0.246:11212',
                        '192.168.0.247:11212',
                        '192.168.0.249:11212']

    ring = HashRing(memcache_servers)
    server = ring.get_node('my_key')
    print(server)