概述

由前面【Dubbo】集群容错机制 文章可知，AbstractClusterInvoker#doSelect会调用LoadBalance#select方法选取一个Invoker并返回。

首先，先来看一下LoadBalance接口，LoadBalance默认使用的负载均衡实现是随机算法，而且过url的loadbalance参数进行自适应选择扩展。

  1. @SPI(RandomLoadBalance.NAME)
  2. public interface LoadBalance {
  4.     /**
  5.      * @param invokers   供选择的invoker列表.
  6.      * @param url        refer url
  7.      * @param invocation 消费者的调用会被封装成一个invocaion
  8.      * @return 选择的invoker.
  9.      */
  10.     @Adaptive("loadbalance")
  11.     <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException;
  13. }

接下来看一下LoadBalance接口的类图，如下所示：LoadBalance是顶层接口，AbstractLoadBalance封装了通用的实现逻辑。AbstractLoadBalance抽象类包含四个子类，分别对应随机、一致性哈希、轮询、最少活跃调用（慢的provider收到更少的请求）。【Dubbo】负载均衡 - 图1

源码解析

AbstractLoadBalance

select方法

代码如下所示，逻辑比较简单，不多说。

  1. public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) {
  2.     //如果为空，返回空；如果只有一个，直接返回
  3.     if (invokers == null || invokers.isEmpty())
  4.         return null;
  5.     if (invokers.size() == 1)
  6.         return invokers.get(0);
  7.     //调用子类实现的模板方法
  8.     return doSelect(invokers, url, invocation);
  9. }
  11. protected abstract <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation);

getWeight方法

该方法用来获取权重，主要逻辑就是计算服务的运行时间，当运行时间小于预热时间时，进行降权，防止服务一启动就进入高负载状态。代码如下所示。

  1. protected int getWeight(Invoker<?> invoker, Invocation invocation) {
  2.     //获取URl里weight参数，默认100
  3.     int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT);
  4.     if (weight > 0) {
  5.         //获取provider启动时间戳
  6.         long timestamp = invoker.getUrl().getParameter(Constants.REMOTE_TIMESTAMP_KEY, 0L);
  7.         if (timestamp > 0L) {
  8.             //计算provider运行时间
  9.             int uptime = (int) (System.currentTimeMillis() - timestamp);
  10.             //获取服务预热时间，默认10分钟
  11.             int warmup = invoker.getUrl().getParameter(Constants.WARMUP_KEY, Constants.DEFAULT_WARMUP);
  12.             //如果运行时间小于预热时间，则重新计算权重
  13.             if (uptime > 0 && uptime < warmup) {
  14.                 weight = calculateWarmupWeight(uptime, warmup, weight);
  15.             }
  16.         }
  17.     }
  18.     return weight;
  19. }
  20. static int calculateWarmupWeight(int uptime, int warmup, int weight) {
  21.     //(uptime / warmup) * weight
  22.     //（运行时间/预热时间）* 权重
  23.     int ww = (int) ((float) uptime / ((float) warmup / (float) weight));
  24.     return ww < 1 ? 1 : (ww > weight ? weight : ww);
  25. }

RandomLoadBalance

RandomLoadBalance是加权随机负载均衡算法的具体实现，其算法思想比较简单，下面举俩🌰说明一下。

  1. 🌰1：
  2. [10,10,10]
  3. 三个Invoker的权重都一样，则直接取小于列表长度3的随机数（0，1，2），然后返回对应index的Invoker
  5. 🌰2：
  6. [3,4,5]
  7. 总权重=12，则会取小于12的随机数，假设随机数是6。
  8.   index=0,6-3=3
  9.   index=1,3-4=-1，-1小于0，所以index=1
  10. 返回列表内index为1的Invoker

代码如下所示：

  1. protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
  2.     int length = invokers.size(); // Number of invokers
  3.     int totalWeight = 0; // The sum of weights
  4.     boolean sameWeight = true; // Every invoker has the same weight?
  5.     //计算总权重，并记录下是否有不一样的权重值。
  6.     //如果权重值全都一样，则直接取小于列表大小的随机数，可看最后一步。
  7.     for (int i = 0; i < length; i++) {
  8.         int weight = getWeight(invokers.get(i), invocation);
  9.         totalWeight += weight; // Sum
  10.         if (sameWeight && i > 0
  11.                 && weight != getWeight(invokers.get(i - 1), invocation)) {
  12.             sameWeight = false;
  13.         }
  14.     }
  15.     //权重值存在不一样的情况。
  16.     //取一个小于总权重的随机数，然后从第一个开始遍历，一个个减掉Invoker对应的权重，当减到负数时，说明落在该Invoker对应的权重区间上。
  17.     if (totalWeight > 0 && !sameWeight) {
  18.         int offset = random.nextInt(totalWeight);
  19.         for (int i = 0; i < length; i++) {
  20.             offset -= getWeight(invokers.get(i), invocation);
  21.             if (offset < 0) {
  22.                 return invokers.get(i);
  23.             }
  24.         }
  25.     }
  26.     return invokers.get(random.nextInt(length));
  27. }

RoundRobinLoadBalance

RoundRobinLoadBalance是平滑加权轮询负载均衡算法的具体实现，主要包含以下逻辑：

  1. 遍历Invoker列表，如果Invoker对应的WeightedRoundRobin不存在，则新建一个并初始化权重
  2. 将当前current = current + weight，找出当前current最大的，并更新lastUpdate
  3. 删除长时间未更新lastUpdate的WeightedRoundRobin
  4. current = current - 总权重，返回选中的Invoker

下面看个例子：

  1. 🌰：
  2. [1,2,3] 对应 A B C三个Invoker  
  3.  第一次：[0,0,0] -> [1,2,3] -> [1,2,-3] 选C
  4.  第二次：[1,2,-3] -> [2,4,0] -> [2,-2,0] 选B
  5.  第三次：[2,-2,0] -> [3,0,3] -> [3,0,-3] 选C
  6.  第四次：[3,0,-3] -> [4,2,0] -> [-2,2,0] 选A
  7.  第五次：[-2,2,0] -> [-1,4,3] -> [-1,-2,3] 选B
  8.  第六次：[-1,-2,3] -> [0,0,6] -> [0,0,0] 选C
  9. A B C 分别对应 1 2 3次
  11. ps：
  12.  中间一列，每个都加上自身权重。
  13.  最后一列，会用最大的current建议一个totalWeight。
  14.  所以总加的数量和总减的数量是一样的。
  15.  自身权重越大的节点增长越快，那么比其他节点大的几率就越高，被选中的机会就越多。

代码如下所示：

  1. protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
  2.     //服务名+方法名
  3.     String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
  4.     ConcurrentMap<String, WeightedRoundRobin> map = methodWeightMap.get(key);
  5.     if (map == null) {
  6.         methodWeightMap.putIfAbsent(key, new ConcurrentHashMap<String, WeightedRoundRobin>());
  7.         map = methodWeightMap.get(key);
  8.     }
  9.     int totalWeight = 0;
  10.     long maxCurrent = Long.MIN_VALUE;
  11.     long now = System.currentTimeMillis();
  12.     Invoker<T> selectedInvoker = null;
  13.     WeightedRoundRobin selectedWRR = null;
  14.     //遍历所有Invoker
  15.     for (Invoker<T> invoker : invokers) {
  16.         String identifyString = invoker.getUrl().toIdentityString();
  17.         WeightedRoundRobin weightedRoundRobin = map.get(identifyString);
  18.         int weight = getWeight(invoker, invocation);
  19.         if (weight < 0) {
  20.             weight = 0;
  21.         }
  22.         //如果map中不存在对应的WeightedRoundRobin，则初始化一个put进去
  23.         if (weightedRoundRobin == null) {
  24.             weightedRoundRobin = new WeightedRoundRobin();
  25.             weightedRoundRobin.setWeight(weight);
  26.             map.putIfAbsent(identifyString, weightedRoundRobin);
  27.             weightedRoundRobin = map.get(identifyString);
  28.         }
  29.         //如果权重与通过getWeight得到的不一样，则设置成getWeight得到的
  30.         //这里不相等的原因可能是还在预热，因为在预热阶段，weight是会随着时间变大的。
  31.         if (weight != weightedRoundRobin.getWeight()) {
  32.             //weight changed
  33.             weightedRoundRobin.setWeight(weight);
  34.         }
  36.         //increaseCurrent = current + weight
  37.         long cur = weightedRoundRobin.increaseCurrent();
  38.         weightedRoundRobin.setLastUpdate(now);
  39.         //这里是为了找出最大的那一个
  40.         if (cur > maxCurrent) {
  41.             maxCurrent = cur;
  42.             selectedInvoker = invoker;
  43.             selectedWRR = weightedRoundRobin;
  44.         }
  45.         //累加权重
  46.         totalWeight += weight;
  47.     }
  49.     //上面那个for循环会更新lastUpdate，如果map里某个WeightedRoundRobin的lastupdate长时间未被更新，说明他可能挂了，所以把他移除掉。
  50.     if (!updateLock.get() && invokers.size() != map.size()) {
  51.         if (updateLock.compareAndSet(false, true)) {
  52.             try {
  53.                 // copy -> modify -> update reference
  54.                 ConcurrentMap<String, WeightedRoundRobin> newMap = new ConcurrentHashMap<String, WeightedRoundRobin>();
  55.                 newMap.putAll(map);
  56.                 Iterator<Entry<String, WeightedRoundRobin>> it = newMap.entrySet().iterator();
  57.                 while (it.hasNext()) {
  58.                     Entry<String, WeightedRoundRobin> item = it.next();
  59.                     if (now - item.getValue().getLastUpdate() > RECYCLE_PERIOD) {
  60.                         it.remove();
  61.                     }
  62.                 }
  63.                 methodWeightMap.put(key, newMap);
  64.             } finally {
  65.                 updateLock.set(false);
  66.             }
  67.         }
  68.     }
  69.     if (selectedInvoker != null) {
  70.         //current = current-weight
  71.         selectedWRR.sel(totalWeight);
  72.         return selectedInvoker;
  73.     }
  74.     // should not happen here
  75.     return invokers.get(0);
  76. }

LeastActiveLoadBalance

LeastActiveLoadBalance是最少活跃调用数负载均衡算法的集体实现，主要包含以下逻辑：

  1. 遍历invoker列表，找到最小活跃数的invoker数组（可能存在多个活跃数=最小活跃数的情况）
  2. 如果只有一个直接返回
  3. 如果存在多个，则根据权重是否相同来决定使用加权随机还是随机。

代码如下所示：

  1. protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
  2.     int length = invokers.size(); // Number of invokers
  3.     int leastActive = -1; // The least active value of all invokers
  4.     int leastCount = 0; // The number of invokers having the same least active value (leastActive)
  5.     int[] leastIndexs = new int[length]; // The index of invokers having the same least active value (leastActive)
  6.     int totalWeight = 0; // The sum of with warmup weights
  7.     int firstWeight = 0; // Initial value, used for comparision
  8.     boolean sameWeight = true; // Every invoker has the same weight value?
  9.     //遍历invokers列表
  10.     for (int i = 0; i < length; i++) {
  11.         Invoker<T> invoker = invokers.get(i);
  12.         //获取当前invoker的活跃适量
  13.         int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive(); // Active number
  14.         //权重
  15.         int afterWarmup = getWeight(invoker, invocation); // Weight
  16.         //如果活跃数量=-1 (说明是第一个遍历) 或者 当前活跃数量小于最小活跃数
  17.         //将当前活跃数量设置到最小活跃数，并更新其他字段
  18.         if (leastActive == -1 || active < leastActive) { // Restart, when find a invoker having smaller least active value.
  19.             leastActive = active; // Record the current least active value
  20.             leastCount = 1; // Reset leastCount, count again based on current leastCount
  21.             leastIndexs[0] = i; // Reset
  22.             totalWeight = afterWarmup; // Reset
  23.             firstWeight = afterWarmup; // Record the weight the first invoker
  24.             sameWeight = true; // Reset, every invoker has the same weight value?
  25.         }
  26.         //如果当前活跃数量 = 最小活跃数，则加入到leastIndexs数组，并累加权重
  27.         else if (active == leastActive) { // If current invoker's active value equals with leaseActive, then accumulating.
  28.             leastIndexs[leastCount++] = i; // Record index number of this invoker
  29.             totalWeight += afterWarmup; // Add this invoker's weight to totalWeight.
  30.             // If every invoker has the same weight?
  31.             //判断活跃数相等的，权重是否相等
  32.             if (sameWeight && i > 0
  33.                     && afterWarmup != firstWeight) {
  34.                 sameWeight = false;
  35.             }
  36.         }
  37.     }
  38.     // assert(leastCount > 0)
  39.     //如果只有一个活跃数=最小活跃数，直接返回
  40.     if (leastCount == 1) {
  41.         // If we got exactly one invoker having the least active value, return this invoker directly.
  42.         return invokers.get(leastIndexs[0]);
  43.     }
  44.     //存在多个活跃数=最小活跃数并且权重不相等的情况，进行加权随机
  45.     if (!sameWeight && totalWeight > 0) {
  46.         // If (not every invoker has the same weight & at least one invoker's weight>0), select randomly based on totalWeight.
  47.         int offsetWeight = random.nextInt(totalWeight) + 1;
  48.         // Return a invoker based on the random value.
  49.         for (int i = 0; i < leastCount; i++) {
  50.             int leastIndex = leastIndexs[i];
  51.             offsetWeight -= getWeight(invokers.get(leastIndex), invocation);
  52.             if (offsetWeight <= 0)
  53.                 return invokers.get(leastIndex);
  54.         }
  55.     }
  56.     // If all invokers have the same weight value or totalWeight=0, return evenly.
  57.     //存在多个活跃数=最小活跃数并且权重相等
  58.     return invokers.get(leastIndexs[random.nextInt(leastCount)]);
  59. }

ConsistentHashLoadBalance

一致性hash算法的原理此处不做解释，直接看一下Dubbo一致性hash负载均衡的实现。

doSelect方法

首先，ConsistentHashLoadBalance内部维护了一个ConcurrentMap，缓存了方法以及对应的ConsistentHashSelector。doSelect方法主要做的事情，就是通过hashCode来检测Invoker列表是否发生了变化，如果发生了变化，则重新初始化ConsistentHashSelector，然后调用ConsistentHashSelector#select方法选择一个Invoker并返回。代码如下所示

  1. private final ConcurrentMap<String, ConsistentHashSelector<?>> selectors = new ConcurrentHashMap<String, ConsistentHashSelector<?>>();
  3. @Override
  4. protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
  5.     String methodName = RpcUtils.getMethodName(invocation);
  6.     String key = invokers.get(0).getUrl().getServiceKey() + "." + methodName;
  7.     int identityHashCode = System.identityHashCode(invokers);
  8.     ConsistentHashSelector<T> selector = (ConsistentHashSelector<T>) selectors.get(key);
  9.     //如果没有初始化过或者invokers列表发生了变化，则需要重新初始化一次。
  10.     if (selector == null || selector.identityHashCode != identityHashCode) {
  11.         selectors.put(key, new ConsistentHashSelector<T>(invokers, methodName, identityHashCode));
  12.         selector = (ConsistentHashSelector<T>) selectors.get(key);
  13.     }
  14.     //选择一个Invoker
  15.     return selector.select(invocation);
  16. }

ConsistentHashSelector

接下来看一下ConsistentHashSelector的源码，注释很详细了，代码如下所示：

  1. private static final class ConsistentHashSelector<T> {
  3.     //虚拟节点，用TreeMap来实现，TreeMap的tailMap(K fromKey, boolean inclusive)方法可以获取比fromKey大的子map
  4.     private final TreeMap<Long, Invoker<T>> virtualInvokers;
  6.     //虚拟节点数量，默认会取160
  7.     private final int replicaNumber;
  9.     //这个用来校验Invoker列表是不是发生了变化
  10.     private final int identityHashCode;
  12.     //在配置中可以用hash.arguments指定需要进行hash值计算的参数，比如0,1,2就是用第0,1,2这三个参数，默认取0
  13.     private final int[] argumentIndex;
  15.     //初始化一致性hash选择器
  16.     ConsistentHashSelector(List<Invoker<T>> invokers, String methodName, int identityHashCode) {
  17.         this.virtualInvokers = new TreeMap<Long, Invoker<T>>();
  18.         this.identityHashCode = identityHashCode;
  19.         URL url = invokers.get(0).getUrl();
  20.         this.replicaNumber = url.getMethodParameter(methodName, "hash.nodes", 160);
  21.         String[] index = Constants.COMMA_SPLIT_PATTERN.split(url.getMethodParameter(methodName, "hash.arguments", "0"));
  22.         argumentIndex = new int[index.length];
  23.         for (int i = 0; i < index.length; i++) {
  24.             argumentIndex[i] = Integer.parseInt(index[i]);
  25.         }
  27.         //遍历invokers列表，
  28.         for (Invoker<T> invoker : invokers) {
  29.             String address = invoker.getUrl().getAddress();
  30.             for (int i = 0; i < replicaNumber / 4; i++) {
  31.                 byte[] digest = md5(address + i);
  32.                 for (int h = 0; h < 4; h++) {
  33.                     long m = hash(digest, h);
  34.                     virtualInvokers.put(m, invoker);
  35.                 }
  36.             }
  37.         }
  38.     }
  40.     public Invoker<T> select(Invocation invocation) {
  41.         //将参数转成key，根据argumentIndex拼接key
  42.         String key = toKey(invocation.getArguments());
  43.         //md5算法，生成16个字节数组
  44.         byte[] digest = md5(key);
  45.         //计算hash值，使用selectForKey方法查询大于该hash值的第一个节点
  46.         return selectForKey(hash(digest, 0));
  47.     }
  49.     //根据argumentIndex数据，拼接key，key是后面用来计算hash的
  50.     //具体逻辑是，根据argumentIndex里的值取参数，比如0,2，就会取args里index=0,2的参数进行拼接
  51.     private String toKey(Object[] args) {
  52.         StringBuilder buf = new StringBuilder();
  53.         for (int i : argumentIndex) {
  54.             if (i >= 0 && i < args.length) {
  55.                 buf.append(args[i]);
  56.             }
  57.         }
  58.         return buf.toString();
  59.     }
  61.     //使用TreeMap.tailMap方法获取大于hash的子map
  62.     private Invoker<T> selectForKey(long hash) {
  63.         Map.Entry<Long, Invoker<T>> entry = virtualInvokers.tailMap(hash, true).firstEntry();
  64.         if (entry == null) {
  65.             entry = virtualInvokers.firstEntry();
  66.         }
  67.         return entry.getValue();
  68.     }
  70.     private long hash(byte[] digest, int number) {
  71.         return (((long) (digest[3 + number * 4] & 0xFF) << 24)
  72.                 | ((long) (digest[2 + number * 4] & 0xFF) << 16)
  73.                 | ((long) (digest[1 + number * 4] & 0xFF) << 8)
  74.                 | (digest[number * 4] & 0xFF))
  75.                 & 0xFFFFFFFFL;
  76.     }
  78.     private byte[] md5(String value) {
  79.         MessageDigest md5;
  80.         try {
  81.             md5 = MessageDigest.getInstance("MD5");
  82.         } catch (NoSuchAlgorithmException e) {
  83.             throw new IllegalStateException(e.getMessage(), e);
  84.         }
  85.         md5.reset();
  86.         byte[] bytes;
  87.         try {
  88.             bytes = value.getBytes("UTF-8");
  89.         } catch (UnsupportedEncodingException e) {
  90.             throw new IllegalStateException(e.getMessage(), e);
  91.         }
  92.         md5.update(bytes);
  93.         return md5.digest();
  94.     }
  96. }

参考

Dubbo官网

《深入理解Apache Dubbo与实战》