Dubbo官网负载均衡介绍

负载均衡策略

目前 Dubbo 内置了如下负载均衡算法，用户可直接配置使用：

算法 特性 备注
RandomLoadBalance 加权随机 默认算法，默认权重相同
RoundRobinLoadBalance 加权轮询 借鉴于 Nginx 的平滑加权轮询算法，默认权重相同，
LeastActiveLoadBalance 最少活跃优先 + 加权随机 背后是能者多劳的思想
ShortestResponseLoadBalance 最短响应优先 + 加权随机 更加关注响应速度
ConsistentHashLoadBalance 一致性 Hash 确定的入参，确定的提供者，适用于有状态请求

源码版本

查看Dubbo源码版本是2.77，maven依赖如下：

  1. <dependency>
  2.             <groupId>org.apache.dubbo</groupId>
  3.             <artifactId>dubbo-spring-boot-starter</artifactId>
  4.             <version>${dubbo-starter.version}</version>
  5. </dependency>

负载均衡的抽象父类

  1. public abstract class AbstractLoadBalance implements LoadBalance {
  3.     static int calculateWarmupWeight(int uptime, int warmup, int weight) {
  4.           //uptime：启动时间 warmup:invoker的预热时间,weight:invoker配置的权重
  5.         //权重=uptime/(warmup/weight)
  6.         int ww = (int) ( uptime / ((float) warmup / weight));
  7.           //计算后的权重小于1则返回1，否则去计算后的权重和配置的权重中较小那个
  8.         return ww < 1 ? 1 : (Math.min(ww, weight));
  9.     }
  11.     @Override
  12.     public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) {
  13.         if (CollectionUtils.isEmpty(invokers)) {
  14.             return null;
  15.         }
  16.         if (invokers.size() == 1) {
  17.             return invokers.get(0);
  18.         }
  19.         return doSelect(invokers, url, invocation);
  20.     }
  22.       //抽象方法，由负载均衡实现类实现方法
  23.     protected abstract <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation);
  27.     int getWeight(Invoker<?> invoker, Invocation invocation) {
  28.         int weight;
  29.         URL url = invoker.getUrl();
  30.         // Multiple registry scenario, load balance among multiple registries
  31.         if (REGISTRY_SERVICE_REFERENCE_PATH.equals(url.getServiceInterface())) {
  32.             weight = url.getParameter(REGISTRY_KEY + "." + WEIGHT_KEY, DEFAULT_WEIGHT);
  33.         } else {
  34.             // 获取invoker配置的权重，默认权重（int DEFAULT_WEIGHT = 100;）
  35.             weight = url.getMethodParameter(invocation.getMethodName(), WEIGHT_KEY, DEFAULT_WEIGHT);
  36.             if (weight > 0) {
  37.                   // 获取invoker的启动时间戳
  38.                 long timestamp = invoker.getUrl().getParameter(TIMESTAMP_KEY, 0L);
  39.                 if (timestamp > 0L) {
  40.                       //invoker 启动时间
  41.                     long uptime = System.currentTimeMillis() - timestamp;
  42.                     if (uptime < 0) {
  43.                         return 1;
  44.                     }
  45.                       // invoker的预热时间，默认10分钟（int DEFAULT_WARMUP = 10 * 60 * 1000;）
  46.                     int warmup = invoker.getUrl().getParameter(WARMUP_KEY, DEFAULT_WARMUP);
  47.                     // 如果服务启动时间小于预热时间，则重新计算权重
  48.                       if (uptime > 0 && uptime < warmup) {
  49.                           //计算算法
  50.                         weight = calculateWarmupWeight((int)uptime, warmup, weight);
  51.                     }
  52.                 }
  53.             }
  54.         }
  55.         return Math.max(weight, 0);
  56.     }
  57. }

加权随机(默认)

源码

  1. public class RandomLoadBalance extends AbstractLoadBalance {
  3.     public static final String NAME = "random";
  5.     /**
  6.      * Select one invoker between a list using a random criteria
  7.      * @param invokers List of possible invokers
  8.      * @param url URL
  9.      * @param invocation Invocation
  10.      * @param <T>
  11.      * @return The selected invoker
  12.      */
  13.     @Override
  14.     protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
  15.         // invoker集合长度
  16.         int length = invokers.size();
  17.         // 先假设所有invoker的权重都一样
  18.         boolean sameWeight = true;
  19.         // invokers是个集合，weights记录各个invoker的权重
  20.         int[] weights = new int[length];
  21.         // 先获取第一个invoker的权重
  22.         int firstWeight = getWeight(invokers.get(0), invocation);
  23.         weights[0] = firstWeight;
  24.         // 所有invoker总权重,
  25.         int totalWeight = firstWeight;
  26.         // 循环更新weights,累加totalWeight,并根据权重修改sameWeight
  27.         for (int i = 1; i < length; i++) {
  28.             int weight = getWeight(invokers.get(i), invocation);
  29.             // save for later use
  30.             weights[i] = weight;
  31.             // Sum
  32.             totalWeight += weight;
  33.             if (sameWeight && weight != firstWeight) {
  34.                 sameWeight = false;
  35.             }
  36.         }
  37.         // 存在invoker权重不一致，总权重大于0
  38.         if (totalWeight > 0 && !sameWeight) {
  39.             // 随机获取[0,totalWeight)范围的数值，取得一个随机权重
  40.             int offset = ThreadLocalRandom.current().nextInt(totalWeight);
  41.             // 下面的循环是从权重数组里获取对应invoker的权重
  42.             for (int i = 0; i < length; i++) {
  43.                 // 随机权重减去invoker的权重值
  44.                 offset -= weights[i];
  45.                 if (offset < 0) {
  46.                     // 如果随机权重见到小于0后，返回当前下标的invoker
  47.                     return invokers.get(i);
  48.                 }
  49.             }
  50.         }
  51.         // 如果总权重为0,或者所有权重一样,随机返回一个
  52.         return invokers.get(ThreadLocalRandom.current().nextInt(length));
  53.     }
  55. }

最短优先+加权轮询随机、最少活跃优先+加权随机策略最后的实现与加权随机基本一致。

加权轮询

官网介绍

  • 加权轮询，按公约后的权重设置轮询比率，循环调用节点
  • 缺点：同样存在慢的提供者累积请求的问题。

加权轮询过程过程中，如果某节点权重过大，会存在某段时间内调用过于集中的问题。
例如 ABC 三节点有如下权重：{A: 3, B: 2, C: 1}
那么按照最原始的轮询算法，调用过程将变成：A A A B B C

对此，Dubbo 借鉴 Nginx 的平滑加权轮询算法，对此做了优化，调用过程可抽象成下表:

轮前加和权重 本轮胜者 合计权重 轮后权重（胜者减去合计权重）
起始轮 \ \ A(0), B(0), C(0)
A(3), B(2), C(1) A 6 A(-3), B(2), C(1)
A(0), B(4), C(2) B 6 A(0), B(-2), C(2)
A(3), B(0), C(3) A 6 A(-3), B(0), C(3)
A(0), B(2), C(4) C 6 A(0), B(2), C(-2)
A(3), B(4), C(-1) B 6 A(3), B(-2), C(-1)
A(6), B(0), C(0) A 6 A(0), B(0), C(0)

我们发现经过合计权重（3+2+1）轮次后，循环又回到了起点，整个过程中节点流量是平滑的，且哪怕在很短的时间周期内，概率都是按期望分布的。

如果用户有加权轮询的需求，可放心使用该算法。

源码

  1. public class RoundRobinLoadBalance extends AbstractLoadBalance {
  2.     public static final String NAME = "roundrobin";
  4.     private static final int RECYCLE_PERIOD = 60000;
  6.     protected static class WeightedRoundRobin {
  7.           // 服务提供者权重
  8.         private int weight;
  9.           // 当前权重
  10.         private AtomicLong current = new AtomicLong(0);
  11.           // 最后一次更新时间
  12.         private long lastUpdate;
  14.         public int getWeight() {
  15.             return weight;
  16.         }
  18.         public void setWeight(int weight) {
  19.             this.weight = weight;
  20.               // 初始值为0
  21.             current.set(0);
  22.         }
  24.         public long increaseCurrent() {
  25.               // CAS操作 current = current + weight
  26.             return current.addAndGet(weight);
  27.         }
  29.         public void sel(int total) {
  30.               // CAS操作 current = current - total
  31.             current.addAndGet(-1 * total);
  32.         }
  34.         public long getLastUpdate() {
  35.             return lastUpdate;
  36.         }
  38.         public void setLastUpdate(long lastUpdate) {
  39.             this.lastUpdate = lastUpdate;
  40.         }
  41.     }
  43.     private ConcurrentMap<String, ConcurrentMap<String, WeightedRoundRobin>> methodWeightMap = new ConcurrentHashMap<String, ConcurrentMap<String, WeightedRoundRobin>>();
  45.     /**
  46.      * get invoker addr list cached for specified invocation
  47.      * <p>
  48.      * <b>for unit test only</b>
  49.      *
  50.      * @param invokers
  51.      * @param invocation
  52.      * @return
  53.      */
  54.     protected <T> Collection<String> getInvokerAddrList(List<Invoker<T>> invokers, Invocation invocation) {
  55.         String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
  56.         Map<String, WeightedRoundRobin> map = methodWeightMap.get(key);
  57.         if (map != null) {
  58.             return map.keySet();
  59.         }
  60.         return null;
  61.     }
  63.     @Override
  64.     protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
  65.           // 获取url到weightRoundRobin映射表，如果为空，则创建一个新的
  66.         String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
  67.         ConcurrentMap<String, WeightedRoundRobin> map = methodWeightMap.computeIfAbsent(key, k -> new ConcurrentHashMap<>());
  68.         int totalWeight = 0;
  69.         long maxCurrent = Long.MIN_VALUE;
  70.         long now = System.currentTimeMillis();
  71.         Invoker<T> selectedInvoker = null;
  72.         WeightedRoundRobin selectedWRR = null;          
  73.         for (Invoker<T> invoker : invokers) {
  74.             String identifyString = invoker.getUrl().toIdentityString();
  75.             int weight = getWeight(invoker, invocation);
  76.               //检测当前Invoker是否有相应的WeightedRoundRobin,没有则创建
  77.             WeightedRoundRobin weightedRoundRobin = map.computeIfAbsent(identifyString, k -> {
  78.                 WeightedRoundRobin wrr = new WeightedRoundRobin();
  79.                   // 设置Invoker 权重
  80.                 wrr.setWeight(weight);
  81.                 return wrr;
  82.             });
  83.                         // Invoker 权重不等于weightedRoundRobin中保存的权重，说明权重变化，进行更新
  84.             if (weight != weightedRoundRobin.getWeight()) {
  85.                 //weight changed
  86.                 weightedRoundRobin.setWeight(weight);
  87.             }
  88.               // CAS操作 cur = current+weight
  89.             long cur = weightedRoundRobin.increaseCurrent();
  90.               // 设置 lastUpdate，表示近期更新过
  91.             weightedRoundRobin.setLastUpdate(now);
  92.             if (cur > maxCurrent) {
  93.                 maxCurrent = cur;
  94.                   // 将具有最大current 权重的Invoker赋值给selectedInvoker
  95.                 selectedInvoker = invoker;
  96.                   // 将Invoker对应的weightedRoundRobin赋值给selectedWRR，后面用到
  97.                 selectedWRR = weightedRoundRobin;
  98.             }
  99.               //计算权重之和
  100.             totalWeight += weight;
  101.         }
  102.           // 如果Invoker个数与ConcurrentMap<String, WeightedRoundRobin>的key个数不一致
  103.         if (invokers.size() != map.size()) {
  104.               //去除长时间未被更新的节点（60s）
  105.             map.entrySet().removeIf(item -> now - item.getValue().getLastUpdate() > RECYCLE_PERIOD);
  106.         }
  107.         if (selectedInvoker != null) {
  108.               // CAS操作，将最大权重的Invoker减去权重总和
  109.             selectedWRR.sel(totalWeight);
  110.               // 返回具有最大current的Invoker
  111.             return selectedInvoker;
  112.         }
  113.         // should not happen here
  114.         return invokers.get(0);
  115.     }
  117. }

最短响应优先 + 加权随机策略

  1. org.apache.dubbo.rpc.cluster.loadbalance.ShortestResponseLoadBalance

主要流程如下：

  1. 从多个服务提供者中选择出调用成功的且响应时间最短的服务提供者，由于满足这样条件的服务提供者有可能有多个。所以当选择出多个服务提供者后要根据他们的权重做分析
  2. 如果只选择出来了一个，直接用选出来这个
  3. 如果真的有多个，看它们的权重是否一样，如果不一样，则走加权随机算法的逻辑
  4. 如果它们的权重是一样的，则随机调用一个

负载均衡实现代码如下：

  1. protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
  2.     //服务提供者的数量
  3.     int length = invokers.size();
  4.     //所有服务提供者的估计最短响应时间(只是一个临时变量，在循环中存储当前最短响应时间是多少)
  5.     long shortestResponse = Long.MAX_VALUE;
  6.     //具有相同最短响应时间的服务提供者个数，初始化为 0
  7.     int shortestCount = 0;
  8.     //数组里面放的是具有相同最短响应时间的服务提供者的下标
  9.     int[] shortestIndexes = new int[length];
  10.     //每一个服务提供者的权重
  11.     int[] weights = new int[length];
  12.     //多个具有相同最短响应时间的服务提供者对应的预热
  13.     int totalWeight = 0;
  14.     //第一个具有最短响应时间的服务提供者的权重
  15.     int firstWeight = 0;
  16.     //多个满足条件的提供者的权重是否一致
  17.     boolean sameWeight = true;
  19.     // Filter out all the shortest response invokers
  20.     for (int i = 0; i < length; i++) {
  21.         Invoker<T> invoker = invokers.get(i);
  22.         RpcStatus rpcStatus = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName());
  23.         //获取调用成功的平均时间
  24.           //getSucceededAverageElapsed():调用成功的请求数总数对应的总耗时 / 调用成功的请求数总数
  25.         long succeededAverageElapsed = rpcStatus.getSucceededAverageElapsed();
  26.         int active = rpcStatus.getActive();
  27.         long estimateResponse = succeededAverageElapsed * active;
  28.           //获取预热后的权重
  29.         int afterWarmup = getWeight(invoker, invocation);
  30.         weights[i] = afterWarmup;
  31.         // Same as LeastActiveLoadBalance
  32.         if (estimateResponse < shortestResponse) {
  33.               //如果出现有更短的响应时间的服务提供者，记录更短的响应时间和当前服务者的下标
  34.             shortestResponse = estimateResponse;
  35.             shortestCount = 1;
  36.             shortestIndexes[0] = i;
  37.             totalWeight = afterWarmup;
  38.             firstWeight = afterWarmup;
  39.             sameWeight = true;
  40.         } else if (estimateResponse == shortestResponse) {
  41.               //如果出现时间一样长的服务提供者，更新shortestIndexes下标，计算总权重，判断权重是否相等
  42.             shortestIndexes[shortestCount++] = i;
  43.             totalWeight += afterWarmup;
  44.             if (sameWeight && i > 0
  45.                     && afterWarmup != firstWeight) {
  46.                 sameWeight = false;
  47.             }
  48.         }
  49.     }
  50.     if (shortestCount == 1) {
  51.           //经过选择后只有一个服务提供者，直接返回这个服务提供者
  52.         return invokers.get(shortestIndexes[0]);
  53.     }
  54.     //加权随机负载均衡的实现
  55.     if (!sameWeight && totalWeight > 0) {
  56.         int offsetWeight = ThreadLocalRandom.current().nextInt(totalWeight);
  57.         for (int i = 0; i < shortestCount; i++) {
  58.             int shortestIndex = shortestIndexes[i];
  59.             offsetWeight -= weights[shortestIndex];
  60.             if (offsetWeight < 0) {
  61.                 return invokers.get(shortestIndex);
  62.             }
  63.         }
  64.     }
  65.       //从多个满足条件且权重一样的服务提供者中随机选择一个
  66.     return invokers.get(shortestIndexes[ThreadLocalRandom.current().nextInt(shortestCount)]);
  67. }

最少活跃优先 + 加权随机策略

  1. org.apache.dubbo.rpc.cluster.loadbalance.LeastActiveLoadBalance

代码如下：

  1. public class LeastActiveLoadBalance extends AbstractLoadBalance {
  3.     public static final String NAME = "leastactive";
  5.     @Override
  6.     protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
  7.         //首先获取服务提供者（invoker)的数量，目前我们的服务提供者大小为3
  8.         int length = invokers.size();
  9.         //初始化最小活跃次数(minActive)
  10.           //活跃次数一定是一个大于等于0的数，所以这里设置为-1就是为了minActive绐替換掉
  11.         int leastActive = -1;
  12.         // minactive相同的invoker的数量
  13.         int leastCount = 0;
  14.         // invokers是个集合，leastIndexes的作用是记录minActive相同的invoker下标
  15.         int[] leastIndexes = new int[length];
  16.         // invokers是个集合，weights记录各个invoker的权重，当前权重是[5,10,15]
  17.         int[] weights = new int[length];
  18.         // 用于记录所有invoker的权重之和，当前权重是5+10+15；
  19.         int totalWeight = 0;
  20.         // 记录第一个minActive的invoker的权重
  21.                 // 在下面的循环中和其他的具有相同minActive的invoker进行对比
  22.           // 作用是判断是否所有有相同minActive的invoker的权重都一样
  23.           // 例子：A,8两个服务器权重都配置的200,minActive也相等，则随机选择一个
  24.         int firstWeight = 0;
  25.         // 先假设所有invoker的权重都一样
  26.         boolean sameWeight = true;
  29.         // Filter out all the least active invokers
  30.         for (int i = 0; i < length; i++) {
  31.               // 遍历出每个invoker
  32.             Invoker<T> invoker = invokers.get(i);
  33.             // 读取当前invoker的活跃数
  34.             int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive();
  35.             // 获取当前invoker的权重
  36.             int afterWarmup = getWeight(invoker, invocation);
  37.             // 将invoker的权重存储到weights对应的下标
  38.             weights[i] = afterWarmup;
  39.             // 第一次循环的时候minActive从leastActive=-1变成了leastActive = active;（当前invoker活跃数）
  40.               // 后续循环，找到活跃数更小的进行替换
  41.               // leastCount、leastIndexs、totalWeight、firstWeight、sameWeight都重新开始
  42.             if (leastActive == -1 || active < leastActive) {
  43.                 // 替换活跃度更小的值
  44.                 leastActive = active;
  45.                 // 最小活跃度的invoker重置为1
  46.                 leastCount = 1;
  47.                 // 记录活跃度更小的invoker的下标到leastIndexes
  48.                 leastIndexes[0] = i;
  49.                 // Reset totalWeight
  50.                 totalWeight = afterWarmup;
  51.                 // Record the weight the first least active invoker
  52.                 firstWeight = afterWarmup;
  53.                 // Each invoke has the same weight (only one invoker here)
  54.                 sameWeight = true;
  55.                 // If current invoker's active value equals with leaseActive, then accumulating.
  56.             } else if (active == leastActive) {
  57.                 // 当前invoker的活跃数如果和minActive相等，在leastIndexes记录当前下标，leastCount+1
  58.                 leastIndexes[leastCount++] = i;
  59.                 // 更新总权重
  60.                 totalWeight += afterWarmup;
  61.                 // 判断这个具有相同active的invoker，和之前记录的invoker权重是否相等
  62.                 if (sameWeight && afterWarmup != firstWeight) {
  63.                     sameWeight = false;
  64.                 }
  65.             }
  66.         }
  67.         // 最小活跃数的invoker为1，那么直接取leastIndexes第一个
  68.         if (leastCount == 1) {
  69.             // If we got exactly one invoker having the least active value, return this invoker directly.
  70.             return invokers.get(leastIndexes[0]);
  71.         }
  72.           // 走到这里肯定是leastCount>1
  73.           // 加权随机负载均衡的实现
  74.         if (!sameWeight && totalWeight > 0) {
  75.             int offsetWeight = ThreadLocalRandom.current().nextInt(totalWeight);
  76.             for (int i = 0; i < leastCount; i++) {
  77.                 int leastIndex = leastIndexes[i];
  78.                 offsetWeight -= weights[leastIndex];
  79.                 if (offsetWeight < 0) {
  80.                     return invokers.get(leastIndex);
  81.                 }
  82.             }
  83.         }
  84.         // 两个及以上权重相同或总权重为0，随机返回一个invoker
  85.         return invokers.get(leastIndexes[ThreadLocalRandom.current().nextInt(leastCount)]);
  86.     }
  87. }

想要模拟最小活跃数，先把生产者里三个服务提供者修改修改不同的权重，provider01、provider02、provider03权重分别设置5、10、15，为了让多个消费者都有一些调用次数，生产者所有接口响应都进行休眠10分钟，并修改接口超时时间，其中以provider03代码示例如下

  1. /**
  2.  * 功能描述:
  3.  *
  4.  * @param:
  5.  * @return:
  6.  * @auther: Zywoo Lee
  7.  * @date: 2021/11/20 3:02 下午
  8.  */
  9. @DubboService(weight = 15,timeout = 700000, loadbalance = "leastactive")
  10. public class DemoServiceImpl implements DemoService {
  12.     private final Logger logger = LoggerFactory.getLogger(getClass());
  14.     @Value("${dubbo.application.name}")
  15.     private String serviceName;
  17.     @Override
  18.     public String sayHello(String name) {
  19.         System.out.println("权重为15的服务器接受到请求了:" + name);
  20.         try {
  21.             TimeUnit.MINUTES.sleep(10);
  22.         } catch (InterruptedException e) {
  23.             e.printStackTrace();
  24.         }
  25.         return String.format("[%s] : Hello, %s", serviceName, name);
  26.     }
  28. }

然后服务消费者进行改造，一共发送21个请求，前20请求分别落在三台服务提供者上，然后最后debug第21个线程，代码如下：

  1. /**
  2.  * @Auther: Zywoo Lee
  3.  * @Date: 2021/11/20 15:35
  4.  * @Description:
  5.  */
  6. @RestController
  7. @RequestMapping("/test")
  8. @Api(tags = "Dubbo服务测试类")
  9. public class DemoController {
  11.         @DubboReference(loadbalance = "leastactive",filter = "activelimit")
  12.     private DemoService demoService;
  14.     @ApiOperation(value = "服务调用测试", notes = "")
  15.     @GetMapping("sayHello")
  16.     public void sayHello() {
  17.         for (int i = 0; i < 20; i++) {
  18.             int finalI = i;
  19.             new Thread(() -> {
  20.                 demoService.sayHello("ZywooLee-" + finalI);
  21.             }).start();
  22.         }
  23.         try {
  24.             TimeUnit.SECONDS.sleep(10);
  25.         } catch (InterruptedException e) {
  26.             e.printStackTrace();
  27.         }
  28.         demoService.sayHello("ZywooLee-debug");
  29.     }
  32. }

active为0的情况

请注意：@DubboReference一定要配置filter = “activelimit”，不然在负载均衡的doSelect()方法会出现active为0的情况。原因在于客户端没有配置ActiveLimitFilter

  1. int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive();

测试结果

image.png

image.png

image.png

然后开始打断点，进入第21个请求。最终可以看到请求落向选择最少连接数的提供者

image.png

一致性哈希负载均衡

  1. org.apache.dubbo.rpc.cluster.loadbalance.ConsistentHashLoadBalance

因为原代码debug查看hash环生成不方便，所以copy源码加入一些输出打印，并继承AbstractLoadBalance，然后指定自定义负载均衡

  2. package com.dubbo.loadbalance;
  4. import com.alibaba.dubbo.rpc.support.RpcUtils;
  5. import org.apache.dubbo.common.URL;
  6. import org.apache.dubbo.rpc.Invocation;
  7. import org.apache.dubbo.rpc.Invoker;
  8. import org.apache.dubbo.rpc.cluster.loadbalance.AbstractLoadBalance;
  9. import org.springframework.stereotype.Component;
  11. import java.nio.charset.StandardCharsets;
  12. import java.security.MessageDigest;
  13. import java.security.NoSuchAlgorithmException;
  14. import java.util.List;
  15. import java.util.Map;
  16. import java.util.TreeMap;
  17. import java.util.concurrent.ConcurrentHashMap;
  18. import java.util.concurrent.ConcurrentMap;
  20. import static org.apache.dubbo.common.constants.CommonConstants.COMMA_SPLIT_PATTERN;
  22. /**
  23.  * @Auther: Zywoo Lee
  24.  * @Date: 2021/11/20 18:00
  25.  * @Description:
  26.  */
  27. public class ZywooLeeConsistentHashLoadBalance extends AbstractLoadBalance {
  29.     public static final String NAME = "consistenthash";
  31.     /**
  32.      * Hash nodes name
  33.      */
  34.     public static final String HASH_NODES = "hash.nodes";
  36.     /**
  37.      * Hash arguments name
  38.      */
  39.     public static final String HASH_ARGUMENTS = "hash.arguments";
  41.     private final ConcurrentMap<String, ConsistentHashSelector<?>> selectors = new ConcurrentHashMap<String, ConsistentHashSelector<?>>();
  43.     @SuppressWarnings("unchecked")
  44.     @Override
  45.     protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
  46.         String methodName = RpcUtils.getMethodName(invocation);
  47.         String key = invokers.get(0).getUrl().getServiceKey() + "." + methodName;
  49.         System.out.println("selectors中获取value的key="+key);
  51.         // 获取invokers的hashCode
  52.         int invokersHashCode = invokers.hashCode();
  53.         ConsistentHashSelector<T> selector = (ConsistentHashSelector<T>) selectors.get(key);
  54.         // 如果invokers是一个新的List对象，意味着服务提供者数量发生变化，可能新增也可能减少
  55.         // 此时selector.identityHashCode != invokersHashCode条件成立
  56.         // 如果第一次调用 selector == null成立
  57.         if (selector == null || selector.identityHashCode != invokersHashCode) {
  59.             System.out.println("新的invokers:"+ invokersHashCode+",原"+(selector==null ? "null" : selector.identityHashCode));
  62.             selectors.put(key, new ConsistentHashSelector<T>(invokers, methodName, invokersHashCode));
  63.             selector = (ConsistentHashSelector<T>) selectors.get(key);
  65.             System.out.println("--哈希环构建完成--");
  66.             for (Map.Entry<Long, Invoker<T>> entry : selector.virtualInvokers.entrySet()) {
  67.                 System.out.println("key(哈希值)="+entry.getKey()+", value(虚拟节点)="+entry.getValue());
  68.             }
  70.         }
  71.         System.out.println("---执行select方法选择invoker---");
  73.         // 创建新的ConsistentHashSelector的select方法选择invoker
  74.         return selector.select(invocation);
  75.     }
  77.     private static final class ConsistentHashSelector<T> {
  78.         // 使用TreeMap存储Invoker的虚拟节点
  79.         private final TreeMap<Long, Invoker<T>> virtualInvokers;
  80.         // 虚拟节点数
  81.         private final int replicaNumber;
  82.         // hashcode
  83.         private final int identityHashCode;
  84.         // 请求中的参数下标
  85.         // 需要对请求中对应的下标的参数进行哈希计算
  86.         private final int[] argumentIndex;
  88.         ConsistentHashSelector(List<Invoker<T>> invokers, String methodName, int identityHashCode) {
  89.             this.virtualInvokers = new TreeMap<Long, Invoker<T>>();
  90.             this.identityHashCode = identityHashCode;
  91.             URL url = invokers.get(0).getUrl();
  92.             // 即时启动多个invoker，每个invoker对应的url上的虚拟节点数配置的都是一样的
  93.             // 这里默认160个，可以在服务提供者配置parameters->hash.nodes,我这边设置为4个便于debug
  94.             this.replicaNumber = url.getMethodParameter(methodName, HASH_NODES, 160);
  95.             // 获取参与哈希计算的参数下标值，默认对第一个参数进行哈希计算
  96.             String[] index = COMMA_SPLIT_PATTERN.split(url.getMethodParameter(methodName, HASH_ARGUMENTS, "0"));
  97.             argumentIndex = new int[index.length];
  98.             // for循环对argumentIndex赋值
  99.             for (int i = 0; i < index.length; i++) {
  100.                 argumentIndex[i] = Integer.parseInt(index[i]);
  101.             }
  102.             // 遍历服务提供者
  103.             for (Invoker<T> invoker : invokers) {
  104.                 // 获取每个invoker的地址
  105.                 String address = invoker.getUrl().getAddress();
  106.                 for (int i = 0; i < replicaNumber / 4; i++) {
  107.                     // adress+i进行md5运算得到一个长度为16的字节数组
  108.                     byte[] digest = md5(address + i);
  109.                     // 对digest部分字节进行4次hash运算得到四个不同的long型正整数
  110.                     for (int h = 0; h < 4; h++) {
  111.                         // h==0:digest中下标0~3的4个字节进行位运算
  112.                         // h==1:digest中下标4~7的4个字节进行位运算
  113.                         // h==2:digest中下标8~11的4个字节进行位运算
  114.                         // h==3:digest中下标12~15的4个字节进行位运算
  115.                         long m = hash(digest, h);
  116.                         // 将hash到invoker的映射关系存储到virtualInvokers中
  117.                         // virtualInvokers需要提供高效查询操作，使用TreeMap数据结构
  118.                         virtualInvokers.put(m, invoker);
  119.                     }
  120.                 }
  121.             }
  122.         }
  124.         public Invoker<T> select(Invocation invocation) {
  125.             String key = toKey(invocation.getArguments());
  126.             byte[] digest = md5(key);
  127.             // 取digest数组的前四个字节进行hash运算，再将hash值传给selectForKey（）
  128.             // 寻找合适的invoker
  130.             long hash = hash(digest, 0);
  131.             System.out.println("参与hash计算的key:"+"经过hash计算后hash="+hash);
  132.             return selectForKey(hash);
  133.         }
  135.         // argumentIndex转换成key
  136.         private String toKey(Object[] args) {
  137.             StringBuilder buf = new StringBuilder();
  138.             for (int i : argumentIndex) {
  139.                 if (i >= 0 && i < args.length) {
  140.                     buf.append(args[i]);
  141.                 }
  142.             }
  143.             return buf.toString();
  144.         }
  146.         private Invoker<T> selectForKey(long hash) {
  147.             //到TreeMap中找到第一个节点值大于或等于当前hash的Invoker
  148.             Map.Entry<Long, Invoker<T>> entry = virtualInvokers.ceilingEntry(hash);
  149.             //如果hash大于Invoker在圆环上最大位置，此时entry=null
  150.             if (entry == null) {
  151.                 // 将TreeMap头节点赋值给entry
  152.                 entry = virtualInvokers.firstEntry();
  153.             }
  154.             System.out.println("根据key计算出hash="+hash+",获取出来的invoker是:"+entry.getValue());
  155.             return entry.getValue();
  156.         }
  158.         private long hash(byte[] digest, int number) {
  159.             return (((long) (digest[3 + number * 4] & 0xFF) << 24)
  160.                     | ((long) (digest[2 + number * 4] & 0xFF) << 16)
  161.                     | ((long) (digest[1 + number * 4] & 0xFF) << 8)
  162.                     | (digest[number * 4] & 0xFF))
  163.                     & 0xFFFFFFFFL;
  164.         }
  166.         private byte[] md5(String value) {
  167.             MessageDigest md5;
  168.             try {
  169.                 md5 = MessageDigest.getInstance("MD5");
  170.             } catch (NoSuchAlgorithmException e) {
  171.                 throw new IllegalStateException(e.getMessage(), e);
  172.             }
  173.             md5.reset();
  174.             byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
  175.             md5.update(bytes);
  176.             return md5.digest();
  177.         }
  179.     }
  181. }

测试结果

服务提供者代码如下：
parameters如果不指定的话，默认为160个，在源码里可以看到

  1. @DubboService(timeout = 700000, loadbalance = "zywooLeeConsistentHash",parameters={"hash.nodes","4"})
  2. public class DemoServiceImpl implements DemoService {
  4.     private final Logger logger = LoggerFactory.getLogger(getClass());
  6.     @Value("${dubbo.application.name}")
  7.     private String serviceName;
  9.     @Override
  10.     public String sayHello(String name) {
  11.         return String.format("[%s] : Hello, %s", serviceName, name);
  12.     }
  14. }

服务消费者代码如下：

  1. /**
  2.  * @Auther: Zywoo Lee
  3.  * @Date: 2021/11/20 15:35
  4.  * @Description:
  5.  */
  6. @RestController
  7. @RequestMapping("/test")
  8. @Api(tags = "Dubbo服务测试类")
  9. public class DemoController {
  11.     @DubboReference(loadbalance = "zywooLeeConsistentHash",filter = "activelimit")
  12.     private DemoService demoService;
  14.     @ApiOperation(value = "服务调用测试", notes = "")
  15.     @GetMapping("sayHello")
  16.     public void sayHello() {
  17.         String name = demoService.sayHello("ZywooLee-debug");
  18.         System.out.println(name);
  19.     }
  22. }

首先先启动两个服务提供者，调用方法控制台输出如下：

image.png

服务提供者数量不变，第二次调用，可以看到selector不为null，invokers没有变化，自然selector.identityHashCode != invokersHashCode为false，直接进入selector.select(invocation)逻辑。

image.png

第二次调用控制台输出如下：

image.png

然后开启第三个服务提供者，此时invokers自然发生了变化，再次发起服务请求，控制台打印如下：

image.png