Dubbo之LoadBalance源码分析
Posted Java后端笔记
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Dubbo之LoadBalance源码分析相关的知识，希望对你有一定的参考价值。
前言
LoadBalance，就是负载均衡，那么何为负载均衡，就是让服务提供者相对平摊请求，不要出现请求总落在一个提供者的情况
接口定义
@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;
}
select方法作用是从invokers选出下一个被调用的invoker，具体有哪些策略，如下
然后这个LoadBalance主要使用在Cluster模块中。比如failover选择下一个invoker。
下面开始源码讲解
源码
AbstractLoadBalance
上述4中策略的实现，都会继承AbstractLoadBalance这个模板类，在这个模板类中封装了getWeight方法，获取invoker的权重，特别的是，这个权重和预热时间有关，只有提供者在线时长到达了预热时间，调用这个方法获取invoker权重的时候，才能获得100%的权重。在子类中获取invoker权重都是调用这个方法
看下带有预热逻辑的权重方法
//计算预热权重
protected int getWeight(Invoker<?> invoker, Invocation invocation) {
int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT);
if (weight > 0) {
long timestamp = invoker.getUrl().getParameter(Constants.REMOTE_TIMESTAMP_KEY, 0L);
if (timestamp > 0L) {
//提供者在线时长
int uptime = (int) (System.currentTimeMillis() - timestamp);
//预热时间默认10分钟
int warmup = invoker.getUrl().getParameter(Constants.WARMUP_KEY, Constants.DEFAULT_WARMUP);
if (uptime > 0 && uptime < warmup) {
weight = calculateWarmupWeight(uptime, warmup, weight);
}
}
}
return weight;
}
//用于计算预热权重
static int calculateWarmupWeight(int uptime, int warmup, int weight) {
int ww = (int) ((float) uptime / ((float) warmup / (float) weight));
return ww < 1 ? 1 : (ww > weight ? weight : ww);
}
AbstractLoadBalance实现了select方法，增加了对invoker数量的判断，如果只有一个直接返回，invokers超过1个才需要使用负载均衡选择逻辑，具体负载均衡逻辑由子类实现doSelect方法
@Override
public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) {
if (invokers == null || invokers.isEmpty())
return null;
//如果只有一个提供者直接返回，预热失效
if (invokers.size() == 1)
return invokers.get(0);
return doSelect(invokers, url, invocation);
}
//让子类实现doSelect
protected abstract <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation);
为什么要预热，jvm运行时会对字节码进行优化，刚启动的字节码肯定不是最优的。或者是提供者本身有其他缓存需要初始化之类的。所以预热是有必要的。不要一启动就和其他提供者承受同样流量，可能效率会变慢。当然，如果只有一个提供者的情况下，预热就失效了。
RandomLoadBalance
随机算法，如果每个invokers权重一样，那么就是普通的随机算法，如果不同就是加权随机
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
int length = invokers.size(); // Number of invokers
int totalWeight = 0; // The sum of weights
boolean sameWeight = true; // Every invoker has the same weight?
for (int i = 0; i < length; i++) {
int weight = getWeight(invokers.get(i), invocation);
totalWeight += weight; // Sum
if (sameWeight && i > 0
&& weight != getWeight(invokers.get(i - 1), invocation)) {
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.
int offset = random.nextInt(totalWeight);
// Return a invoker based on the random value.
for (int i = 0; i < length; i++) {
offset -= getWeight(invokers.get(i), invocation);
if (offset < 0) {
return invokers.get(i);
}
}
}
//如果提供者权重都一样，普通随机
// If all invokers have the same weight value or totalWeight=0, return evenly.
return invokers.get(random.nextInt(length));
}
RoundRobinLoadBalance
轮训算法。如果每个invoker权重一样，就是普通的轮训算法。如果不同，是加权的轮训算法。
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
int length = invokers.size(); // Number of invokers
int maxWeight = 0; // The maximum weight
int minWeight = Integer.MAX_VALUE; // The minimum weight
final LinkedHashMap<Invoker<T>, IntegerWrapper> invokerToWeightMap = new LinkedHashMap<Invoker<T>, IntegerWrapper>();
int weightSum = 0;
for (int i = 0; i < length; i++) {
int weight = getWeight(invokers.get(i), invocation);
maxWeight = Math.max(maxWeight, weight); // Choose the maximum weight
minWeight = Math.min(minWeight, weight); // Choose the minimum weight
if (weight > 0) {
invokerToWeightMap.put(invokers.get(i), new IntegerWrapper(weight));
weightSum += weight;
}
}
AtomicPositiveInteger sequence = sequences.get(key);
if (sequence == null) {
sequences.putIfAbsent(key, new AtomicPositiveInteger());
sequence = sequences.get(key);
}
int currentSequence = sequence.getAndIncrement();
//如果每个提供者权重不一样，采用加权轮训
if (maxWeight > 0 && minWeight < maxWeight) {
int mod = currentSequence % weightSum;
for (int i = 0; i < maxWeight; i++) {
for (Map.Entry<Invoker<T>, IntegerWrapper> each : invokerToWeightMap.entrySet()) {
final Invoker<T> k = each.getKey();
final IntegerWrapper v = each.getValue();
if (mod == 0 && v.getValue() > 0) {
return k;
}
if (v.getValue() > 0) {
v.decrement();
mod--;
}
}
}
}
//每个服务提供者权重一样，就是普通轮训
// Round robin
return invokers.get(currentSequence % length);
}
LeastActiveLoadBalance
最少活跃调用数。如果最小活跃调用数的invokers大于1，如果这些invokers权重相同，采用随机算法选出invoker。如不同，采用加权随机算法。
@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
int length = invokers.size(); // Number of invokers
int leastActive = -1; // The least active value of all invokers
int leastCount = 0; // The number of invokers having the same least active value (leastActive)
int[] leastIndexs = new int[length]; // The index of invokers having the same least active value (leastActive)
int totalWeight = 0; // The sum of weights
int firstWeight = 0; // Initial value, used for comparision
boolean sameWeight = true; // Every invoker has the same weight value?
//获取leasractive的数组
for (int i = 0; i < length; i++) {
Invoker<T> invoker = invokers.get(i);
int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive(); // Active number
int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT); // Weight
if (leastActive == -1 || active < leastActive) { // Restart, when find a invoker having smaller least active value.
leastActive = active; // Record the current least active value
leastCount = 1; // Reset leastCount, count again based on current leastCount
leastIndexs[0] = i; // Reset
totalWeight = weight; // Reset
firstWeight = weight; // Record the weight the first invoker
sameWeight = true; // Reset, every invoker has the same weight value?
} else if (active == leastActive) { // If current invoker's active value equals with leaseActive, then accumulating.
leastIndexs[leastCount++] = i; // Record index number of this invoker
totalWeight += weight; // Add this invoker's weight to totalWeight.
// If every invoker has the same weight?
if (sameWeight && i > 0
&& weight != firstWeight) {
sameWeight = false;
}
}
}
// assert(leastCount > 0)
if (leastCount == 1) {
// If we got exactly one invoker having the least active value, return this invoker directly.
return invokers.get(leastIndexs[0]);
}
//在leastactive数组里面加权随机选择一个
if (!sameWeight && totalWeight > 0) {
// If (not every invoker has the same weight & at least one invoker's weight>0), select randomly based on totalWeight.
int offsetWeight = random.nextInt(totalWeight);
// Return a invoker based on the random value.
for (int i = 0; i < leastCount; i++) {
int leastIndex = leastIndexs[i];
offsetWeight -= getWeight(invokers.get(leastIndex), invocation);
if (offsetWeight <= 0)
return invokers.get(leastIndex);
}
}
//在leastative数组内随机选择一个
// If all invokers have the same weight value or totalWeight=0, return evenly.
return invokers.get(leastIndexs[random.nextInt(leastCount)]);
}
活跃调用次数会通过ActiveLimitFilter记录在RpcStatus中
ConsistentHashLoadBalance
一致性hash算法。通过调用的参数进行一致性hash,和权重无关。
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
String methodName = RpcUtils.getMethodName(invocation);
String key = invokers.get(0).getUrl().getServiceKey() + "." + methodName;
int identityHashCode = System.identityHashCode(invokers);
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);
}
一致性hash的主要逻辑都在ConsistentHashSelector中，在它的构造函数中会生成虚拟节点。默认每个invoker 160个。hash环的总节点数为2的32次方-1个
ConsistentHashSelector(List<Invoker<T>> invokers, String methodName, int identityHashCode) {
this.virtualInvokers = new TreeMap<Long, Invoker<T>>();
this.identityHashCode = identityHashCode;
URL url = invokers.get(0).getUrl();
//每个invoker生成的虚拟节点数
this.replicaNumber = url.getMethodParameter(methodName, "hash.nodes", 160);
String[] index = Constants.COMMA_SPLIT_PATTERN.split(url.getMethodParameter(methodName, "hash.arguments", "0"));
argumentIndex = new int[index.length];
for (int i = 0; i < index.length; i++) {
argumentIndex[i] = Integer.parseInt(index[i]);
}
for (Invoker<T> invoker : invokers) {
//生成虚拟节点
String address = invoker.getUrl().getAddress();
for (int i = 0; i < replicaNumber / 4; i++) {
byte[] digest = md5(address + i);
for (int h = 0; h < 4; h++) {
long m = hash(digest, h);
virtualInvokers.put(m, invoker);
}
}
}
}
然后通过对方法参数的hash去取得对应的invoker
public Invoker<T> select(Invocation invocation) {
String key = toKey(invocation.getArguments());
byte[] digest = md5(key);
return selectForKey(hash(digest, 0));
}
private String toKey(Object[] args) {
StringBuilder buf = new StringBuilder();
for (int i : argumentIndex) {
if (i >= 0 && i < args.length) {
buf.append(args[i]);
}
}
return buf.toString();
}
private Invoker<T> selectForKey(long hash) {
//取大于hash的下一个节点
Map.Entry<Long, Invoker<T>> entry = virtualInvokers.tailMap(hash, true).firstEntry();
if (entry == null) {
//hash大于最后一个节点，取第一个节点
entry = virtualInvokers.firstEntry();
}
return entry.getValue();
}
tailMap方法用于取得virtualInvokers中key的hash大于参数hash的子Map，由于virtualInvokers是TreeMap，并且key为long类型，所以子Map的第一个Entry就对应hash环中的相匹配的invoker。
关于一致性hash可以看下面这篇文章(https://www.cnblogs.com/lpfuture/p/5796398.html)
node对应我们的invoker的hash 键对应我们参数的hash 通过一致性hash，能够保证大部分情况下，参数一致的请求落到同一个提供者。如果提供者发生上下线，只会影响一小部分的请求。
总结
LoadBalance中好多算法，加权随机，加权轮训以及一致性hash真是有意思。大家好好体会这个源码，看懂了，真是很有意思。
最后
以上是关于Dubbo之LoadBalance源码分析的主要内容，如果未能解决你的问题，请参考以下文章