第五周作业

关注

发布于: 2020 年 11 月 22 日

1.用你熟悉的编程语言实现一致性 hash 算法。

简介

一致性 hash 被广泛用于负载均衡领域，我们常见的一些应用如 nginx 和 memcached 都采用了一致性 hash 来作为集群负载均衡方案。一致性 hash 通过构建环状的 hash 空间来解决集群数量变化而导致的 hash 映射失效的情况。下图中 NODE1、2、3 就是服务器映射到 hash 环上的节点。

上图无虚拟节点。

当消息被送到负载均衡器上后，计算出该消息的 hash 值后，它也会被映射到环上，然后沿着顺时针方向找到第一个节点，这个节点就是处理该消息的服务器。如图箭头所指的方向。

新增服务器，例如 NODE3 后，相当于在 hash 环空白位置加上一个节点，之后落在该节点逆时针附近的一些消息就会被分派到该服务器上处理。

当然，这个算法有些问题，node 节点的数量相对于环来说太稀松了，落点很难均匀分布，所以负载往往非常的不平衡。为了更加均匀的分配负载，人们想到了虚拟节点的方案：就是让一个服务器映射到多个 hash 节点上去，这样的话节点就没有那么稀松了，负载也就更加均衡了。

工具

hash 算法：FNV_32_hash 算法，它能将所有的字符串映射为一个特定的 32 位 int 值

private static int getHashCode(String key) {        // 结果太聚集，不使用        // int ret = key.hashCode();        final int p = 16777619;        int ret = (int) 2166136261L;        for (int i = 0; i < key.length(); i++) {            ret = (ret ^ key.charAt(i)) * p;        }        ret += ret << 13;        ret ^= ret >> 7;        ret += ret << 3;        ret ^= ret >> 17;        ret += ret << 5;        return ret < 0 ? -ret : ret;    }

复制代码

标准差：（每个样本-平均值）的平方/（服务器数-1），结果再取平方根

hash 环：容器，使用 TreeMap，天生自带 ceilingEntry 和 firstEntry

hash 环初始流程：

初始化 hash 环->服务器列表->计算服务器 hashcode->加入 hash 环->计算服务器虚拟节点 hashcode->加入 hash 环

获取服务器流程：

输入 key->计算 key 的 hashcode->查找大于等于 key 的 hashcode 的第一台服务器->找到服务器返回,没有找到返回第一台服务器

代码

哈希环类 ConsistentHashRing

package kd.autops.service;
import java.util.ArrayList;import java.util.List;import java.util.Map;import java.util.TreeMap;
/** * @author chenhao * @date 2020/11/22 */public class ConsistentHashRing {    private TreeMap<Integer, Cache> map = new TreeMap<>();    private List<Cache> servers = new ArrayList<>();
    /**     * 往环中加入服务器     *     * @param servers           要加入的服务器列表     * @param virtualNumPerNode 每台服务器增加多少个虚拟节点，0表示不增加虚拟节点     */    void addServers(String[] servers, int virtualNumPerNode) {        if (servers != null) {            for (String server : servers) {                Cache cacheServer = new CacheServerMock(server);                map.put(getHashCode(server), cacheServer);                // 加虚拟节点                for (int i = 1; i <= virtualNumPerNode; i++) {                    map.put(getVirtualHashCode(server, i), cacheServer);                }                this.servers.add(cacheServer);            }        }    }
    /**     * 根据指定的缓存key，查找环上的对应服务器返回     *     * @param cacheKey 缓存key     * @return 服务器     */    Cache getServer(String cacheKey) {        if (map.size() <= 0) {            throw new IllegalArgumentException("you must add server before call this method.");        }        if (cacheKey == null || cacheKey.isEmpty()) {            throw new IllegalArgumentException("cacheKey can't be empty");        }        int hash = getHashCode(cacheKey);        Map.Entry<Integer, Cache> server = map.ceilingEntry(hash);        if (server == null) {            return map.firstEntry().getValue();        }        return server.getValue();    }
    /**     * 返回第i个虚拟结点的hash值，注意不能用随机值，以免重启导致变化。     *     * @param key 服务器值     * @param i   第几个虚拟节点     * @return hash code     */    private int getVirtualHashCode(String key, int i) {        key = key + "&&Virtual" + i;        return getHashCode(key);    }
    private static int getHashCode(String key) {        // 结果太聚集，不使用        // int ret = key.hashCode();        final int p = 16777619;        int ret = (int) 2166136261L;        for (int i = 0; i < key.length(); i++) {            ret = (ret ^ key.charAt(i)) * p;        }        ret += ret << 13;        ret ^= ret >> 7;        ret += ret << 3;        ret ^= ret >> 17;        ret += ret << 5;        return ret < 0 ? -ret : ret;    }
    @Override    public String toString() {        int totalNum = 0;        for (Cache cache : servers) {            CacheServerMock mock = (CacheServerMock) cache;            totalNum += mock.getCaches().size();        }        int avg = totalNum / servers.size();        StringBuilder sb = new StringBuilder();        sb.append("总缓存个数：").append(totalNum)                .append(" 服务器台数：").append(servers.size())                .append(" 每服务器虚拟节点数：").append(map.size() / servers.size() - 1)                .append(" 平均值：").append(avg)                .append("\r\n");        // 标准差公式： (每个样本 - 平均值)的平方 / (服务器数 - 1)，结果再取平方根        long diff = 0;        for (Cache cache : servers) {            CacheServerMock mock = (CacheServerMock) cache;            int size = mock.getCaches().size();            // sb.append("Server-Name:")            //         .append(mock.getServerName())            //         .append(" Cache-num: ")            //         .append(size)            //         .append("\r\n");            long tmp = size - avg;            diff += (tmp * tmp);        }        diff /= (servers.size() - 1);        sb.append("标准差为： ").append(Math.sqrt(diff)).append("\r\n");        return sb.toString();    }}

复制代码

cache 接口

package kd.autops.service;
/** * 缓存接口 * * @author chenhao * @date 2020/11/22 */public interface Cache {    /**     * 根据key读取缓存数据     *     * @param key key     * @return 缓存     */    String get(String key);
    /**     * 设置缓存     *     * @param key 缓存key     * @param val 缓存值     */    void set(String key, String val);}

复制代码

CacheImplement

package kd.autops.service;
/** * @author chenhao * @date 2020/11/22 */public class CacheImplement implements Cache {    private ConsistentHashRing ring = new ConsistentHashRing();
    CacheImplement(String[] servers, int virtualNumPerNode) {        ring.addServers(servers, virtualNumPerNode);    }
    @Override    public String get(String key) {        Cache serverKey = ring.getServer(key);        return serverKey.get(key);    }
    @Override    public void set(String key, String val) {        Cache serverKey = ring.getServer(key);        serverKey.set(key, val);    }
    @Override    public String toString() {        return ring.toString();    }}

复制代码

CacheServerMock 类

package kd.autops.service;
import java.util.HashMap;import java.util.Map;
/** * @author chenhao * @date 2020/11/22 */public class CacheServerMock implements Cache {    private Map<String, String> caches = new HashMap<>();    private String serverName;
    CacheServerMock(String serverName) {        this.serverName = serverName;    }
    public String getServerName() {        return serverName;    }
    Map<String, String> getCaches() {        return caches;    }
    @Override    public String get(String key) {        return caches.get(key);    }
    @Override    public void set(String key, String val) {        caches.put(key, val);    }}

复制代码

Main 测试类

略微修改使用多线程来测试为什么 150 的虚拟节点的时候，负载均衡性比较好

package kd.autops.service;
import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;
/** * @author chenhao * @date 2020/11/22 */public class Main {    private static String[] servers = new String[]{            "10.0.0.1",            "10.0.0.2",            "10.0.0.3",            "10.0.0.4",            "10.0.0.5",            "10.0.0.6",            "10.0.0.7",            "10.0.0.8",            "10.0.0.9",            "10.0.0.10",    };    private static int maxCacheNum = 1000000;
    public static void main(String[] args) {        testCache();    }
    private static final int VIRTUAL_NODE_1000 = 1000;
    private static void testCache() {        ExecutorService executorService = Executors.newFixedThreadPool(20);        for (int i = 0; i < VIRTUAL_NODE_1000; i++) {            int finalI = i;            executorService.execute(() -> {                Cache cache = new CacheImplement(servers, finalI);                fillCache(cache, maxCacheNum);                System.out.println(cache);            });        }
    }
    private static void fillCache(Cache cache, int cacheNum) {        for (int i = 0; i < cacheNum; i++) {            String key = "key" + i;            String val = "val" + key;            cache.set(key, val);        }    }}

复制代码

输出：

从 0-1000 个虚拟节点从 150 开始，标准差在 5000 左右，但是越往后，标准差接近 3000+，所以，对于 150 的虚拟节点的结论，无法确认

总缓存个数：1000000 服务器台数：10 每服务器虚拟节点数：16 平均值：100000标准差为： 14512.571998098752
总缓存个数：1000000 服务器台数：10 每服务器虚拟节点数：14 平均值：100000标准差为： 16318.876493190332
总缓存个数：1000000 服务器台数：10 每服务器虚拟节点数：17 平均值：100000标准差为： 18141.534637400444
总缓存个数：1000000 服务器台数：10 每服务器虚拟节点数：15 平均值：100000标准差为： 15913.320269510068
总缓存个数：1000000 服务器台数：10 每服务器虚拟节点数：13 平均值：100000标准差为： 16177.60705419686
...