dubbo2.6源码-负载均衡

秒速五厘米 2021-12-10 05:37 329阅读 0赞

生产者集群启动时，才会使用负载均衡，否则跳过

  1. <!--
  2.     use dubbo protocol to export service on port 20880
  3.     修改端口，即可启动集群，此时负载均衡才会使用
  4.     -->
  5.     <dubbo:protocol name="dubbo" port="20881"/>

LoadBalance

  1. @SPI(RandomLoadBalance.NAME)
  2. public interface LoadBalance {
  4.     /**
  5.      * select one invoker in list.
  6.      *
  7.      * @param invokers   invokers.
  8.      * @param url        refer url
  9.      * @param invocation invocation.
  10.      * @return selected invoker.
  11.      */
  12.     @Adaptive("loadbalance")
  13.     <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException;
  15. }

AbstractLoadBalance

  1. public abstract class AbstractLoadBalance implements LoadBalance {
  3.     static int calculateWarmupWeight(int uptime, int warmup, int weight) {
  4.         int ww = (int) ((float) uptime / ((float) warmup / (float) weight));
  5.         return ww < 1 ? 1 : (ww > weight ? weight : ww);
  6.     }
  8.     public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) {
  9.         if (invokers == null || invokers.isEmpty())
  10.             return null;
  11.         if (invokers.size() == 1)
  12.             return invokers.get(0);
  13.         return doSelect(invokers, url, invocation);
  14.     }
  16.     protected abstract <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation);
  18.     protected int getWeight(Invoker<?> invoker, Invocation invocation) {
  19.         int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT);
  20.         if (weight > 0) {
  21.             long timestamp = invoker.getUrl().getParameter(Constants.REMOTE_TIMESTAMP_KEY, 0L);
  22.             if (timestamp > 0L) {
  23.                 int uptime = (int) (System.currentTimeMillis() - timestamp);
  24.                 int warmup = invoker.getUrl().getParameter(Constants.WARMUP_KEY, Constants.DEFAULT_WARMUP);
  25.                 if (uptime > 0 && uptime < warmup) {
  26.                     weight = calculateWarmupWeight(uptime, warmup, weight);
  27.                 }
  28.             }
  29.         }
  30.         return weight;
  31.     }
  33. }

四种负载均衡策略

RandomLoadBalance随机加权，默认

  1. public class RandomLoadBalance extends AbstractLoadBalance {
  3.     public static final String NAME = "random";
  5.     private final Random random = new Random();
  7.     protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
  8.         int length = invokers.size(); // Number of invokers
  9.         int totalWeight = 0; // The sum of weights
  10.         boolean sameWeight = true; // Every invoker has the same weight?
  11.         for (int i = 0; i < length; i++) {
  12.             int weight = getWeight(invokers.get(i), invocation);
  13.             totalWeight += weight; // Sum
  14.             if (sameWeight && i > 0
  15.                     && weight != getWeight(invokers.get(i - 1), invocation)) {
  16.                 sameWeight = false;
  17.             }
  18.         }
  19.         if (totalWeight > 0 && !sameWeight) {
  20.             // If (not every invoker has the same weight & at least one invoker's weight>0), select randomly based on totalWeight.
  21.             int offset = random.nextInt(totalWeight);
  22.             // Return a invoker based on the random value.
  23.             for (int i = 0; i < length; i++) {
  24.                 offset -= getWeight(invokers.get(i), invocation);
  25.                 if (offset < 0) {
  26.                     return invokers.get(i);
  27.                 }
  28.             }
  29.         }
  30.         // If all invokers have the same weight value or totalWeight=0, return evenly.
  31.         return invokers.get(random.nextInt(length));
  32.     }

生产者改变权重

  1. <dubbo:provider delay="-1" timeout="1000000" retries="0" weight="80"/>

默认权重值时100，

我们现在假设集群有四个节点分别对应的权重为｛A:1,B:2,C:3,D:4｝，分别将权重套入到代码中进行分析，该随机算法按总权重进行加权随机，A节点负载请求的概率为1/(1+2+3+4)，依次类推，B，C，D负载的请求概率分别是20%，30%，40%。在这种方式下，用户可以根据机器的实际性能动态调整权重比率，如果发现机器D负载过大，请求堆积过多，通过调整权重可以缓解机器D处理请求的压力。

轮询加权

修改负载均衡策略

  1. <dubbo:reference id="userService" check="false" interface="com.study.dubbo.userapi.service.UserService" loadbalance="roundrobin"/>

消费者添加 loadbalance=“roundrobin”，就是类中Name常量

RoundRobinLoadBalance

  1. public class RoundRobinLoadBalance extends AbstractLoadBalance {
  3.     public static final String NAME = "roundrobin";
  5.     private final ConcurrentMap<String, AtomicPositiveInteger> sequences = new ConcurrentHashMap<String, AtomicPositiveInteger>();
  7.     protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
  8.         String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
  9.         int length = invokers.size(); // Number of invokers
  10.         int maxWeight = 0; // The maximum weight
  11.         int minWeight = Integer.MAX_VALUE; // The minimum weight
  12.         final LinkedHashMap<Invoker<T>, IntegerWrapper> invokerToWeightMap = new LinkedHashMap<Invoker<T>, IntegerWrapper>();
  13.         int weightSum = 0;
  14.         for (int i = 0; i < length; i++) {
  15.             int weight = getWeight(invokers.get(i), invocation);
  16.             maxWeight = Math.max(maxWeight, weight); // Choose the maximum weight
  17.             minWeight = Math.min(minWeight, weight); // Choose the minimum weight
  18.             if (weight > 0) {
  19.                 invokerToWeightMap.put(invokers.get(i), new IntegerWrapper(weight));
  20.                 weightSum += weight;
  21.             }
  22.         }
  23.         AtomicPositiveInteger sequence = sequences.get(key);
  24.         if (sequence == null) {
  25.             sequences.putIfAbsent(key, new AtomicPositiveInteger());
  26.             sequence = sequences.get(key);
  27.         }
  28.         int currentSequence = sequence.getAndIncrement();
  29.         if (maxWeight > 0 && minWeight < maxWeight) {
  30.             int mod = currentSequence % weightSum;
  31.             for (int i = 0; i < maxWeight; i++) {
  32.                 for (Map.Entry<Invoker<T>, IntegerWrapper> each : invokerToWeightMap.entrySet()) {
  33.                     final Invoker<T> k = each.getKey();
  34.                     final IntegerWrapper v = each.getValue();
  35.                     if (mod == 0 && v.getValue() > 0) {
  36.                         return k;
  37.                     }
  38.                     if (v.getValue() > 0) {
  39.                         v.decrement();
  40.                         mod--;
  41.                     }
  42.                 }
  43.             }
  44.         }
  45.         // Round robin
  46.         return invokers.get(currentSequence % length);
  47.     }
  49.     private static final class IntegerWrapper {
  50.         private int value;
  52.         public IntegerWrapper(int value) {
  53.             this.value = value;
  54.         }
  56.         public int getValue() {
  57.             return value;
  58.         }
  60.         public void setValue(int value) {
  61.             this.value = value;
  62.         }
  64.         public void decrement() {
  65.             this.value--;
  66.         }
  67.     }
  69. }

最少调用

LeastActiveLoadBalance

  1. public class LeastActiveLoadBalance extends AbstractLoadBalance {
  3.     public static final String NAME = "leastactive";
  5.     private final Random random = new Random();
  7.     protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
  8.         int length = invokers.size(); // Number of invokers
  9.         int leastActive = -1; // The least active value of all invokers
  10.         int leastCount = 0; // The number of invokers having the same least active value (leastActive)
  11.         int[] leastIndexs = new int[length]; // The index of invokers having the same least active value (leastActive)
  12.         int totalWeight = 0; // The sum of weights
  13.         int firstWeight = 0; // Initial value, used for comparision
  14.         boolean sameWeight = true; // Every invoker has the same weight value?
  15.         for (int i = 0; i < length; i++) {
  16.             Invoker<T> invoker = invokers.get(i);
  17.             int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive(); // Active number
  18.             int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT); // Weight
  19.             if (leastActive == -1 || active < leastActive) { // Restart, when find a invoker having smaller least active value.
  20.                 leastActive = active; // Record the current least active value
  21.                 leastCount = 1; // Reset leastCount, count again based on current leastCount
  22.                 leastIndexs[0] = i; // Reset
  23.                 totalWeight = weight; // Reset
  24.                 firstWeight = weight; // Record the weight the first invoker
  25.                 sameWeight = true; // Reset, every invoker has the same weight value?
  26.             } else if (active == leastActive) { // If current invoker's active value equals with leaseActive, then accumulating.
  27.                 leastIndexs[leastCount++] = i; // Record index number of this invoker
  28.                 totalWeight += weight; // Add this invoker's weight to totalWeight.
  29.                 // If every invoker has the same weight?
  30.                 if (sameWeight && i > 0
  31.                         && weight != firstWeight) {
  32.                     sameWeight = false;
  33.                 }
  34.             }
  35.         }
  36.         // assert(leastCount > 0)
  37.         if (leastCount == 1) {
  38.             // If we got exactly one invoker having the least active value, return this invoker directly.
  39.             return invokers.get(leastIndexs[0]);
  40.         }
  41.         if (!sameWeight && totalWeight > 0) {
  42.             // If (not every invoker has the same weight & at least one invoker's weight>0), select randomly based on totalWeight.
  43.             int offsetWeight = random.nextInt(totalWeight);
  44.             // Return a invoker based on the random value.
  45.             for (int i = 0; i < leastCount; i++) {
  46.                 int leastIndex = leastIndexs[i];
  47.                 offsetWeight -= getWeight(invokers.get(leastIndex), invocation);
  48.                 if (offsetWeight <= 0)
  49.                     return invokers.get(leastIndex);
  50.             }
  51.         }
  52.         // If all invokers have the same weight value or totalWeight=0, return evenly.
  53.         return invokers.get(leastIndexs[random.nextInt(leastCount)]);
  54.     }
  55. }

一致hash

使用consistenthash

  1. <dubbo:reference id="userService" check="false" interface="com.study.dubbo.userapi.service.UserService" loadbalance="consistenthash"/>
  3. public class ConsistentHashLoadBalance extends AbstractLoadBalance {
  5.     private final ConcurrentMap<String, ConsistentHashSelector<?>> selectors = new ConcurrentHashMap<String, ConsistentHashSelector<?>>();
  7.     @SuppressWarnings("unchecked")
  8.     @Override
  9.     protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
  10.         String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
  11.         int identityHashCode = System.identityHashCode(invokers);
  12.         ConsistentHashSelector<T> selector = (ConsistentHashSelector<T>) selectors.get(key);
  13.         if (selector == null || selector.identityHashCode != identityHashCode) {
  14.             selectors.put(key, new ConsistentHashSelector<T>(invokers, invocation.getMethodName(), identityHashCode));
  15.             selector = (ConsistentHashSelector<T>) selectors.get(key);
  16.         }
  17.         return selector.select(invocation);
  18.     }
  20.     private static final class ConsistentHashSelector<T> {
  22.         private final TreeMap<Long, Invoker<T>> virtualInvokers;
  24.         private final int replicaNumber;
  26.         private final int identityHashCode;
  28.         private final int[] argumentIndex;
  30.         ConsistentHashSelector(List<Invoker<T>> invokers, String methodName, int identityHashCode) {
  31.             this.virtualInvokers = new TreeMap<Long, Invoker<T>>();
  32.             this.identityHashCode = identityHashCode;
  33.             URL url = invokers.get(0).getUrl();
  34.             this.replicaNumber = url.getMethodParameter(methodName, "hash.nodes", 160);
  35.             String[] index = Constants.COMMA_SPLIT_PATTERN.split(url.getMethodParameter(methodName, "hash.arguments", "0"));
  36.             argumentIndex = new int[index.length];
  37.             for (int i = 0; i < index.length; i++) {
  38.                 argumentIndex[i] = Integer.parseInt(index[i]);
  39.             }
  40.             for (Invoker<T> invoker : invokers) {
  41.                 String address = invoker.getUrl().getAddress();
  42.                 for (int i = 0; i < replicaNumber / 4; i++) {
  43.                     byte[] digest = md5(address + i);
  44.                     for (int h = 0; h < 4; h++) {
  45.                         long m = hash(digest, h);
  46.                         virtualInvokers.put(m, invoker);
  47.                     }
  48.                 }
  49.             }
  50.         }
  52.         public Invoker<T> select(Invocation invocation) {
  53.             String key = toKey(invocation.getArguments());
  54.             byte[] digest = md5(key);
  55.             return selectForKey(hash(digest, 0));
  56.         }
  58.         private String toKey(Object[] args) {
  59.             StringBuilder buf = new StringBuilder();
  60.             for (int i : argumentIndex) {
  61.                 if (i >= 0 && i < args.length) {
  62.                     buf.append(args[i]);
  63.                 }
  64.             }
  65.             return buf.toString();
  66.         }
  68.         private Invoker<T> selectForKey(long hash) {
  69.             Map.Entry<Long, Invoker<T>> entry = virtualInvokers.tailMap(hash, true).firstEntry();
  70.             if (entry == null) {
  71.                 entry = virtualInvokers.firstEntry();
  72.             }
  73.             return entry.getValue();
  74.         }
  76.         private long hash(byte[] digest, int number) {
  77.             return (((long) (digest[3 + number * 4] & 0xFF) << 24)
  78.                     | ((long) (digest[2 + number * 4] & 0xFF) << 16)
  79.                     | ((long) (digest[1 + number * 4] & 0xFF) << 8)
  80.                     | (digest[number * 4] & 0xFF))
  81.                     & 0xFFFFFFFFL;
  82.         }
  84.         private byte[] md5(String value) {
  85.             MessageDigest md5;
  86.             try {
  87.                 md5 = MessageDigest.getInstance("MD5");
  88.             } catch (NoSuchAlgorithmException e) {
  89.                 throw new IllegalStateException(e.getMessage(), e);
  90.             }
  91.             md5.reset();
  92.             byte[] bytes;
  93.             try {
  94.                 bytes = value.getBytes("UTF-8");
  95.             } catch (UnsupportedEncodingException e) {
  96.                 throw new IllegalStateException(e.getMessage(), e);
  97.             }
  98.             md5.update(bytes);
  99.             return md5.digest();
  100.         }
  102.     }
  104. }

构造函数中，每个实际的提供者均有160个（默认值，可调整）虚拟节点，每个提供者对应的虚拟节点将平均散列到哈希环上，当有请求时，先计算该请求参数对应的哈希值，然后顺时针寻找最近的虚拟节点，得到实际的提供者节点。

ConsistentHashLoadBalance: 一致性Hash，相同参数的请求总是发到同一提供者。
当某一台提供者挂时，原本发往该提供者的请求，基于虚拟节点，平摊到其它提供者，不会引起剧烈变动。

发表评论

表情：
评论列表 （有 0 条评论，329人围观）

还没有评论，来说两句吧...

相关阅读

    相关 Dubbo源码 - 负载均衡

    前言 负载均衡，英文名称为Load Balance，其含义就是指将负载（工作任务）进行平衡、分摊到多个操作单元上进行运行。 例如：在Dubbo中，同一个服务有多个服务提

    女爷i女爷i/ 0 赞/ 182 阅读

    相关 Dubbo-负载均衡

    切勿将负载均衡策略写死在代码里，将来我们可以用控制台来进行控制。 负载均衡 在集群负载均衡时，Dubbo 提供了多种均衡策略，缺省为 `random` 随机调

    柔光的暖阳◎柔光的暖阳◎/ 0 赞/ 294 阅读