（给ImportNew加星标，提高Java技能）
ytao.top/2020/05/02/23-dubbo-loadbalance/
前言
负载均衡是指在集群中，将多个数据请求分散在不同单元上进行执行，主要为了提高系统容错能力和加强系统对数据的处理能力。
在 Dubbo 中，一次服务的调用就是对所有实体域 Invoker 的一次筛选过滤，最终选定具体调用的 Invoker。首先在 Directory 中获取全部 Invoker 列表，通过路由筛选出符合规则的 Invoker，最后再经过负载均衡选出具体的 Invoker。所以 Dubbo 负载均衡机制是决定一次服务调用使用哪个提供者的服务。
整体结构
Dubbo 负载均衡的分析入口是
org.apache.dubbo.rpc.cluster.loadbalance.AbstractLoadBalance 抽象类，查看这个类继承关系。
这个被 RandomLoadBalance、LeastActiveLoadBalance、RoundRobinLoadBalanc
e 及 ConsistentHashLoadBalance 类继承，这四个类是 Dubbo 中提供的四种负载均衡算法的实现。
以上则是 Dubbo 提供的四种负载均衡算法。
从上图中，看到 AbstractLoadBalance 实现了 LoadBalance 接口，同时是一个 SPI 接口，指定默认实现为 RandomLoadBalance 随机算法机制。
抽象类 AbstractLoadBalance 中，实现了负载均衡通用的逻辑，同时给子类声明了一个抽象方法供子类实现其负载均衡的逻辑。
public abstract class AbstractLoadBalance implements LoadBalance {/** * * @param 运行时间（毫秒） * @param 预热时间（毫秒） * @param 要计算的 Invoker 权重值 */static int calculateWarmupWeight(int uptime, int warmup, int weight) {// 计算预热时期的权重int ww = (int) ((float) uptime / ((float) warmup / (float) weight));// 返回的权重值区间在: 1 ~ weightreturn ww < 1 ? 1 : (ww > weight ? weight : ww); }@Overridepublic <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) {// 校验 invokers 是否为空if (CollectionUtils.isEmpty(invokers)) {return null; }// 当到达负载均衡流程时，invokers 中只有一个 Invoker 时，直接返回该 Invokerif (invokers.size() == 1) {return invokers.get(0); }// 在不同负载均衡策略中完成具体的实现return doSelect(invokers, url, invocation); }// 声明抽象方法，在子类中具体实现protected abstract <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation);protected int getWeight(Invoker<?> invoker, Invocation invocation) {// 获取当前Invoker配置的权重值int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), WEIGHT_KEY, DEFAULT_WEIGHT);if (weight > 0) {// 服务启动时间long timestamp = invoker.getUrl().getParameter(REMOTE_TIMESTAMP_KEY, 0L);if (timestamp > 0L) {// 服务已运行时长int uptime = (int) (System.currentTimeMillis() – timestamp);// 服务预热时间，默认 DEFAULT_WARMUP = 10 * 60 * 1000 ，预热十分钟int warmup = invoker.getUrl().getParameter(WARMUP_KEY, DEFAULT_WARMUP);// 如果服务运行时长小于预热时长，重新计算出预热时期的权重if (uptime > 0 && uptime < warmup) { weight = calculateWarmupWeight(uptime, warmup, weight); } } }// 保证最后返回的权重值不小于0return weight >= 0 ? weight : 0; }}
??
在 AbstractLoadBalance 中，getWeight 和 calculateWarmupWeight 方法是获取和计算当前 Invoker 的权重值。
getWeight 中获取当前权重值，通过 URL 获取当前 Invoker 设置的权重，如果当前服务提供者启动时间小于预热时间，则会重新计算权重值，对服务进行降权处理，保证服务能在启动初期不分发设置比例的全部流量，健康运行下去。
calculateWarmupWeight 是重新计算权重值的方法，计算公式为：服务运行时长/(预热时长/设置的权重值)，等价于(服务运行时长/预热时长)*设置的权重值，同时条件服务运行时长<预热时长。由该公式可知，预热时长和设置的权重值不变，服务运行时间越长，计算出的值越接近 weight，但不会等于 weight。在返回计算后的权重结果中，对小于1和大于设置的权重值进行了处理，当重新计算后的权重小于1时返回1；处于1和设置的权重值之间时，直接返回计算后的结果；当权重大于设置的权重值时（因为条件限制，不会出现该类情况），返回设置的权重值。所以得出结论：重新计算后的权重值为 1 ~ 设置的权重值，运行时间越长，计算出的权重值越接近设置的权重值。
配置方式
服务端
通过 XML 配置方式：
<!– 服务级别配置 –><dubbo:service id=”xXXXService” interface=”top.ytao.service.XXXXService” class=”top.ytao.service.impl.XXXXServiceImpl” loadbalance=”负载策略” /><!– 方法级别配置 –><dubbo:service id=”xXXXService” interface=”top.ytao.service.XXXXService” class=”top.ytao.service.impl.XXXXServiceImpl”><dubbo:method name=”方法名” loadbalance=”负载策略”/></dubbo:service>
通过 Properties 配置：dubbo.service.loadbalance=负载策略
通过注解方式：
@Service(loadbalance =”负载策略”)
客户端通过 XML 配置方式：<!– 服务级别配置 –><dubbo:reference id=”xXXXService” interface=”top.ytao.service.XXXXService” loadbalance=”负载策略” /><!– 方法级别配置 –><dubbo:reference id=”xXXXService” interface=”top.ytao.service.XXXXService”><dubbo:method name=”方法名” loadbalance=”负载策略”/></dubbo:reference>
??
通过 Properties 配置：
dubbo.reference.loadbalance=负载策略
通过注解配置方式：
@Reference(loadbalance = “负载策略”)
实现方式也可通过 Dubbo-Admin 管理后台进行配置，如图：
随机算法
加权随机算法负载均衡策略（RandomLoadBalance）是 dubbo 负载均衡的默认实现方式，根据权重分配各个 Invoker 随机选中的比例。这里的意思是：将到达负载均衡流程的 Invoker 列表中的 权重进行求和，然后求出单个 Invoker 权重在总权重中的占比，随机数就在总权重值的范围内生成。
如图，假如当前有 192.168.1.10和 192.168.1.11两个负载均衡的服务，权重分别为 4、6 ，则它们的被选中的比例为 2/5、3/5。
当生成随机数为 6 时，就会选中 192.168.1.11的服务。
dubbo 中 RandomLoadBalance 的 doSelect 实现代码：
public class RandomLoadBalance extends AbstractLoadBalance{public static final String NAME = “random”;@Override protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation){// Invoker 数量int length = invokers.size();// 标识所有 Invoker 的权重是否都一样boolean sameWeight = true;// 用一个数组保存每个 Invoker 的权重int[] weights = new int[length]// 第一个 Invoker 的权重int firstWeight = getWeight(invokers.get(0), invocation); weights[0] = firstWeight;// 求和总权重int totalWeight = firstWeight;for (int i = 1; i < length; i ) {int weight = getWeight(invokers.get(i), invocation);// 保存每个 Invoker 的权重到数组总 weights[i] = weight;// 累加求和总权重 totalWeight = weight;// 如果不是所有 Invoker 的权重都一样，就给标记上 sameWeight = falseif (sameWeight && weight != firstWeight) { sameWeight = false; } }// 计算随机数取到的 Invoker，条件是必须总权重大于0，并且每个 Invoker 的权重都不一样if (totalWeight > 0 && !sameWeight) {// 基于 0~总数 范围内生成随机数int offset = ThreadLocalRandom.current().nextInt(totalWeight );// 计算随机数对应的 Invokerfor (int i = 0; i < length; i ) { offset -= weights[i];if (offset < 0) {return invokers.get(i); } } }// 如果所有 Invoker 的权重都一样则随机从 Invoker 列表中返回一个return invokers.get(ThreadLocalRandom.current().nextInt(length)); }}
??
通过遍历权重数组，生成的数累减当前权重值，当 offset 为 0 时，就表示 offset 对应当前的 Invoker 服务。
以生成的随机数为 6 为例，遍历 Invokers 长度：
第一轮：offset = 6 – 4 = 2 不满足 offset < 0，继续遍历。
第二轮：offset = 2 – 6 = -4 满足 offset < 0，返回当前索引对应的 Invoker。因为 offset 返回负数，表示 offset 落在当前 Invoker 权重的区间里。
加权随机策略并非一定按照比例被选到，理论上调用次数越多，分布的比例越接近权重所占的比例。
最少活跃数算法
最小活跃数负载均衡策略（LeastActiveLoadBalance）是从最小活跃数的 Invoker 中进行选择。什么是活跃数呢？活跃数是一个 Invoker 正在处理的请求的数量，当 Invoker 开始处理请求时，会将活跃数加 1，完成请求处理后，将相应 Invoker 的活跃数减 1。找出最小活跃数后，最后根据权重进行选择最终的 Invoker。如果最后找出的最小活跃数相同，则随机从中选中一个 Invoker。
public class LeastActiveLoadBalance extends AbstractLoadBalance {public static final String NAME = “leastactive”;@Overrideprotected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {// Invoker 数量int length = invokers.size();// 所有 Invoker 中的最小活跃值都是 -1int leastActive = -1;// 最小活跃值 Invoker 的数量int leastCount = 0;// 最小活跃值 Invoker 在 Invokers 列表中对应的下标位置int[] leastIndexes = new int[length];// 保存每个 Invoker 的权重int[] weights = new int[length];// 总权重int totalWeight = 0;// 第一个最小活跃数的权重int firstWeight = 0;// 最小活跃数 Invoker 列表的权重是否一样boolean sameWeight = true;// 找出最小活跃数 Invoker 的下标for (int i = 0; i < length; i ) { Invoker<T> invoker = invokers.get(i);// 获取最小活跃数int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive();// 获取权重int afterWarmup = getWeight(invoker, invocation);// 保存权重 weights[i] = afterWarmup;// 如果当前最小活跃数为-1（-1为最小值）或小于leastActiveif (leastActive == -1 || active < leastActive) {// 重置最小活跃数 leastActive = active;// 重置最小活跃数 Invoker 的数量 leastCount = 1;// 保存当前 Invoker 在 Invokers 列表中的索引至leastIndexes数组中 leastIndexes[0] = i;// 重置最小活跃数 invoker 的总权重值 totalWeight = afterWarmup;// 记录当前 Invoker 权重为第一个最小活跃数 Invoker 的权重 firstWeight = afterWarmup;// 因为当前 Invoker 重置为第一个最小活跃数 Invoker ，所以标识所有最小活跃数 Invoker 权重都一样的值为 true sameWeight = true;// 如果当前最小活跃数和已声明的最小活跃数相等 } else if (active == leastActive) {// 记录当前 Invoker 的位置 leastIndexes[leastCount ] = i;// 累加当前 Invoker 权重到总权重中 totalWeight = afterWarmup;// 如果当前权重与firstWeight不相等，则将 sameWeight 改为 falseif (sameWeight && i > 0 && afterWarmup != firstWeight) { sameWeight = false; } } }// 如果最小活跃数 Invoker 只有一个，直接返回该 Invokerif (leastCount == 1) {return invokers.get(leastIndexes[0]); }if (!sameWeight && totalWeight > 0) {// 根据权重随机从最小活跃数 Invoker 列表中选择一个 int offsetWeight = ThreadLocalRandom.current().nextInt(totalWeight);for (int i = 0; i < leastCount; i ) {int leastIndex = leastIndexes[i]; offsetWeight -= weights[leastIndex];if (offsetWeight < 0) {return invokers.get(leastIndex); } } }// 如果所有 Invoker 的权重都一样则随机从 Invoker 列表中返回一个return invokers.get(leastIndexes[ThreadLocalRandom.current().nextInt(leastCount)]); }}
??
这段代码的整个逻辑就是，从 Invokers 列表中筛选出最小活跃数的 Invoker，然后类似加权随机算法策略方式选择最终的 Invoker 服务。
轮询算法
加权轮询负载均衡策略（RoundRobinLoadBalance）是基于权重来决定轮询的比例。普通轮询会将请求均匀的分布在每个节点，但不能很好调节不同性能服务器的请求处理，所以加权负载均衡来根据权重在轮询机制中分配相对应的请求比例给每台服务器。
public class RoundRobinLoadBalance extends AbstractLoadBalance {public static final String NAME = “roundrobin”;private static final int RECYCLE_PERIOD = 60000;protected static class WeightedRoundRobin {private int weight;private AtomicLong current = new AtomicLong(0);private long lastUpdate;public int getWeight() {return weight; }public void setWeight(int weight) {this.weight = weight; current.set(0); }public long increaseCurrent() {return current.addAndGet(weight); }public void sel(int total) { current.addAndGet(-1 * total); }public long getLastUpdate() {return lastUpdate; }public void setLastUpdate(long lastUpdate) {this.lastUpdate = lastUpdate; } }private ConcurrentMap<String, ConcurrentMap<String, WeightedRoundRobin>> methodWeightMap = new ConcurrentHashMap<String, ConcurrentMap<String, WeightedRoundRobin>>();private AtomicBoolean updateLock = new AtomicBoolean();/** * get invoker addr list cached for specified invocation * <p> * <b>for unit test only</b> * * @param invokers * @param invocation * @return */protected <T> Collection<String> getInvokerAddrList(List<Invoker<T>> invokers, Invocation invocation) { String key = invokers.get(0).getUrl().getServiceKey() “.” invocation.getMethodName(); Map<String, WeightedRoundRobin> map = methodWeightMap.get(key);if (map != null) {return map.keySet(); }return null; } @Overrideprotected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {// key 为 接口名 方法名 String key = invokers.get(0).getUrl().getServiceKey() “.” invocation.getMethodName();// 查看缓存中是否存在相应服务接口的信息，如果没有则新添加一个元素到缓存中 ConcurrentMap<String, WeightedRoundRobin> map = methodWeightMap.get(key);if (map == null) { methodWeightMap.putIfAbsent(key, new ConcurrentHashMap<String, WeightedRoundRobin>()); map = methodWeightMap.get(key); }// 总权重int totalWeight = 0;long maxCurrent = Long.MIN_VALUE;// 当前时间戳long now = System.currentTimeMillis();// 最大 current 的 Invoker Invoker<T> selectedInvoker = null;// 保存选中的 WeightedRoundRobin 对象 WeightedRoundRobin selectedWRR = null;// 遍历 Invokers 列表for (Invoker<T> invoker : invokers) {// 从缓存中获取 WeightedRoundRobin 对象 String identifyString = invoker.getUrl().toIdentityString(); WeightedRoundRobin weightedRoundRobin = map.get(identifyString);// 获取当前 Invoker 对象int weight = getWeight(invoker, invocation);// 如果当前 Invoker 没有对应的 WeightedRoundRobin 对象，则新增一个if (weightedRoundRobin == null) { weightedRoundRobin = new WeightedRoundRobin(); weightedRoundRobin.setWeight(weight); map.putIfAbsent(identifyString, weightedRoundRobin); }// 如果当前 Invoker 权重不等于对应的 WeightedRoundRobin 对象中的权重，则重新设置当前权重到对应的 WeightedRoundRobin 对象中if (weight != weightedRoundRobin.getWeight()) { weightedRoundRobin.setWeight(weight); }// 累加权重到 current 中long cur = weightedRoundRobin.increaseCurrent();// 设置 weightedRoundRobin 对象最后更新时间 weightedRoundRobin.setLastUpdate(now);// 最大 current 的 Invoker，并赋值给相应的变量if (cur > maxCurrent) { maxCurrent = cur; selectedInvoker = invoker; selectedWRR = weightedRoundRobin; }// 累加权重到总权重中 totalWeight = weight; }// 如果 Invokers 列表中的数量不等于缓存map中的数量if (!updateLock.get() && invokers.size() != map.size()) {if (updateLock.compareAndSet(false, true)) {try {// 拷贝 map 到 newMap 中 ConcurrentMap<String, WeightedRoundRobin> newMap = new ConcurrentHashMap<String, WeightedRoundRobin>(); newMap.putAll(map);// newMap 转化为 Iterator Iterator<Entry<String, WeightedRoundRobin>> it = newMap.entrySet().iterator();// 循环删除超过设定时长没更新的缓存while (it.hasNext()) { Entry<String, WeightedRoundRobin> item = it.next();if (now – item.getValue().getLastUpdate() > RECYCLE_PERIOD) { it.remove(); } }// 将当前newMap服务缓存中 methodWeightMap.put(key, newMap); } finally { updateLock.set(false); } } }// 如果存在被选中的 Invokerif (selectedInvoker != null) {// 计算 current = current – totalWeight selectedWRR.sel(totalWeight);return selectedInvoker; }// 正常情况这里不会到达return invokers.get(0); }}
上面选中 Invoker 逻辑为：每个 Invoker 都有一个 current 值，初始值为自身权重。在每个 Invoker 中current=current weight。遍历完 Invoker 后，current 最大的那个 Invoker 就是本次选中的 Invoker。选中 Invoker 后，将本次 current 值计算current=current-totalWeight。
以上面192.168.1.10和192.168.1.11两个负载均衡的服务，权重分别为 4、6 。基于选中前current=current weight、选中后current=current-totalWeight计算公式得出如下
一致性 Hash 算法
一致性 Hash 负载均衡策略（ConsistentHashLoadBalance）是让参数相同的请求分配到同一机器上。把每个服务节点分布在一个环上，请求也分布在环形中。以请求在环上的位置，顺时针寻找换上第一个服务节点。如图所示：
同时，为避免请求散列不均匀，dubbo 中会将每个 Invoker 再虚拟多个节点出来，使得请求调用更加均匀。
一致性 Hash 修改配置如下：
<!– dubbo 默认只对第一个参数进行 hash 标识，指定 hash参数–><dubbo:parameter key =”hash.arguments” value=”1″/><!–虚拟节点数量 –><dubbo:parameter key=”hash.nodes” value=”200″/>
一致性 Hash 实现如下：
public class ConsistentHashLoadBalance extends AbstractLoadBalance {public static final String NAME = “consistenthash”;/** * Hash nodes name */public static final String HASH_NODES = “hash.nodes”;/** * Hash arguments name */public static final String HASH_ARGUMENTS = “hash.arguments”;private final ConcurrentMap<String, ConsistentHashSelector<?>> selectors = new ConcurrentHashMap<String, ConsistentHashSelector<?>>();@SuppressWarnings(“unchecked”)@Overrideprotected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {// 获取请求的方法名String methodName = RpcUtils.getMethodName(invocation);// key = 接口名 方法名String key = invokers.get(0).getUrl().getServiceKey() “.” methodName;// invokers 的 hashcode int identityHashCode = System.identityHashCode(invokers);// 查看缓存中是否存在对应 key 的数据，或 Invokers 列表是否有过变动。如果没有，则新添加到缓存中，并且返回负载均衡得出的 Invoker ConsistentHashSelector<T> selector = (ConsistentHashSelector<T>) selectors.get(key);if (selector == null || selector.identityHashCode != identityHashCode) { selectors.put(key, new ConsistentHashSelector<T>(invokers, methodName, identityHashCode)); selector = (ConsistentHashSelector<T>) selectors.get(key); }return selector.select(invocation); }// ConsistentHashSelector class …}
doSelect 中主要实现缓存检查和 Invokers 变动检查，一致性 hash 负载均衡的实现在这个内部类 ConsistentHashSelector 中实现。
Dubbo 负载均衡的实现，技巧上还是比较优雅，可以多多学习其编码思维。在研究其代码时，需要仔细研究其实现原理，否则比较难懂其思想。
Java 线程通信之 wait/notify 机制
复习 2 个月拿下美团 offer ，我都做了些啥
一个基于 RabbitMQ 的可复用的分布式事务消息架构方案！
看完本文有收获？请转发分享给更多人
