2.3.3-4. Memcached 分布式算法

1. 一致性 hash 算法应用领域

• 分布式数据存储场景：缓存、ES、Hadoop、分布式数据库

  2. 一致性 hash 算法引出

单节点服务器，以缓存为例

• 使用缓存的目的：提升数据访问性能，缓解数据库压力！

• 常用缓存中间件

  • Memcached
  • Redis

    高并发问题

    互联网公司的分布式高并发系统有什么特点？

• 高并发

• 海量数据
• 高并发问题如何处理？（应用集群）

单机缓存能扛起高并发吗？

• 不能

  Redis、Memcached 的并发能力有多强？

• 很强：10w 并发

• 如果并发量达 30w 怎么办？（主从集群）

海量数据对缓存有什么影响？（分片集群）

• 缓存的数据量很大，会超出单机内存容量。

• 数据如何均衡分布到缓存集群的节点上？

  3. 均衡分布方式

方式一：hash(key) % 集群节点数

• 示例：3 个节点的集群，数据 zp:888
  • 假设：hash(zp) = 200
  • 则数据放到服务器 2 上 200 % 3 = 2
• 搞促销活动，临时增加一台服务器来提高集群性能，方便扩展集群吗？
  • hash(zp) = 200；200 % 4 = 0
• 增加一个节点后，有多大比例的数据缓存命不中？
  • 3 -> 4，3/4
  • 99 -> 100，99%
• 对系统会有什么影响？（缓存雪崩）
  • 大量缓存命不中，就会访问数据库。
  • 瞬间失去了缓存的分压，数据库不堪重负，崩溃。

• 对程序员的影响

  • 加班：凌晨3/4点，扩容，预热数据。

    方式二：一致性 hash 算法

    

• hash 值一个非负整数，把非负整数的值范围做成一个圆环。（0 - int.max）

• 对集群的节点的某个属性求 hash 值（如节点名称），根据 hash 值把节点放到环上。
• 对数据的 key 求 hash，一样的把数据也放到环上，按顺时针方向，找离它最近的节点，就存储到这个节点上。
• 新增节点能均衡缓解原有节点的压力吗？
  • 不能。
  • 假如节点 3 靠近节点 1，影响数据少，但是会导致节点 3 上数据少。
  • 假如节点 3 靠近节点 2，影响数据多，并且导致节点 2 上数据很少。
• 集群的节点一定会均衡分布在环上吗？

• 不一定，hash 是散列值。
• 节点 0、2、3 处于“打酱油”状态。

  方式三：一致性 Hash 算法 + 虚拟节点

• 一致性 Hash 算法，只是解决缩容、扩容问题（程序员加班问题）。
• 引入虚拟节点解决数据分布不均衡问题。

4. 一致性 Hash 算法需要考虑的问题

Hash 算法选择

• hashCode()，不够散列，会有负值（取绝对值即可）

• 其他 hash 算法：比如 CRC32_HASH、FNV1_32_HASH、KETAMA_HASH 等，其中 KETAMA_HASH 是默认的 memcached 推荐的一致性 Hash 算法。

  虚拟节点放到环上，具体是要做什么？

• 根据 Hash 值排序存储。

• 排序存储要被快速查找。
• 这个排序存储还要能方便变更。
• 考虑：
  • Array：排序可以做到，但是变更会有问题，并且快速查找也会有问题（根据值查找）。
  • 链表：变更和查找不方便。
  • 二叉树：变更和查找都很快。会有遍历层级过高，特殊情况会退化成链表的问题。
  • 红黑树：解决层级过高的问题。
  • TreeMap

    5. 代码实现一致性 Hash 算法

FNV1_32_HASH

public class FNV1_32_HASH {
    /** 尽量让 hash 值散列开 */
    public static int getHash(String str) {
        final int p = 16777619;
        int hash = (int) 2166136261L;
        for (int i = 0; i < str.length(); i ++) {
            hash = (hash ^ str.charAt(i)) * p;
        }
        hash += hash << 13;
        hash ^= hash >> 7;
        hash += hash << 3;
        hash ^= hash >> 17;
        hash += hash << 5;
        // 如果算出来的值为负数则取其绝对值
        if (hash < 0) {
            hash = Math.abs(hash);
        }
        return hash;
    }
}

测试 String.hashCode() 和 以上算法的均衡度

public class HashTest {
    public static void main(String[] args) {
        System.out.println("hashCode(): ");
        System.out.println("192.168.0.0:1111 的哈希值：" + "192.168.0.0:1111".hashCode());
        System.out.println("192.168.0.1:1111 的哈希值：" + "192.168.0.1:1111".hashCode());
        System.out.println("192.168.0.2:1111 的哈希值：" + "192.168.0.2:1111".hashCode());
        System.out.println("192.168.0.3:1111 的哈希值：" + "192.168.0.3:1111".hashCode());
        System.out.println("192.168.0.4:1111 的哈希值：" + "192.168.0.4:1111".hashCode());
        System.out.println("192.168.1.0:1111 的哈希值：" + "192.168.1.0:1111".hashCode());
        System.out.println();
        System.out.println("FNV1_32_HASH: ");
        System.out.println("192.168.0.0:1111 的哈希值：" + FNV1_32_HASH.getHash("192.168.0.0:1111"));
        System.out.println("192.168.0.1:1111 的哈希值：" + FNV1_32_HASH.getHash("192.168.0.1:1111"));
        System.out.println("192.168.0.2:1111 的哈希值：" + FNV1_32_HASH.getHash("192.168.0.2:1111"));
        System.out.println("192.168.0.3:1111 的哈希值：" + FNV1_32_HASH.getHash("192.168.0.3:1111"));
        System.out.println("192.168.0.4:1111 的哈希值：" + FNV1_32_HASH.getHash("192.168.0.4:1111"));
        System.out.println("192.168.1.0:1111 的哈希值：" + FNV1_32_HASH.getHash("192.168.1.0:1111"));
    }
}

简单实现一个一致性算法

public class ConsistenceHash {
    /** 物理节点集合 */
    private List<String> realNodes = new ArrayList<>();
    /** 虚拟节点数，用户进行指定 */
    private int virtualNums = 100;
    /**
     * 物理节点和虚拟节点的对应关系
     *      key：物理节点
     *      value：虚拟 hash 值
     */
    private Map<String, List<Integer>> real2VirtualMap = new HashMap<>();
    /**
     * 排序存储结构，红黑树
     *      key：虚拟节点 hash 值
     *      value：物理节点
     */
    private SortedMap<Integer, String> sortedMap = new TreeMap<>();
    public ConsistenceHash() { }
    public ConsistenceHash(int virtualNums) {
        this.virtualNums = virtualNums;
    }
    /**
     * 添加服务节点
     */
    public void addServer(String node) {
        realNodes.add(node);
        String vnode = null;
        // i：v-001，count：产生了虚拟节点的次数
        int i = 0, count = 0;
        List<Integer> virtualNodes = new ArrayList<>();
        real2VirtualMap.put(node, virtualNodes);
        // 创建虚拟节点，并且放到圆环上去（排序存储）
        while (count < this.virtualNums) {
            i++;
            vnode = node + "V-" + i;
            int hashValue = FNV1_32_HASH.getHash(vnode);
            // 解决 hash 碰撞
            if (! sortedMap.containsKey(hashValue)) {
                virtualNodes.add(hashValue);
                sortedMap.put(hashValue, node);
                count ++;
            }
        }
    }
    /**
     * 移除服务器
     */
    public void removeServer(String node) {
        List<Integer> virtualNodes = real2VirtualMap.get(node);
        if (virtualNodes != null && virtualNodes.size() > 0) {
            for (Integer hashVal : virtualNodes) {
                sortedMap.remove(hashVal);
            }
        }
        realNodes.remove(node);
        real2VirtualMap.remove(node);
    }
    /**
     * 根据 key 查找物理节点
     */
    public String getServer(String key) {
        // 根据 key 产生 hash 值
        int hashValue = FNV1_32_HASH.getHash(key);
        SortedMap<Integer, String> subMap = sortedMap.tailMap(hashValue);
        if (subMap == null || subMap.isEmpty()) {
            return sortedMap.get(sortedMap.firstKey());
        }
        return sortedMap.get(subMap.firstKey());
    }
    public static void main(String[] args) {
        ConsistenceHash ch = new ConsistenceHash();
        ch.addServer("192.168.1.10");
        ch.addServer("192.168.1.11");
        ch.addServer("192.168.1.12");
        for (int i = 0; i < 10; i ++) {
            System.out.println("a" + i + " 对应的服务器是：" + ch.getServer("a" + i));
        }
    }
}