dubbo源码分析负载均衡

dubbo的负载均衡算法

Dubbo 需要对服务消费者的调用请求进行分配&＃xff0c;避免少数服务提供者负载过大。服务提供者负载过大&＃xff0c;会导致部分请求超时。因此将负载均衡到每个服务提供者上&＃xff0c;是非常必要的。Dubbo 提供了4种负载均衡实现&＃xff0c;分别是基于权重随机算法的 RandomLoadBalance、基于最少活跃调用数算法的 LeastActiveLoadBalance、基于 hash 一致性的 ConsistentHashLoadBalance&＃xff0c;以及基于加权轮询算法的 RoundRobinLoadBalance。

LoadBalance

LoadBalance是一个SPI扩展接口

&＃64;SPI("random")
public interface LoadBalance {&＃64;Adaptive({"loadbalance"}) Invoker select(List> var1, URL var2, Invocation var3) throws RpcException;
}

LoadBalance有4种实现&＃xff0c;其中AbstractLoadBalance直接实现LoadBalance&＃xff0c;其余实现类继承AbstractLoadBalance

AbstractLoadBalance

public abstract class AbstractLoadBalance implements LoadBalance {public AbstractLoadBalance() {}static int calculateWarmupWeight(int uptime, int warmup, int weight) {// 计算权重&＃xff0c;下面代码逻辑上形似于 (uptime / warmup) * weight。// 随着服务运行时间 uptime 增大&＃xff0c;权重计算值 ww 会慢慢接近配置值 weightint ww &＃61; (int)((float)uptime / ((float)warmup / (float)weight));return ww <1 ? 1 : (ww > weight ? weight : ww);}public Invoker select(List> invokers, URL url, Invocation invocation) {if (invokers !&＃61; null && invokers.size() !&＃61; 0) {return invokers.size() &＃61;&＃61; 1 ? (Invoker)invokers.get(0) : this.doSelect(invokers, url, invocation);} else {return null;}}protected abstract Invoker doSelect(List> var1, URL var2, Invocation var3);protected int getWeight(Invoker invoker, Invocation invocation) {// 从 url 中获取权重 weight 配置值int weight &＃61; invoker.getUrl().getMethodParameter(invocation.getMethodName(), "weight", 100);if (weight > 0) {// 获取服务提供者启动时间戳 即开始运行的时间戳long timestamp &＃61; invoker.getUrl().getParameter("remote.timestamp", 0L);if (timestamp > 0L) {// 计算服务提供者运行时长 int uptime &＃61; (int)(System.currentTimeMillis() - timestamp);// 获取服务预热时间&＃xff0c;默认为10分钟int warmup &＃61; invoker.getUrl().getParameter("warmup", 600000);// 如果服务运行时间小于预热时间&＃xff0c;则重新计算服务权重&＃xff0c;即降权if (uptime > 0 && uptime }


该过程主要用于保证当服务运行时长小于服务预热时间时&＃xff0c;对服务进行降权&＃xff0c;避免让服务在启动之初就处于高负载状态。服务预热是一个优化手段&＃xff0c;与此类似的还有 JVM 预热。主要目的是让服务启动后“低功率”运行一段时间&＃xff0c;使其效率慢慢提升至最佳状态。

RandomLoadBalance

RandomLoadBalance 是加权随机算法的具体实现&＃xff0c;它的算法思想很简单。假设我们有一组服务器 servers &＃61; [A, B, C]&＃xff0c;他们对应的权重为 weights &＃61; [5, 3, 2]&＃xff0c;权重总和为10。现在把这些权重值平铺在一维坐标值上&＃xff0c;[0, 5) 区间属于服务器 A&＃xff0c;[5, 8) 区间属于服务器 B&＃xff0c;[8, 10) 区间属于服务器 C。接下来通过随机数生成器生成一个范围在 [0, 10) 之间的随机数&＃xff0c;然后计算这个随机数会落到哪个区间上。比如数字3会落到服务器 A 对应的区间上&＃xff0c;此时返回服务器 A 即可。权重越大的机器&＃xff0c;在坐标轴上对应的区间范围就越大&＃xff0c;因此随机数生成器生成的数字就会有更大的概率落到此区间内。只要随机数生成器产生的随机数分布性很好&＃xff0c;在经过多次选择后&＃xff0c;每个服务器被选中的次数比例接近其权重比例。比如&＃xff0c;经过一万次选择后&＃xff0c;服务器 A 被选中的次数大约为5000次&＃xff0c;服务器 B 被选中的次数约为3000次&＃xff0c;服务器 C 被选中的次数约为2000次。

public class RandomLoadBalance extends AbstractLoadBalance {public static final String NAME &＃61; "random";private final Random random &＃61; new Random();public RandomLoadBalance() {}protected Invoker doSelect(List> invokers, URL url, Invocation invocation) {int length &＃61; invokers.size();int totalWeight &＃61; 0;boolean sameWeight &＃61; true;int offset;int i;//for(offset &＃61; 0; offset 0 && i !&＃61; this.getWeight((Invoker)invokers.get(offset - 1), invocation)) {sameWeight &＃61; false;}}// 下面的 if 分支主要用于获取随机数&＃xff0c;并计算随机数落在哪个区间上if (totalWeight > 0 && !sameWeight) {//获得权重区间内随机值offset &＃61; this.random.nextInt(totalWeight);// 循环让 offset 数减去服务提供者权重值&＃xff0c;当 offset 小于0时&＃xff0c;返回相应的 Invoker。// 举例说明一下&＃xff0c;我们有 servers &＃61; [A, B, C]&＃xff0c;weights &＃61; [5, 3, 2]&＃xff0c;offset &＃61; 7。// 第一次循环&＃xff0c;offset - 5 &＃61; 2 > 0&＃xff0c;即 offset > 5&＃xff0c;// 表明其不会落在服务器 A 对应的区间上。// 第二次循环&＃xff0c;offset - 3 &＃61; -1 <0&＃xff0c;即 5 }


RandomLoadBalance 的算法思想比较简单&＃xff0c;在经过多次请求后&＃xff0c;能够将调用请求按照权重值进行“均匀”分配。当然 RandomLoadBalance 也存在一定的缺点&＃xff0c;当调用次数比较少时&＃xff0c;Random 产生的随机数可能会比较集中&＃xff0c;此时多数请求会落到同一台服务器上。这个缺点并不是很严重&＃xff0c;多数情况下可以忽略。RandomLoadBalance 是一个简单&＃xff0c;高效的负载均衡实现&＃xff0c;因此 Dubbo 选择它作为缺省实现。 问题&＃xff1a;deSelect方法的参数invokers固定时 可以实现以上的"均匀"分配吧

LeastActiveLoadBalance

LeastActiveLoadBalance 翻译过来是最小活跃数负载均衡。活跃调用数越小&＃xff0c;表明该服务提供者效率越高&＃xff0c;单位时间内可处理更多的请求。此时应优先将请求分配给该服务提供者。在具体实现中&＃xff0c;每个服务提供者对应一个活跃数 active。初始情况下&＃xff0c;所有服务提供者活跃数均为0。每收到一个请求&＃xff0c;活跃数加1&＃xff0c;完成请求后则将活跃数减1。在服务运行一段时间后&＃xff0c;性能好的服务提供者处理请求的速度更快&＃xff0c;因此活跃数下降的也越快&＃xff0c;此时这样的服务提供者能够优先获取到新的服务请求、这就是最小活跃数负载均衡算法的基本思想。除了最小活跃数&＃xff0c;LeastActiveLoadBalance 在实现上还引入了权重值。所以准确的来说&＃xff0c;LeastActiveLoadBalance 是基于加权最小活跃数算法实现的。举个例子说明一下&＃xff0c;在一个服务提供者集群中&＃xff0c;有两个性能优异的服务提供者。某一时刻它们的活跃数相同&＃xff0c;此时 Dubbo 会根据它们的权重去分配请求&＃xff0c;权重越大&＃xff0c;获取到新请求的概率就越大。如果两个服务提供者权重相同&＃xff0c;此时随机选择一个即可。

public class LeastActiveLoadBalance extends AbstractLoadBalance {public static final String NAME &＃61; "leastactive";private final Random random &＃61; new Random();public LeastActiveLoadBalance() {}protected Invoker doSelect(List> invokers, URL url, Invocation invocation) {int length &＃61; invokers.size();int leastActive &＃61; -1;int leastCount &＃61; 0;int[] leastIndexs &＃61; new int[length];int totalWeight &＃61; 0;int firstWeight &＃61; 0;boolean sameWeight &＃61; true;int offsetWeight;int leastIndex;for(offsetWeight &＃61; 0; offsetWeight invoker &＃61; (Invoker)invokers.get(offsetWeight);leastIndex &＃61; RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive();int weight &＃61; invoker.getUrl().getMethodParameter(invocation.getMethodName(), "weight", 100);if (leastActive !&＃61; -1 && leastIndex >&＃61; leastActive) {if (leastIndex &＃61;&＃61; leastActive) {leastIndexs[leastCount&＃43;&＃43;] &＃61; offsetWeight;totalWeight &＃43;&＃61; weight;if (sameWeight && offsetWeight > 0 && weight !&＃61; firstWeight) {sameWeight &＃61; false;}}} else {leastActive &＃61; leastIndex;leastCount &＃61; 1;leastIndexs[0] &＃61; offsetWeight;totalWeight &＃61; weight;firstWeight &＃61; weight;sameWeight &＃61; true;}}if (leastCount &＃61;&＃61; 1) {return (Invoker)invokers.get(leastIndexs[0]);} else {if (!sameWeight && totalWeight > 0) {offsetWeight &＃61; this.random.nextInt(totalWeight);for(int i &＃61; 0; i }


1. 遍历 invokers 列表&＃xff0c;寻找活跃数最小的 Invoker
2. 如果有多个 Invoker 具有相同的最小活跃数&＃xff0c;此时记录下这些 Invoker 在 invokers 集合中的下标&＃xff0c;并累加它们的权重&＃xff0c;比较它们的权重值是否相等
3. 如果只有一个 Invoker 具有最小的活跃数&＃xff0c;此时直接返回该 Invoker 即可
4. 如果有多个 Invoker 具有最小活跃数&＃xff0c;且它们的权重不相等&＃xff0c;此时处理方式和 RandomLoadBalance 一致
5. 如果有多个 Invoker 具有最小活跃数&＃xff0c;但它们的权重相等&＃xff0c;此时随机返回一个即可

ConsistentHashLoadBalance

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

public class ConsistentHashLoadBalance extends AbstractLoadBalance {private final ConcurrentMap> selectors &＃61; new ConcurrentHashMap();public ConsistentHashLoadBalance() {}protected Invoker doSelect(List> invokers, URL url, Invocation invocation) {String key &＃61; ((Invoker)invokers.get(0)).getUrl().getServiceKey() &＃43; "." &＃43; invocation.getMethodName();int identityHashCode &＃61; System.identityHashCode(invokers);ConsistentHashLoadBalance.ConsistentHashSelector selector &＃61; (ConsistentHashLoadBalance.ConsistentHashSelector)this.selectors.get(key);if (selector &＃61;&＃61; null || selector.identityHashCode !&＃61; identityHashCode) {this.selectors.put(key, new ConsistentHashLoadBalance.ConsistentHashSelector(invokers, invocation.getMethodName(), identityHashCode));selector &＃61; (ConsistentHashLoadBalance.ConsistentHashSelector)this.selectors.get(key);}return selector.select(invocation);}private static final class ConsistentHashSelector {private final TreeMap> virtualInvokers &＃61; new TreeMap();private final int replicaNumber;private final int identityHashCode;private final int[] argumentIndex;ConsistentHashSelector(List> invokers, String methodName, int identityHashCode) {this.identityHashCode &＃61; identityHashCode;URL url &＃61; ((Invoker)invokers.get(0)).getUrl();this.replicaNumber &＃61; url.getMethodParameter(methodName, "hash.nodes", 160);String[] index &＃61; Constants.COMMA_SPLIT_PATTERN.split(url.getMethodParameter(methodName, "hash.arguments", "0"));this.argumentIndex &＃61; new int[index.length];for(int i &＃61; 0; i invoker &＃61; (Invoker)i$.next();String address &＃61; invoker.getUrl().getAddress();for(int i &＃61; 0; i select(Invocation invocation) {String key &＃61; this.toKey(invocation.getArguments());byte[] digest &＃61; this.md5(key);return this.selectForKey(this.hash(digest, 0));}private String toKey(Object[] args) {StringBuilder buf &＃61; new StringBuilder();int[] arr$ &＃61; this.argumentIndex;int len$ &＃61; arr$.length;for(int i$ &＃61; 0; i$ &＃61; 0 && i selectForKey(long hash) {Long key &＃61; hash;if (!this.virtualInvokers.containsKey(key)) {SortedMap> tailMap &＃61; this.virtualInvokers.tailMap(key);if (tailMap.isEmpty()) {key &＃61; (Long)this.virtualInvokers.firstKey();} else {key &＃61; (Long)tailMap.firstKey();}}Invoker invoker &＃61; (Invoker)this.virtualInvokers.get(key);return invoker;}private long hash(byte[] digest, int number) {return ((long)(digest[3 &＃43; number * 4] & 255) <<24 | (long)(digest[2 &＃43; number * 4] & 255) <<16 | (long)(digest[1 &＃43; number * 4] & 255) <<8 | (long)(digest[number * 4] & 255)) & 4294967295L;}private byte[] md5(String value) {MessageDigest md5;try {md5 &＃61; MessageDigest.getInstance("MD5");} catch (NoSuchAlgorithmException var6) {throw new IllegalStateException(var6.getMessage(), var6);}md5.reset();byte[] bytes;try {bytes &＃61; value.getBytes("UTF-8");} catch (UnsupportedEncodingException var5) {throw new IllegalStateException(var5.getMessage(), var5);}md5.update(bytes);return md5.digest();}}
}


doSelect 方法主要做了一些前置工作&＃xff0c;比如检测 invokers 列表是不是变动过&＃xff0c;以及创建 ConsistentHashSelector。这些工作做完后&＃xff0c;接下来开始调用 ConsistentHashSelector 的 select 方法执行负载均衡逻辑。 ConsistentHashSelector 的构造方法执行了一系列的初始化逻辑&＃xff0c;比如从配置中获取虚拟节点数以及参与 hash 计算的参数下标&＃xff0c;默认情况下只使用第一个参数进行 hash。需要特别说明的是&＃xff0c;ConsistentHashLoadBalance 的负载均衡逻辑只受参数值影响&＃xff0c;具有相同参数值的请求将会被分配给同一个服务提供者。ConsistentHashLoadBalance 不 关系权重&＃xff0c;因此使用时需要注意一下。

RoundRobinLoadBalance

所谓轮询是指将请求轮流分配给每台服务器。举个例子&＃xff0c;我们有三台服务器 A、B、C。我们将第一个请求分配给服务器 A&＃xff0c;第二个请求分配给服务器 B&＃xff0c;第三个请求分配给服务器 C&＃xff0c;第四个请求再次分配给服务器 A。这个过程就叫做轮询。轮询是一种无状态负载均衡算法&＃xff0c;实现简单&＃xff0c;适用于每台服务器性能相近的场景下。但现实情况下&＃xff0c;我们并不能保证每台服务器性能均相近。如果我们将等量的请求分配给性能较差的服务器&＃xff0c;这显然是不合理的。因此&＃xff0c;这个时候我们需要对轮询过程进行加权&＃xff0c;以调控每台服务器的负载。经过加权后&＃xff0c;每台服务器能够得到的请求数比例&＃xff0c;接近或等于他们的权重比。比如服务器 A、B、C 权重比为 5:2:1。那么在8次请求中&＃xff0c;服务器 A 将收到其中的5次请求&＃xff0c;服务器 B 会收到其中的2次请求&＃xff0c;服务器 C 则收到其中的1次请求。

public class RoundRobinLoadBalance extends AbstractLoadBalance { public static final String NAME &＃61; "roundrobin"; private final ConcurrentMap sequences &＃61; new ConcurrentHashMap();

public RoundRobinLoadBalance() {
}protected Invoker doSelect(List> invokers, URL url, Invocation invocation) {String key &＃61; ((Invoker)invokers.get(0)).getUrl().getServiceKey() &＃43; "." &＃43; invocation.getMethodName();int length &＃61; invokers.size();int maxWeight &＃61; 0;int minWeight &＃61; 2147483647;LinkedHashMap, RoundRobinLoadBalance.IntegerWrapper> invokerToWeightMap &＃61; new LinkedHashMap();int weightSum &＃61; 0;int currentSequence;for(int i &＃61; 0; i 0) {invokerToWeightMap.put(invokers.get(i), new RoundRobinLoadBalance.IntegerWrapper(currentSequence));weightSum &＃43;&＃61; currentSequence;}}AtomicPositiveInteger sequence &＃61; (AtomicPositiveInteger)this.sequences.get(key);if (sequence &＃61;&＃61; null) {this.sequences.putIfAbsent(key, new AtomicPositiveInteger());sequence &＃61; (AtomicPositiveInteger)this.sequences.get(key);}currentSequence &＃61; sequence.getAndIncrement();if (maxWeight > 0 && minWeight , RoundRobinLoadBalance.IntegerWrapper> each &＃61; (Entry)i$.next();Invoker k &＃61; (Invoker)each.getKey();RoundRobinLoadBalance.IntegerWrapper v &＃61; (RoundRobinLoadBalance.IntegerWrapper)each.getValue();if (mod &＃61;&＃61; 0 && v.getValue() > 0) {return k;}if (v.getValue() > 0) {v.decrement();--mod;}}}}return (Invoker)invokers.get(currentSequence % length);
}private static final class IntegerWrapper {private int value;public IntegerWrapper(int value) {this.value &＃61; value;}public int getValue() {return this.value;}public void setValue(int value) {this.value &＃61; value;}public void decrement() {--this.value;}
}


}