1 Dubbo的负载均衡策略概述
Dubbo的负载均衡策略应用于服务消费方。当服务提供者是集群时，通过在消费方设置负载均衡策略，避免大量请求一直集中在其中的某一个或者某几个服务提供方机器上。
Dubbo提供了多种负载均衡策略，默认为随机策略-Random LoadBalance，即每次随机调用一台服务提供者的服务。
负载均衡策略的核心方法是各负载均衡策略类的 doSelect() 方法，用于从服务提供者列表中选择其中一个来调用。
调用该方法的地方为 AbstractLoadBalance#select() 方法。源码如下所示。
public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) {
if (CollectionUtils.isEmpty(invokers)) {
return null;
}
if (invokers.size() == 1) {
return invokers.get(0);
}
return doSelect(invokers, url, invocation);
}2 源码剖析
下面将介绍一些主要的负载均衡策略，以及对其源码进行解读。
2.1 随机策略-Random LoadBalance
2.1.1 概述
要点：每次随机调用一台服务提供者的服务，并且可以设置不同服务提供者的权重。
使用：
<dubbo:reference ... loadbalance="random"/>2.1.2 源码剖析
实现类是RandomLoadBalance，源码如下所示。
/**
* This class select one provider from multiple providers randomly.
* You can define weights for each provider:
* If the weights are all the same then it will use random.nextInt(number of invokers).
* If the weights are different then it will use random.nextInt(w1 + w2 + ... + wn)
* Note that if the performance of the machine is better than others, you can set a larger weight.
* If the performance is not so good, you can set a smaller weight.
*/
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
// Number of invokers
int 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 invoker, the minWeight = 0 or the maxWeight of the last invoker
int[] weights = new int[length];
// The sum of weights
int totalWeight = 0;
for (int i = 0; i < length; i++) {
int weight = getWeight(invokers.get(i), invocation);
// Sum
totalWeight += weight;
// save for later use
weights[i] = totalWeight;
if (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.
int offset = ThreadLocalRandom.current().nextInt(totalWeight);
// Return an invoker based on the random value.
if (length <= 4) {
for (int i = 0; i < length; i++) {
// 注：和dubbo老版本不一样的点是权重大的不一定会首先被选中，因为 weights[i] = totalWeight
if (offset < weights[i]) {
return invokers.get(i);
}
}
} else {
int i = Arrays.binarySearch(weights, offset);
if (i < 0) {
i = -i - 1;
} else {
while (weights[i+1] == offset) {
i++;
}
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));
}2.2 轮循策略-RoundRobin LoadBalance
2.2.1 概述
要点：轮循，按照公约后的权重设置轮循比率。当某台机器执行很慢时，将导致该机器上的请求积压。
2.2.2 源码剖析
实现类是RoundRobinLoadBalance，源码如下所示。
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
String key = invokers.get(0).getUrl().getServiceKey() + "." + RpcUtils.getMethodName(invocation);
ConcurrentMap<String, WeightedRoundRobin> map = ConcurrentHashMapUtils.computeIfAbsent(methodWeightMap, 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) {
String identifyString = invoker.getUrl().toIdentityString();
int weight = getWeight(invoker, invocation);
WeightedRoundRobin weightedRoundRobin = ConcurrentHashMapUtils.computeIfAbsent(map, identifyString, k -> {
WeightedRoundRobin wrr = new WeightedRoundRobin();
wrr.setWeight(weight);
return wrr;
});
if (weight != weightedRoundRobin.getWeight()) {
//weight changed
weightedRoundRobin.setWeight(weight);
}
long cur = weightedRoundRobin.increaseCurrent();
weightedRoundRobin.setLastUpdate(now);
if (cur > maxCurrent) {
maxCurrent = cur;
selectedInvoker = invoker;
selectedWRR = weightedRoundRobin;
}
totalWeight += weight;
}
if (invokers.size() != map.size()) {
map.entrySet().removeIf(item -> now - item.getValue().getLastUpdate() > RECYCLE_PERIOD);
}
if (selectedInvoker != null) {
selectedWRR.sel(totalWeight);
return selectedInvoker;
}
// should not happen here
return invokers.get(0);
}2.3 最少活跃调用数策略-LeastActive LoadBalance
2.3.1 概述
要点：在每个服务提供者里维护一个活跃数计数器，用来记录当前同时处理请求的个数。路由选择时会选择该活跃数最小的机器进行调用。
2.3.2 源码剖析
实现类是LeastActiveLoadBalance，源码如下所示。
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
// Number of invokers
int 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)
int[] leastIndexes = new int[length];
// the weight of every invokers
int[] weights = new int[length];
// The sum of the warmup weights of all the least active invokers
int totalWeight = 0;
// The weight of the first least active invoker
int firstWeight = 0;
// Every least active invoker has the same weight value?
boolean sameWeight = true;
// Filter out all the least active invokers
for (int i = 0; i < length; i++) {
Invoker<T> invoker = invokers.get(i);
// Get the active number of the invoker
int active = RpcStatus.getStatus(invoker.getUrl(), RpcUtils.getMethodName(invocation)).getActive();
// Get the weight of the invoker's configuration. The default value is 100.
int afterWarmup = getWeight(invoker, invocation);
// save for later use
weights[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
leastActive = active;
// Reset the number of least active invokers
leastCount = 1;
// Put the first least active invoker first in leastIndexes
leastIndexes[0] = i;
// Reset totalWeight
totalWeight = afterWarmup;
// Record the weight the first least active invoker
firstWeight = afterWarmup;
// Each invoke has the same weight (only one invoker here)
sameWeight = true;
// If current invoker's active value equals with leaseActive, then accumulating.
} else if (active == leastActive) {
// Record the index of the least active invoker in leastIndexes order
leastIndexes[leastCount++] = i;
// Accumulate the total weight of the least active invoker
totalWeight += afterWarmup;
// If every invoker has the same weight?
if (sameWeight && afterWarmup != firstWeight) {
sameWeight = false;
}
}
}
// Choose an invoker from all the least active invokers
if (leastCount == 1) {
// If we got exactly one invoker having the least active value, return this invoker directly.
return invokers.get(leastIndexes[0]);
}
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 = ThreadLocalRandom.current().nextInt(totalWeight);
// Return a invoker based on the random value.
for (int i = 0; i < leastCount; i++) {
int leastIndex = leastIndexes[i];
offsetWeight -= weights[leastIndex];
if (offsetWeight < 0) {
return invokers.get(leastIndex);
}
}
}
// If all invokers have the same weight value or totalWeight=0, return evenly.
return invokers.get(leastIndexes[ThreadLocalRandom.current().nextInt(leastCount)]);
}2.4 一致性Hash策略-ConsistentHash LoadBalance
2.4.1 概述
要点：相同参数的请求总是发到同一服务提供者。
2.4.2 源码剖析
实现类是ConsistentHashLoadBalance，源码如下所示。
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;
// using the hashcode of list to compute the hash only pay attention to the elements in the list
int invokersHashCode = invokers.hashCode();
ConsistentHashSelector<T> selector = (ConsistentHashSelector<T>) selectors.get(key);
if (selector == null || selector.identityHashCode != invokersHashCode) {
selectors.put(key, new ConsistentHashSelector<T>(invokers, methodName, invokersHashCode));
selector = (ConsistentHashSelector<T>) selectors.get(key);
}
return selector.select(invocation);
}3 自定义负载均衡策略
3.1 自定义LoadBalance
创建一个继承 AbstractLoadBalance 的类，并重写 doSelect() 方法。举例如下。
public class MyLoadBalance extends AbstractLoadBalance {
@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
Invoker invoker = null;
// 自定义负载均衡算法，从invokers中选择一个Invoker
return invoker;
}
}3.2 配置和使用
在 resources 目录下, 添加 META-INF/dubbo 目录, 继而添加 org.apache.dubbo.rpc.cluster.LoadBalance 文件。并将自定义的 LoadBalance 类配置到该文件中。
myLoadBalance=org.apache.dubbo.rpc.cluster.loadbalance.MyLoadBalance然后在消费接口时指定使用自定义的策略。
<dubbo:reference ... loadbalance="myLoadBalance"/>备注：Dubbo自带的负载均衡策略配置如下所示
