Consistent Hash Load Balance

背景介绍

最近和小伙伴在一起在做tim这个开源IM服务，考虑到历史消息，路由信息，会话列表这些使用内存缓存性能更高，
所以需要一个一致性hash负载均衡算法，保证服务上下线后缓存命中率更高. 这里不对一致性hash做详细介绍，网上有很多资料，可以自行google.

实现方式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62


public interface LoadBalancer {
    Server select(List<Server> servers, String hashKey);
}

@NoArgsConstructor
@AllArgsConstructor
@Data
public class Server {

    private String ip;

    private int port;

    public String getUrl() {
        return ip + ":" + port;
    }

}

public class ConsistentHashLoadBalancer implements LoadBalancer {

    // 使用murmur哈希算法
    private final static HashFunction hashFunction = Hashing.murmur3_32();

    private final static Charset charset = Charset.forName("utf-8");

    private final static int VIRTUAL_NODE_SIZE = 10;

    private final static String VIRTUAL_NODE_SUFFIX = "#@";

    @Override
    public Server select(List<Server> servers, String hashKey) {
        HashCode hashCode = hashFunction.hashString(hashKey, charset);
        TreeMap<Integer, Server> ring = buildConsistentHashRing(servers);
        Server server = locate(ring, hashCode.asInt());
        return server;
    }

    private Server locate(TreeMap<Integer, Server> ring, int invocationHashCode) {
        // 向右找到第一个 key
        Map.Entry<Integer, Server> locateEntry = ring.ceilingEntry(invocationHashCode);
        if (locateEntry == null) {
            // 超过尾部则取第一个 key
            locateEntry = ring.firstEntry();
        }
        return locateEntry.getValue();
    }

    private TreeMap<Integer, Server> buildConsistentHashRing(List<Server> servers) {
        TreeMap<Integer, Server> virtualNodeRing = new TreeMap<>();
        for (Server server : servers) {
            for (int i = 0; i < VIRTUAL_NODE_SIZE; i++) {
                // 映射虚拟节点与物理节点
                HashCode hashCode = hashFunction
                    .hashString(server.getUrl() + VIRTUAL_NODE_SUFFIX + i, charset);
                virtualNodeRing.put(hashCode.asInt(), server);
            }
        }
        return virtualNodeRing;
    }

测试用例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
public class LoadBalanceTest extends TimCommonTest {
    /**
     * 测试节点新增删除后的变化程度
     */
    @Test
    public void testNodeAddAndRemove() {
        // 原节点
        List<Server> servers = new ArrayList<>();
        for (int i = 0; i < 100; i++) {
            servers.add(new Server(getRandomIp(), 8080));
        }
        // 上线20%节点
        List<Server> serverAdd = new ArrayList<>(120);
        servers.forEach(x -> serverAdd.add(x));
        for (int i = 0; i < 20; i++) {
            serverAdd.add(new Server(getRandomIp(), 8080));
        }
        LoadBalancer chloadBalance = new ConsistentHashLoadBalancer();
        // 构造 10000 随机请求
        List<String> invocations = new ArrayList<>();
        for (int i = 0; i < 10000; i++) {
            invocations.add(UUID.randomUUID().toString());
        }
        int countDel = 0;
        // 下线20%节点
        List<Server> serverDel = servers.subList(0, 80);
        for (String invocation : invocations) {
            Server origin = chloadBalance.select(servers, invocation);
            Server delete = chloadBalance.select(serverDel, invocation);
            if (origin.getUrl().equals(delete.getUrl())) {
                countDel++;
            }
        }
        int countAdd = 0;
        for (String invocation : invocations) {
            Server origin = chloadBalance.select(servers, invocation);
            Server add = chloadBalance.select(serverAdd, invocation);
            if (origin.getUrl().equals(add.getUrl())) {
                countAdd++;
            }
        }
        log.info("下线20%节点命中率:{}", countDel / 10000D);
        log.info("上线20%节点命中率:{}", countAdd / 10000D);
    }
    // ip范围
    int[][] range = {{607649792, 608174079}, // 36.56.0.0-36.63.255.255
        {1038614528, 1039007743}, // 61.232.0.0-61.237.255.255
        {1783627776, 1784676351}, // 106.80.0.0-106.95.255.255
        {2035023872, 2035154943}, // 121.76.0.0-121.77.255.255
        {2078801920, 2079064063}, // 123.232.0.0-123.235.255.255
        {-1950089216, -1948778497}, // 139.196.0.0-139.215.255.255
        {-1425539072, -1425014785}, // 171.8.0.0-171.15.255.255
        {-1236271104, -1235419137}, // 182.80.0.0-182.92.255.255
        {-770113536, -768606209}, // 210.25.0.0-210.47.255.255
        {-569376768, -564133889}, // 222.16.0.0-222.95.255.255
    };
    /*
     * 将十进制转换成IP地址
     */
    public String num2ip(int ip) {
        int[] b = new int[4];
        String x = "";
        b[0] = (int) ((ip >> 24) & 0xff);
        b[1] = (int) ((ip >> 16) & 0xff);
        b[2] = (int) ((ip >> 8) & 0xff);
        b[3] = (int) (ip & 0xff);
        x = b[0] + "." + b[1] + "." + b[2] + "." + b[3];
        return x;
    }

    public String getRandomIp() {
        Random rdint = new Random();
        int index = rdint.nextInt(10);
        String ip = num2ip(
            range[index][0] + new Random().nextInt(range[index][1] - range[index][0]));
        return ip;

    }
}

测试结果

1
2
3
下线20%节点命中率:0.81

上线20%节点命中率:0.8352