dubbo负载均衡策略以及自定义负载均衡

dubbo负载均衡策略以及自定义负载均衡

最近在用dubbo做服务治理，用到了负载均衡，看了下dubbo的源码，整理下。

dubbo的负载均衡类图如下：

dubbo负载均衡策略以及自定义负载均衡

LoadBalance是顶层接口,提供了唯一的接口方法select,如下：

dubbo负载均衡策略以及自定义负载均衡

标注为@SPI的注解，只有标有@SPI注解的接口类才会查找扩展点的实现，dubbo依次从下面这三个路径读取扩展点文件：META-INF/dubbo/internal 、META-INF/dubbo/ 、META-INF/services/，其中dubbo内部实现的各种扩展文件都放在META-INF/dubbo/internal目录下面，如下定义

dubbo负载均衡策略以及自定义负载均衡

所以我们如果要动态扩展LoadBalance，只需要实现该接口，然后将全类名加入到扩展点即可。

AbstractLoadBalance:抽象类，实现了一些通用的权重计算方法，具体的负载均衡交给子类去实现doSelect方法，如下：

dubbo负载均衡策略以及自定义负载均衡

dubbo提供了四种负载均衡策略，如下：

dubbo负载均衡策略以及自定义负载均衡

下面一一介绍这四种负载均衡策略

1.RandomLoadBalance:按权重随机调用，这种方式是dubbo默认的负载均衡策略，源码如下：

实现思路很简单：如果服务多实例权重相同，则进行随机调用；如果权重不同，按照总权重取随机数

根据总权重数生成一个随机数，然后和具体服务实例的权重进行相减做偏移量，然后找出偏移量小于0的，比如随机数为10，某一个服务实例的权重为12，那么10-12=-2<0成立，则该服务被调用，这种策略在随机的情况下尽可能保证权重大的服务会被随机调用。

01 protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
02         int length = invokers.size(); // 总个数
03         int totalWeight = 0; // 总权重
04         boolean sameWeight = true; // 权重是否都一样
05         for (int i = 0; i < length; i++) {
06             int weight = getWeight(invokers.get(i), invocation);
07             totalWeight += weight; // 累计总权重
08             if (sameWeight && i > 0
09                     && weight != getWeight(invokers.get(i - 1), invocation)) {
10                 sameWeight = false; // 计算所有权重是否一样
11             }
12         }
13         if (totalWeight > 0 && ! sameWeight) {
14             // 如果权重不相同且权重大于0则按总权重数随机
15             int offset = random.nextInt(totalWeight);
16             // 并确定随机值落在哪个片断上
17             for (int i = 0; i < length; i++) {
18                 offset -= getWeight(invokers.get(i), invocation);
19                 if (offset < 0) {
20                     return invokers.get(i);
21                 }
22             }
23         }
24         // 如果权重相同或权重为0则均等随机
25         return invokers.get(random.nextInt(length));
26     }

2.RoundRobinLoadBalance：轮询，按公约后的权重设置轮询比率

实现思路：首先计算出多服务实例的最大最小权重，如果权重都一样（maxWeight=minWeight），则直接取模轮询；如果权重不一样，每一轮调用，都计算出一个基础的权重值，然后筛选出权重值大于基础权重值得invoker进行取模随机调用。

01 private final ConcurrentMap<String, AtomicPositiveInteger> sequences = new ConcurrentHashMap<String, AtomicPositiveInteger>();
02  
03 private final ConcurrentMap<String, AtomicPositiveInteger> weightSequences = new ConcurrentHashMap<String, AtomicPositiveInteger>();
04  
05 protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
06     String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
07     int length = invokers.size(); // 总个数
08     int maxWeight = 0; // 最大权重
09     int minWeight = Integer.MAX_VALUE; // 最小权重
10     for (int i = 0; i < length; i++) {
11         int weight = getWeight(invokers.get(i), invocation);
12         maxWeight = Math.max(maxWeight, weight); // 累计最大权重
13         minWeight = Math.min(minWeight, weight); // 累计最小权重
14     }
15     if (maxWeight > 0 && minWeight < maxWeight) { // 权重不一样
16         AtomicPositiveInteger weightSequence = weightSequences.get(key);
17         if (weightSequence == null) {
18             weightSequences.putIfAbsent(key, new AtomicPositiveInteger());
19             weightSequence = weightSequences.get(key);
20         }
21         int currentWeight = weightSequence.getAndIncrement() % maxWeight;
22         List<Invoker<T>> weightInvokers = new ArrayList<Invoker<T>>();
23         for (Invoker<T> invoker : invokers) { // 筛选权重大于当前权重基数的Invoker
24             if (getWeight(invoker, invocation) > currentWeight) {
25                 weightInvokers.add(invoker);
26             }
27         }
28         int weightLength = weightInvokers.size();
29         if (weightLength == 1) {
30             return weightInvokers.get(0);
31         } else if (weightLength > 1) {
32             invokers = weightInvokers;
33             length = invokers.size();
34         }
35     }
36     AtomicPositiveInteger sequence = sequences.get(key);
37     if (sequence == null) {
38         sequences.putIfAbsent(key, new AtomicPositiveInteger());
39         sequence = sequences.get(key);
40     }
41     // 取模轮循
42     return invokers.get(sequence.getAndIncrement() % length);
43 }

3.LeastActiveLoadBalance：最少活跃次数，dubbo框架自定义了一个Filter，用于计算服务被调用的次数，具体实现自己可以看源码

dubbo负载均衡策略以及自定义负载均衡

最小活跃次数思路：首先查找最小活跃数的服务并统计权重和出现的频次，如果最小活跃次数只出现一次，直接使用该服务；如果出现多次且权重不相同，则按照总权重数随机；如果出现多次且权重相同，则随机调用

01 protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
02         int length = invokers.size(); // 总个数
03         int leastActive = -1; // 最小的活跃数
04         int leastCount = 0; // 相同最小活跃数的个数
05         int[] leastIndexs = new int[length]; // 相同最小活跃数的下标
06         int totalWeight = 0; // 总权重
07         int firstWeight = 0; // 第一个权重，用于于计算是否相同
08         boolean sameWeight = true; // 是否所有权重相同
09         for (int i = 0; i < length; i++) {
10             Invoker<T> invoker = invokers.get(i);
11             int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive(); // 活跃数
12             int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT); // 权重
13             if (leastActive == -1 || active < leastActive) { // 发现更小的活跃数，重新开始
14                 leastActive = active; // 记录最小活跃数
15                 leastCount = 1; // 重新统计相同最小活跃数的个数
16                 leastIndexs[0] = i; // 重新记录最小活跃数下标
17                 totalWeight = weight; // 重新累计总权重
18                 firstWeight = weight; // 记录第一个权重
19                 sameWeight = true; // 还原权重相同标识
20             } else if (active == leastActive) { // 累计相同最小的活跃数
21                 leastIndexs[leastCount ++] = i; // 累计相同最小活跃数下标
22                 totalWeight += weight; // 累计总权重
23                 // 判断所有权重是否一样
24                 if (sameWeight && i > 0
25                         && weight != firstWeight) {
26                     sameWeight = false;
27                 }
28             }
29         }
30         // assert(leastCount > 0)
31         if (leastCount == 1) {
32             // 如果只有一个最小则直接返回
33             return invokers.get(leastIndexs[0]);
34         }
35         if (! sameWeight && totalWeight > 0) {
36             // 如果权重不相同且权重大于0则按总权重数随机
37             int offsetWeight = random.nextInt(totalWeight);
38             // 并确定随机值落在哪个片断上
39             for (int i = 0; i < leastCount; i++) {
40                 int leastIndex = leastIndexs[i];
41                 offsetWeight -= getWeight(invokers.get(leastIndex), invocation);
42                 if (offsetWeight <= 0)
43                     return invokers.get(leastIndex);
44             }
45         }
46         // 如果权重相同或权重为0则均等随机
47         return invokers.get(leastIndexs[random.nextInt(leastCount)]);
48     }

4.ConsistentHashLoadBalance：一致性hash

一致性Hash负载均衡涉及到两个主要的配置参数为hash.arguments 与hash.nodes。

hash.arguments ： 当进行调用时候根据调用方法的哪几个参数生成key，并根据key来通过一致性hash算法来选择调用结点

hash.nodes： 为结点的副本数。

01 @SuppressWarnings("unchecked")
02     @Override
03     protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
04         //获取调用方法名称
05         String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
06         //生成调用列表的hashcode
07         int identityHashCode = System.identityHashCode(invokers);
08         //根据方法名key获取一致性hash选择器
09         ConsistentHashSelector<T> selector = (ConsistentHashSelector<T>) selectors.get(key);
10         if (selector == null || selector.getIdentityHashCode() != identityHashCode) {
11             selectors.put(key, new ConsistentHashSelector<T>(invokers, invocation.getMethodName(), identityHashCode));
12             selector = (ConsistentHashSelector<T>) selectors.get(key);
13         }
14         //选择节点
15         return selector.select(invocation);
16     }
01 private static final class ConsistentHashSelector<T> {
02  
03         private final TreeMap<Long, Invoker<T>> virtualInvokers;//虚拟节点
04  
05         private final int                       replicaNumber;//副本数
06          
07         private final int                       identityHashCode;//调用节点的hashcode
08          
09         private final int[]                     argumentIndex;//参数索引数组
10  
11         public ConsistentHashSelector(List<Invoker<T>> invokers, String methodName, int identityHashCode) {
12             this.virtualInvokers = new TreeMap<Long, Invoker<T>>();
13             this.identityHashCode = System.identityHashCode(invokers);
14             URL url = invokers.get(0).getUrl();
15             //获取所配置的虚拟节点数，默认为160个
16             this.replicaNumber = url.getMethodParameter(methodName, "hash.nodes", 160);
17             String[] index = Constants.COMMA_SPLIT_PATTERN.split(url.getMethodParameter(methodName, "hash.arguments", "0"));
18             argumentIndex = new int[index.length];
19             for (int i = 0; i < index.length; i ++) {
20                 argumentIndex[i] = Integer.parseInt(index[i]);
21             }
22             //创建虚拟节点，对每一个Invoker生成replicaNumber个虚拟节点并存放于virtualInvokers中
23             for (Invoker<T> invoker : invokers) {
24                 for (int i = 0; i < replicaNumber / 4; i++) {
25                     byte[] digest = md5(invoker.getUrl().toFullString() + i);
26                     for (int h = 0; h < 4; h++) {
27                         long m = hash(digest, h);
28                         virtualInvokers.put(m, invoker);
29                     }
30                 }
31             }
32         }
33  
34       ..........省略..........
35  
36     }

5.自定义负载均衡策略

自定义类，只需要实现AbstractLoadBalance抽象类即可，然后将该类放入dubbo可发现的扩展点即可。

相关文章：

  • 2021-10-09
  • 2022-01-15
  • 2021-09-11
  • 2021-07-26
  • 2022-12-23
  • 2022-12-23
猜你喜欢
  • 2021-06-08
  • 2021-12-31
相关资源
相似解决方案
热门标签
Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode