中间件（二）dubbo负载均衡介绍

一、负载均衡概述

支持轮询、随机、一致性hash和最小活跃数等。

1、轮询

① sequences：内部的序列计数器
② 服务器接口方法权重一样：（sequences+1）%服务器的数量=（决定调用）哪个服务器的服务。
③ 服务器接口方法权重不一样：找到最大权重（权重数）%（sequences+1），然后找出权重比该取模后的值大服务器列表，最后进行【①】所述。

2、随机

统计服务器上该接口的方法权重总和，然后对这个总和随机nextInt一下，看生成的随机数落到哪个段内，就调用哪个服务器上的该服务。

3、一致性hash

保证了同样的请求（参数）将会落到同一台服务器上。

4、最小活跃数

每个接口和接口方法都对应一个RpcStatus，记录它们的活跃数、失败等相关统计信息，调用时活跃数+1，调用结束时活跃数-1，所以活跃值越大，表明该提供者服务器的该接口方法耗时越长，而消费能力强的提供者接口往往活跃值很低。最少活跃负载均衡保证了“慢”提供者能接收到更少的服务器调用。

二、负载均衡策略配置

1、多注册中心集群负载均衡

2、Cluster Invoker

支持的选址策略如下（dubbo2.7.5+ 版本，具体使用请参见文档）

2-1、指定优先级

<!-- 来自 preferred=“true” 注册中心的地址将被优先选择，
只有该中心无可用地址时才 Fallback 到其他注册中心 -->
<dubbo:registry address="zookeeper://${zookeeper.address1}" preferred="true" />

2-2、同zone优先

<!-- 选址时会和流量中的 zone key 做匹配，流量会优先派发到相同 zone 的地址 -->
<dubbo:registry address="zookeeper://${zookeeper.address1}" zone="beijing" />

2-3、权重轮询

<!-- 来自北京和上海集群的地址，将以 10:1 的比例来分配流量 -->
<dubbo:registry id="beijing" address="zookeeper://${zookeeper.address1}" weight=”100“ />
<dubbo:registry id="shanghai" address="zookeeper://${zookeeper.address2}" weight=”10“ />

三、负载均衡解读

（1）负载均衡：AbstractClusterInvoker.invoke(final Invocation invocation)方法

@Override
public Result invoke(final Invocation invocation) throws RpcException {//...... 省略代码 List<Invoker<T>> invokers = list(invocation);LoadBalance loadbalance = initLoadBalance(invokers, invocation);RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);return doInvoke(invocation, invokers, loadbalance);
}
/** 获取负载均衡的实现方法，未配置时默认random */
protected LoadBalance initLoadBalance(List<Invoker<T>> invokers, Invocation invocation) {if (CollectionUtils.isNotEmpty(invokers)) {return ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl().getMethodParameter(RpcUtils.getMethodName(invocation), LOADBALANCE_KEY, DEFAULT_LOADBALANCE));} else {return ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(DEFAULT_LOADBALANCE);}
}

（2）实现入口：AbstractClusterInvoker.doSelect(…)

① 在dubbo中，所有负载均衡均继承 AbstractLoadBalance 类，该类实现了LoadBalance接口，并封装了一些公共的逻辑。

/** 1. LoadBalance.java接口 */
@SPI(RandomLoadBalance.NAME)
public interface LoadBalance {/*** select one invoker in list.** @param invokers   invokers.* @param url        refer url* @param invocation invocation.* @return selected invoker.*/@Adaptive("loadbalance")<T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException;
}
/** 2. AbstractLoadBalance.java 抽象类 */
@Override
public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) {if (CollectionUtils.isEmpty(invokers)) {return null;}// 如果invokers列表中仅一个Invoker，直接返回即可，无需进行负载均衡if (invokers.size() == 1) {return invokers.get(0);}// 调用doSelect方法进行负载均衡，该方法为抽象方法，由子类实现return doSelect(invokers, url, invocation);
}protected abstract <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation);

/** 2-1. 公共方法，权重计算逻辑 */

protected int getWeight(Invoker<?> invoker, Invocation invocation) {int weight;URL url = invoker.getUrl();// Multiple registry scenario, load balance among multiple registries.if (REGISTRY_SERVICE_REFERENCE_PATH.equals(url.getServiceInterface())) {weight = url.getParameter(REGISTRY_KEY + "." + WEIGHT_KEY, DEFAULT_WEIGHT);} else {// 从url中获取权重 weight配置值weight = url.getMethodParameter(invocation.getMethodName(), WEIGHT_KEY, DEFAULT_WEIGHT);if (weight > 0) {// 获取服务提供者启动时间戳long timestamp = invoker.getUrl().getParameter(TIMESTAMP_KEY, 0L);if (timestamp > 0L) {// 计算服务提供者运行时长long uptime = System.currentTimeMillis() - timestamp;if (uptime < 0) {return 1; // 未启动直接返回权重为1}// 获取服务预热时间，默认为10分钟int warmup = invoker.getUrl().getParameter(WARMUP_KEY, DEFAULT_WARMUP);// 如果服务运行时间小于预热时间，则重新计算服务权重，即降级if (uptime > 0 && uptime < warmup) {weight = calculateWarmupWeight((int)uptime, warmup, weight);}}}}return Math.max(weight, 0);
}

// 2-2. 重新计算服务权重

static int calculateWarmupWeight(int uptime, int warmup, int weight) {// 计算权重，下面代码逻辑上形似于 (uptime / warmup) * weight// 随着服务运行时间 uptime 增大，权重计算值 ww 会慢慢接近配置值 weightint ww = (int) ( uptime / ((float) warmup / weight));return ww < 1 ? 1 : (Math.min(ww, weight));
}

注：
select 方法的逻辑比较简单，首先会检测 invokers 集合的合法性，然后再检测 invokers 集合元素数量。如果只包含一个 Invoker，直接返回该 Inovker 即可。如果包含多个 Invoker，此时需要通过负载均衡算法选择一个 Invoker。具体的负载均衡算法由子类实现。
权重的计算主要用于保证当服务运行时长小于服务预热时间时，对服务进行降权，避免让服务在启动之初就处于高负载状态。服务预热是一个优化手段，与此类似的还有 JVM 预热。主要目的是让服务启动后“低功率”运行一段时间，使其效率慢慢提升至最佳状态。
配置方式（默认100）：dubbo.provider.weight=300dubbo.provider.weight=300

② RandomLoadBalance 加权随机负载均衡
算法思路：根据权重比，随机选择哪台服务器，如：servers=[A,B,C]，weights = [5, 3, 2]，权重和为10，调用A的次数约有50%，B有30%，C有20%。

@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {// Number of invokersint length = invokers.size();// 判断是否需要权重负载均衡if (!needWeightLoadBalance(invokers,invocation)){return invokers.get(ThreadLocalRandom.current().nextInt(length));}// Every invoker has the same weight?boolean sameWeight = true;// the maxWeight of every invokers, the minWeight = 0 or the maxWeight of the last invokerint[] weights = new int[length];// The sum of weightsint totalWeight = 0;// ① 计算总权重，totalWeight；② 检测每个服务提供者的权重是否相同for (int i = 0; i < length; i++) {int weight = getWeight(invokers.get(i), invocation);// SumtotalWeight += weight;// save for later use// 如果权重分别为5,3,2，则weights[0]=5，weights[1]=8，weights[2]=10weights[i] = totalWeight;// 判断每个服务权重是否相同，如果不相同则sameWeight置为falseif (sameWeight && totalWeight != weight * (i + 1)) {sameWeight = false;}}// 各提供者服务权重不一样时，计算随机数落在哪个区间上if (totalWeight > 0 && !sameWeight) {// If (not every invoker has the same weight & at least one invoker's weight>0), select randomly based on totalWeight.// 随机获取一个[0, totalWeight]区间内的随机的数字int offset = ThreadLocalRandom.current().nextInt(totalWeight);// Return a invoker based on the random value.for (int i = 0; i < length; i++) {// weights[0]=5，offset=[0, 5)// weights[1]=8，offset=[5, 8)// weights[2]=10，offset=[8, 10)if (offset < weights[i]) {return invokers.get(i);}}}// If all invokers have the same weight value or totalWeight=0, return evenly.// 如果所有服务提供者权重值相同，此时直接随机返回一个即可return invokers.get(ThreadLocalRandom.current().nextInt(length));
}

注：RandomLoadBalance的算法比较简单，多次请求后，能够按照权重进行“均匀“分配。调用次数少时，可能产生的随机数比较集中，此缺点并不严重，可以忽略。它是一个高效的负载均衡实现，因此Dubbo选择它作为缺省实现。

③ LeastActiveLoadBalance 加权最小活跃数负载均衡
活跃调用数越小，表明该服务提供者效率越高，单位时间内可处理更多的请求。此时应优先将请求分配给该服务提供者。
在具体实现中，每个服务提供者对应一个活跃数 active。初始情况下，所有服务提供者活跃数均为0。每收到一个请求，活跃数加1，完成请求后则将活跃数减1。
除了最小活跃数，LeastActiveLoadBalance 在实现上还引入了权重值。所以准确的来说，LeastActiveLoadBalance 是基于加权最小活跃数算法实现的。

@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {// Number of invokersint length = invokers.size();// The least active value of all invokers// 最小活跃数int leastActive = -1;// The number of invokers having the same least active value (leastActive)// 具有相同“最小活跃数”的服务提供者int leastCount = 0;// The index of invokers having the same least active value (leastActive)// leastIndexes 用于记录具有相同“最小活跃数”的 Invoker 在 invokers 列表中的下标信息int[] leastIndexes = new int[length];// the weight of every invokersint[] weights = new int[length];// The sum of the warmup weights of all the least active invokersint totalWeight = 0;// The weight of the first least active invoker// 第一个最小活跃数的 Invoker 权重值，用于与其他具有相同最小活跃数的 Invoker 的权重进行对比，// 以检测是否“所有具有相同最小活跃数的 Invoker 的权重”均相等int firstWeight = 0;// Every least active invoker has the same weight value?// 表示各服务的权重是否相同boolean sameWeight = true;// Filter out all the least active invokersfor (int i = 0; i < length; i++) {Invoker<T> invoker = invokers.get(i);// Get the active number of the invoker// 获取invoker对应的活跃数int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive();// Get the weight of the invoker's configuration. The default value is 100.// 获取权重int afterWarmup = getWeight(invoker, invocation);// save for later useweights[i] = afterWarmup;// If it is the first invoker or the active number of the invoker is less than the current least active number// 发现更小的活跃数，重新开始if (leastActive == -1 || active < leastActive) {// Reset the active number of the current invoker to the least active number// 使用当前活跃数 active 更新最小活跃数 leastActiveleastActive = active;// Reset the number of least active invokersleastCount = 1;// Put the first least active invoker first in leastIndexes// 记录当前下标值到 leastIndexes 中leastIndexes[0] = i;// Reset totalWeighttotalWeight = afterWarmup;// Record the weight the first least active invokerfirstWeight = afterWarmup;// Each invoke has the same weight (only one invoker here)sameWeight = true;// If current invoker's active value equals with leaseActive, then accumulating.// 当前 Invoker 的活跃数 active 与最小活跃数 leastActive 相同} else if (active == leastActive) {// Record the index of the least active invoker in leastIndexes orderleastIndexes[leastCount++] = i;// Accumulate the total weight of the least active invoker// 累加权重totalWeight += afterWarmup;// If every invoker has the same weight?// 检测当前 Invoker 的权重与 firstWeight 是否相等，不相等则将 sameWeight 置为 falseif (sameWeight && afterWarmup != firstWeight) {sameWeight = false;}}}// Choose an invoker from all the least active invokers// 1. 当只有一个 Invoker 具有最小活跃数，此时直接返回该 Invoker 即可if (leastCount == 1) {// If we got exactly one invoker having the least active value, return this invoker directly.return invokers.get(leastIndexes[0]);}// 2. 有多个 Invoker 具有相同的最小活跃数，但它们之间的权重不同if (!sameWeight && totalWeight > 0) {// If (not every invoker has the same weight & at least one invoker's weight>0), select randomly based on // totalWeight.// 随机生成一个 [0, totalWeight) 之间的数字int offsetWeight = ThreadLocalRandom.current().nextInt(totalWeight);// Return a invoker based on the random value.for (int i = 0; i < leastCount; i++) {// 获取权重数组的下标int leastIndex = leastIndexes[i];// 随机权重 - 具有最小活跃数的 Invoker 的权重值offsetWeight -= weights[leastIndex];if (offsetWeight < 0) { // 与RandomLoadBalance一致，权重越大调用的次数越多return invokers.get(leastIndex);}}}// If all invokers have the same weight value or totalWeight=0, return evenly.// 如果权重相同或权重为0时，随机返回一个 Invokerreturn invokers.get(leastIndexes[ThreadLocalRandom.current().nextInt(leastCount)]);
}

④ ConsistentHashLoadBalance
cache-1、cache-2、cache-3、cache-4分别为不同的节点
根据IP或者其他信息为缓存节点生成一个hash，并将这个hash投射到[0,2^32 - 1] 的圆环上。当有查询或写入请求时，则为缓存项的key生成一个hash值。然后查找第一个大于或等于该hash值的缓存节点，并到这个节点中查询或写入缓存项。
如果当前节点挂了，则在下一次查询或写入缓存时，为缓存项查找另一个大于或其hash值的缓存节点即可。
如下图，每个节点在圆环上占据一个位置，如果缓存项的key的hash值小于缓存节点hash值，则到该缓存节点中存储或读取缓存项。
比如下面绿色点对应的缓存项将会被存储到 cache-2 节点中。由于 cache-3 挂了，原本应该存到该节点中的缓存项最终会存储到 cache-4 节点中。

⑤ RoundRobinLoadBalance
轮询：是指将请求轮流分配给每台服务器。
例如：有三台服务器A、B、C，我们将第一个请求分配给A服务器，第二个请求分配给B服务器，第三个请求分配给C服务器，第四个请求再次分配给A服务器，如此循环，这个过程叫做轮询（轮询是一种无状态负载均衡算法）。适用于每台服务器性能相近的场景下。
轮询加权：对每台性能不一样的服务器进行加权处理，如服务器A、B、C的权重分别为5:2:1时，在8次请求中，服务器A将收到5次请求、B收到2次请求、C收到一次请求（请求次数越多，每台服务器得到的请求数，接近服务器的权重比）。

@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {// key = [group/]path[:version].methodName；注：path = com.jyt.*.service.接口名String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();// 获取key对应值，如果key的值不存在，则将第二个参数的返回值存入并返回ConcurrentMap<String, WeightedRoundRobin> map = methodWeightMap.computeIfAbsent(key, k -> new ConcurrentHashMap<>());int totalWeight = 0;long maxCurrent = Long.MIN_VALUE;long now = System.currentTimeMillis();Invoker<T> selectedInvoker = null;WeightedRoundRobin selectedWRR = null;// 遍历服务提供者for (Invoker<T> invoker : invokers) {// dubbo://11.1.1.109:21001/com.jyt.*.service.类名String identifyString = invoker.getUrl().toIdentityString();// 获取当前服务提供者的权重int weight = getWeight(invoker, invocation);// 根据identifyString获取当前提供者对应的权重，如果不存在则使用第二个参数返回值，并返回WeightedRoundRobin weightedRoundRobin = map.computeIfAbsent(identifyString, k -> {WeightedRoundRobin wrr = new WeightedRoundRobin();wrr.setWeight(weight);return wrr;});// 如果提供者的权重被修改了，则更新weightedRoundRobin的权重值if (weight != weightedRoundRobin.getWeight()) {// weight changedweightedRoundRobin.setWeight(weight);}// current加上weight并获取结果（初始的current为0）long cur = weightedRoundRobin.increaseCurrent();/*** 如果A服务权重weight=500，B权重weight=100时，totalWeight = 600，初始cur=0；服务调用场景* 第一次：A服务器：cur=weight + curA = 500，cur > maxCurrent，所以maxCurrent = curA = 500*        B服务器：cur=weight + curB = 100 <= maxCurrent（500为true）；故走A服务器，即curA = curA - 600 = -100** 第二次：A服务器：cur=weight + curA = 400，cur > maxCurrent，所以maxCurrent = curA = 400*        B服务器：cur=weight + curB = 200 <= maxCurrent（400为true）；故走A服务器，即curA = curA - 600 = -200** 第三次：A服务器：cur=weight + curA = 300，cur > maxCurrent，所以maxCurrent = curA = 300*        B服务器：cur=weight + curB = 300 <= maxCurrent（300为true）；故走A服务器，即curA = curA - 600 = -300** 第四次：A服务器：cur=weight + curA = 200，cur > maxCurrent，所以maxCurrent = curA = 200*        B服务器：cur=weight + curB = 400 <= maxCurrent（200为false）；故走B服务器，即curB = curB - 600 = -200** 第五次：A服务器：cur=weight + curA = 700，cur > maxCurrent，所以maxCurrent = curA = 700*        B服务器：cur=weight + curB = -100 <= maxCurrent（700为true）；故走A服务器，即curA = curA - 600 = 100** 第六次：A服务器：cur=weight + curA = 600，cur > maxCurrent，所以maxCurrent = curA = 600*        B服务器：cur=weight + curB = 0 <= maxCurrent（600为true）；故走A服务器，即curA = curA - 600 = 0*        * ... ... 如此循环：A、A、A、B、A、A*/weightedRoundRobin.setLastUpdate(now);// 判断是否比最大的值大if (cur > maxCurrent) {// 如果大，则将当前服务提供者置为本次服务提供者maxCurrent = cur;selectedInvoker = invoker;selectedWRR = weightedRoundRobin;}// 权重累计totalWeight += weight;}// 当两者大小不一致时，map中可能会存在一些已经下线的服务，本次剔除一些很久节点信息if (invokers.size() != map.size()) {map.entrySet().removeIf(item -> now - item.getValue().getLastUpdate() > RECYCLE_PERIOD);}// 如果存在选择好的提供者，则改变他的current值 - totalWeight;if (selectedInvoker != null) {selectedWRR.sel(totalWeight);return selectedInvoker;}// should not happen herereturn invokers.get(0);
}