向我的 RPC 框架引入负载均衡组件

1
2
3
4

public interface LoadBalance {
    //String 对应于能够提供服务的服务器地址
    String getServerAddress(RpcRequest request, List<ServiceInfo> serverList);
}

1
2
3
4
5
6
7
8

public class ServiceInfo {
    private String address;
    //-1 代表没有权重设置
    private int weight;
    //-1 代表没有连接数的数据
    private int invokeNumber;
  //省略其构造器、setter/getter 方法
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

public class RandomLoadBalance implements LoadBalance {

    private Random random;

    public RandomLoadBalance() {
        random = new Random();
    }

    @Override
    public String getServerAddress(RpcRequest rpcRequest, List<ServiceInfo> serverList) {
        //首先，确定随机整数的最大值
        int max = serverList.size();
        //在 0 - size-1 返回内进行随机
        int index = random.nextInt(max);//返回 [0,max)返回内的一个整数，左闭右开
        //返回服务器地址
        return serverList.get(index).getAddress();
    }

}

 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

public class WeightRandomLoadBalance implements LoadBalance {

    private Random random;

    public WeightRandomLoadBalance() {
        random = new Random();
    }

    @Override
    public String getServerAddress(RpcRequest rpcRequest, List<ServiceInfo> serverList) {
        int sum = 0;
        int[] weightArray = new int[serverList.size()];
        //遍历 serverList，计算权重和，并初始化 weightArray 数组
        for (int i = 0; i < serverList.size(); i++) {
            int weight = serverList.get(i).getWeight();
            weightArray[i] = weight;
            sum += weight;
        }
        //首先在 [0,sum) 范围内进行随机数的计算
        int rad = random.nextInt(sum);
        int cur_total = 0;

        for (int i = weightArray.length - 1; i >= 0; i--) {
            cur_total += weightArray[i];
            if (cur_total > rad) {
                return serverList.get(i).getAddress();
            }
        }
        return serverList.get(0).getAddress();
    }
}

 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

public class RoundRobinLoadBalance implements LoadBalance {

    //为每一个 RPC 的服务名创建一个轮询的索引计数
    //key 服务名，value 轮询所用的索引
    final Map<String, AtomicInteger> roundRobinMap = new HashMap<>();
    //注意线程安全性！！！！
    @Override
    public String getServerAddress(RpcRequest rpcRequest, List<ServiceInfo> serverList) {

        //得到轮询索引
        AtomicInteger index = roundRobinMap.get(rpcRequest.getInterfaceName());

        //进行 double check 式的空值初始化
        if (index == null) {
            synchronized (roundRobinMap) {
                if (roundRobinMap.get(rpcRequest.getInterfaceName())==null) {
                    index = new AtomicInteger(-1);
                    roundRobinMap.put(rpcRequest.getInterfaceName(), index);
                }
            }
        }
        //进行取余操作
        int serverListIndex = Math.abs(index.incrementAndGet() % serverList.size());
        //返回地址
        return serverList.get(serverListIndex).getAddress();
        /**
         * 注意，虽然 AtomicInteger 会有溢出的危险，但是当 AtomicInteger 达到最大值以后，
         * 并不会继续增大，而是会转换为 -2147483648，因此，只要进行取取余后的值进行取绝对值，就符合本轮询的使用方式（取余无所谓从整数变为负数）
         */
    }
}

 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

public class ConsistentHashLoadBalance implements LoadBalance {

    @Override
    public String getServerAddress(RpcRequest rpcRequest, List<ServiceInfo> serverList) {

        //从入口参数 List<ServiceInfo> serverList 处进行构造 Hash 环
        SortedMap<Integer, String> sortedMap = new TreeMap<>();
        /**
         * 利用 serverList 来构造 Hash 环:
         * 不过事实上并没有单独的 Hash 环作为数据结构，
         * Hash 环的存在重在我们如何来看待 SortedMap<Integer, String> 这一个数据结构
         * 把 Hash 环拉直，就是一个 key 为 hash，value 为服务器地址的有序的 HashMap(SortedMap)
         * 可以参考 URL:https://segmentfault.com/a/1190000021199728
         */
        for (int i = serverList.size() - 1; i >= 0; i--) {
            int hash = getHash(serverList.get(i).getAddress());
            sortedMap.put(hash,serverList.get(i).getAddress());
        }

        /**
         * 首先，我们利用 rpcRequest 请求的服务名进行 Hash 运算
         * 服务名是 接口名 还是 接口名+方法名 取决于 Provider 在服务注册中心的服务注册粒度，这里就仅仅以接口
         */
        //1.得到请求的 hash
        int hashOfRequest = getHash(rpcRequest.getInterfaceName());

        //2.得到大于 hashOfRequest 的所有 Map

        SortedMap<Integer,String> subMap = sortedMap.tailMap(hashOfRequest);

        if (subMap.isEmpty()) {
            //如果没有比该key的hash值大的，则从第一个node开始
            Integer i = sortedMap.firstKey();
            //返回对应的服务器
            return sortedMap.get(i);
        } else {
            //第一个Key就是顺时针过去离node最近的那个结点
            Integer i = subMap.firstKey();
            //返回对应的服务器
            return subMap.get(i);
        }
    }

    //一个基于 FNV1_32_HASH 的 Hash 算法，用于计算服务名的 hash 值
    private 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;
    }
}