Dubbo学习源码总结系列五--集群负载均衡

        Dubbo提供了哪些负载均衡机制？如何实现的?

        

        LoadBalance接口：能够看出，经过SPI机制默认为RandomLoadBalance，生成的适配器类执行select()方法。

 1 /**
 2  * LoadBalance. (SPI, Singleton, ThreadSafe) 3 * <p> 4 * <a href="http://en.wikipedia.org/wiki/Load_balancing_(computing)">Load-Balancing</a> 5  * 6  * @see com.alibaba.dubbo.rpc.cluster.Cluster#join(Directory) 7  */ 8 @SPI(RandomLoadBalance.NAME) 9 public interface LoadBalance { 10 11  /** 12  * select one invoker in list. 13  * 14  * @param invokers invokers. 15  * @param url refer url 16  * @param invocation invocation. 17  * @return selected invoker. 18  */ 19  @Adaptive("loadbalance") 20 <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException; 21 22 }

        实现类的基本类图以下所示：

 

        0、AbstractLoadBalance是LoadBalance接口的默认实现抽象类，为子类提供了实现框架。咱们来看看此类具体的实现：

        （1）主要方法固然是select()，经过选择实现了负载均衡策略。实现主要是调用doSelect()方法，它是个抽象方法，留给具体子类实现不一样的负载均衡策略；

        （2）getWeight()方法计算出invoker权重，计算公式为：weight = (int) (uptime(提供者正常运行时间) / warmup(升温时间) /weight(设定权重)))

 1     protected int getWeight(Invoker<?> invoker, Invocation invocation) {
 2         int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT); 3 if (weight > 0) { 4 long timestamp = invoker.getUrl().getParameter(Constants.REMOTE_TIMESTAMP_KEY, 0L); 5 if (timestamp > 0L) { 6 int uptime = (int) (System.currentTimeMillis() - timestamp); 7 int warmup = invoker.getUrl().getParameter(Constants.WARMUP_KEY, Constants.DEFAULT_WARMUP); 8 if (uptime > 0 && uptime < warmup) { 9 weight = calculateWarmupWeight(uptime, warmup, weight); 10  } 11  } 12  } 13 return weight; 14 }

  

        一、Random LoadBalance：随机，按权重设置随机几率。

        在一个截面上碰撞的几率高，但调用量越大分布越均匀，并且按几率使用权重后也比较均匀，有利于动态调整提供者权重。   

        RandomLoadBalance子类，主要经过doSelect()实现按权重的随机算法，实现逻辑为：

        （1）计算总权重；

        （2）若是没有设置权重或者全部权重都同样，直接从invokers列表随机返回一个；

        （3）不然：使用总权重随机计算一个offset（偏移量），循环invokers列表，offset=offset -（当前invoker权重），即剩余权重，而后返回第一个大于offset权重的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         for (int i = 0; i < length; i++) { 6 int weight = getWeight(invokers.get(i), invocation); 7 totalWeight += weight; // Sum 8 if (sameWeight && i > 0 9 && weight != getWeight(invokers.get(i - 1), invocation)) { 10 sameWeight = false; 11  } 12  } 13 if (totalWeight > 0 && !sameWeight) { 14 // If (not every invoker has the same weight & at least one invoker's weight>0), select randomly based on totalWeight. 15 int offset = random.nextInt(totalWeight); 16 // Return a invoker based on the random value. 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 // If all invokers have the same weight value or totalWeight=0, return evenly. 25 return invokers.get(random.nextInt(length)); 26 }

 

         二、RoundRobin LoadBalance：轮循，按公约后的权重设置轮循比率。

        存在慢的提供者累积请求的问题，好比：第二台机器很慢，但没挂，当请求调到第二台时就卡在那，长此以往，全部请求都卡在调到第二台上。

        RoundRobinLoadBalance子类，用doSelect()实现了按公约后的权重设置轮训比率， 

 1     protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
           //一个service接口的一个方法为一个key
 2         String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName(); 3 int length = invokers.size(); // Number of invokers 4 int maxWeight = 0; // The maximum weight 5 int minWeight = Integer.MAX_VALUE; // The minimum weight 6 final LinkedHashMap<Invoker<T>, IntegerWrapper> invokerToWeightMap = new LinkedHashMap<Invoker<T>, IntegerWrapper>(); 7 int weightSum = 0;
 //轮训计算总权重值、最大权重值、最小权重值 8 for (int i = 0; i < length; i++) { 9 int weight = getWeight(invokers.get(i), invocation); 10 maxWeight = Math.max(maxWeight, weight); // Choose the maximum weight 11 minWeight = Math.min(minWeight, weight); // Choose the minimum weight 12 if (weight > 0) { 13 invokerToWeightMap.put(invokers.get(i), new IntegerWrapper(weight)); 14 weightSum += weight; 15  } 16  }
 //给每一个请求方法设置一个原子Integer 17 AtomicPositiveInteger sequence = sequences.get(key); 18 if (sequence == null) { 19 sequences.putIfAbsent(key, new AtomicPositiveInteger()); 20 sequence = sequences.get(key); 21  } 22 int currentSequence = sequence.getAndIncrement(); 23 if (maxWeight > 0 && minWeight < maxWeight) { 24 int mod = currentSequence % weightSum; 25 for (int i = 0; i < maxWeight; i++) { 26 for (Map.Entry<Invoker<T>, IntegerWrapper> each : invokerToWeightMap.entrySet()) { 27 final Invoker<T> k = each.getKey(); 28 final IntegerWrapper v = each.getValue(); 29 if (mod == 0 && v.getValue() > 0) { 30 return k; 31  } 32 if (v.getValue() > 0) { 33  v.decrement(); 34 mod--; 35  } 36  } 37  } 38  } 39 // Round robin 40 return invokers.get(currentSequence % length); 41 }

         算法原理及实现讨论另外写了一篇博客，见：《负载均衡算法WeightedRoundRobin(加权轮询)简介及算法实现》

 

        三、LeastActive LoadBalance：最少活跃调用数，相同活跃数的随机，活跃数指调用先后计数差，即响应一次请求所花费的时长。

        使慢的提供者收到更少请求，由于越慢的提供者的调用先后计数差会越大。

        算法实现逻辑为：

        假如节点活跃数依次为{Node0=3ms,Node1=6ms,Node2=2ms,Node3=2ms,Node4=4ms}，

        （1）没有设置权重，或者权重都同样的状况下，遍历全部节点，找出节点中最小活跃数的节点，结果为{Node2=2ms,Node3=2ms}；

        （2）按照算法约束：相同活跃数的随机取，则从{Node2,Node3}中随机取出一个节点返回；

        （3）设置了权重，且权重不同的状况下，从最小活跃数子集{Node2,Node3}中取出权重大的一个节点返回。具体实现与随机访问算法Random LoadBalance相似，构造一个考虑了权重和轮询（多个相同权重的节点轮询选择）两个因素的算法，使用总权重随机计算一个offset（偏移量），循环invokers列表，offset = offset -（当前invoker权重），即剩余权重，而后返回第一个大于offset权重的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 leastActive = -1; // 记录最少的活跃数
 4         int leastCount = 0; // 拥有最少活跃数，且活跃数相同的节点个数
 5         int[] leastIndexs = new int[length]; // 最少活跃数节点索引数组（数组内节点的活跃数相同）
 6         int totalWeight = 0; // The sum of weights
 7         int firstWeight = 0; // Initial value, used for comparision
 8         boolean sameWeight = true; // Every invoker has the same weight value?
 9         for (int i = 0; i < length; i++) { 10 Invoker<T> invoker = invokers.get(i); 11 int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive(); //从上下文记录中取得方法的活跃数 Active number 12 int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT); // Weight 13 if (leastActive == -1 || active < leastActive) { // 当找到一个更小的活跃数节点时，重置变量Restart, when find a invoker having smaller least active value. 14 leastActive = active; // Record the current least active value 15 leastCount = 1; // Reset leastCount, count again based on current leastCount 16 leastIndexs[0] = i; // Reset 17 totalWeight = weight; // Reset 18 firstWeight = weight; // Record the weight the first invoker 19 sameWeight = true; // Reset, every invoker has the same weight value? 20 } else if (active == leastActive) { // If current invoker's active value equals with leaseActive, then accumulating. 21 leastIndexs[leastCount++] = i; // Record index number of this invoker 22 totalWeight += weight; // Add this invoker's weight to totalWeight. 23 // If every invoker has the same weight? 24 if (sameWeight && i > 0 25 && weight != firstWeight) { 26 sameWeight = false; 27  } 28  } 29  } 30 // assert(leastCount > 0) 31 if (leastCount == 1) { 32 // If we got exactly one invoker having the least active value, return this invoker directly. 33 return invokers.get(leastIndexs[0]); 34  } 35 if (!sameWeight && totalWeight > 0) { 36 // If (not every invoker has the same weight & at least one invoker's weight>0), select randomly based on totalWeight.
 //考虑了权重和轮询（多个相同权重的节点轮询选择）两个因素的算法，使用总权重随机计算一个offset（偏移量），循环invokers列表，offset = offset -（当前invoker权重），即剩余权重，而后返回第一个大于offset权重的invoker 37 int offsetWeight = random.nextInt(totalWeight); 38 // Return a invoker based on the random value. 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 // If all invokers have the same weight value or totalWeight=0, return evenly. 47 return invokers.get(leastIndexs[random.nextInt(leastCount)]); 48 }

  

        四、ConsistentHash LoadBalance：一致性哈希。适用场景为：相同参数的请求始终发送到同一个提供者。

• 当某一台提供者挂时，本来发往该提供者的请求，基于虚拟节点，平摊到其它提供者，不会引发剧烈变更。
• 具体算法原理和实现讨论另外写了一篇博客，见：《一致性哈希算法原理分析及实现》
  
• 缺省只对第一个参数 Hash，若是要修改，请配置 <dubbo:parameter key="hash.arguments" value="0,1" />
• 缺省用 160 份虚拟节点，若是要修改，请配置 <dubbo:parameter key="hash.nodes" value="320" />
具体dubbo实现以下：

 1     private static final class ConsistentHashSelector<T> {
 2 
 3         private final TreeMap<Long, Invoker<T>> virtualInvokers; 4 5 private final int replicaNumber; 6 7 private final int identityHashCode; 8 9 private final int[] argumentIndex; 10 11 ConsistentHashSelector(List<Invoker<T>> invokers, String methodName, int identityHashCode) { 12 this.virtualInvokers = new TreeMap<Long, Invoker<T>>(); 13 this.identityHashCode = identityHashCode; 14 URL url = invokers.get(0).getUrl();
 //没有设置，默认虚拟节点（分片）数160个 15 this.replicaNumber = url.getMethodParameter(methodName, "hash.nodes", 160); 16 String[] index = Constants.COMMA_SPLIT_PATTERN.split(url.getMethodParameter(methodName, "hash.arguments", "0")); 17 argumentIndex = new int[index.length]; 18 for (int i = 0; i < index.length; i++) { 19 argumentIndex[i] = Integer.parseInt(index[i]); 20  } 21 for (Invoker<T> invoker : invokers) { 22 String address = invoker.getUrl().getAddress(); 23 for (int i = 0; i < replicaNumber / 4; i++) { 24 byte[] digest = md5(address + i); 25 for (int h = 0; h < 4; h++) { 26 long m = hash(digest, h); 27  virtualInvokers.put(m, invoker); 28  } 29  } 30  } 31  } 32 33 public Invoker<T> select(Invocation invocation) { 34 String key = toKey(invocation.getArguments()); 35 byte[] digest = md5(key); 36 return selectForKey(hash(digest, 0)); 37  } 38 39 private String toKey(Object[] args) { 40 StringBuilder buf = new StringBuilder(); 41 for (int i : argumentIndex) { 42 if (i >= 0 && i < args.length) { 43  buf.append(args[i]); 44  } 45  } 46 return buf.toString(); 47  } 48 49 private Invoker<T> selectForKey(long hash) { 50 Invoker<T> invoker; 51 Long key = hash; 52 if (!virtualInvokers.containsKey(key)) { 53 SortedMap<Long, Invoker<T>> tailMap = virtualInvokers.tailMap(key); 54 if (tailMap.isEmpty()) { 55 key = virtualInvokers.firstKey(); 56 } else { 57 key = tailMap.firstKey(); 58  } 59  } 60 invoker = virtualInvokers.get(key); 61 return invoker; 62  } 63 64 private long hash(byte[] digest, int number) { 65 return (((long) (digest[3 + number * 4] & 0xFF) << 24) 66 | ((long) (digest[2 + number * 4] & 0xFF) << 16) 67 | ((long) (digest[1 + number * 4] & 0xFF) << 8) 68 | (digest[number * 4] & 0xFF)) 69 & 0xFFFFFFFFL; 70  } 71 72 private byte[] md5(String value) { 73  MessageDigest md5; 74 try { 75 md5 = MessageDigest.getInstance("MD5"); 76 } catch (NoSuchAlgorithmException e) { 77 throw new IllegalStateException(e.getMessage(), e); 78  } 79  md5.reset(); 80 byte[] bytes; 81 try { 82 bytes = value.getBytes("UTF-8"); 83 } catch (UnsupportedEncodingException e) { 84 throw new IllegalStateException(e.getMessage(), e); 85  } 86  md5.update(bytes); 87 return md5.digest(); 88  } 89 90 }

 

        配置：

        服务端服务级别：<dubbo:service interface="..." loadbalance="roundrobin" />

        服务端方法级别：<dubbo:service interface="..."><dubbo:method name="..." loadbalance="roundrobin"/></dubbo:service>

        客户端服务级别：<dubbo:reference interface="..." loadbalance="roundrobin" />

        客户端方法级别：<dubbo:reference interface="..."><dubbo:method name="..." loadbalance="roundrobin"/></dubbo:reference>