Dubbo源码(八) - 负载均衡 - ToB企服应用市场:ToB评测及商务社交产业平台

public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) {
if (invokers == null || invokers.isEmpty())
return null;
// 如果 invokers 列表中仅有一个 Invoker，直接返回即可，无需进行负载均衡
if (invokers.size() == 1)
return invokers.get(0);
// 调用 doSelect 方法进行负载均衡，该方法为抽象方法，由子类实现
return doSelect(invokers, url, invocation);
}

复制代码

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) {
// 计算权重，下面代码逻辑上形似于 (uptime / warmup) * weight。
// 随着服务运行时间 uptime 增大，权重计算值 ww 会慢慢接近配置值 weight
int ww = (int) ((float) uptime / ((float) warmup / (float) weight));
return ww < 1 ? 1 : (ww > weight ? weight : ww);
}

复制代码

public class RandomLoadBalance extends AbstractLoadBalance {
public static final String NAME = "random";
private final Random random = new Random();
@Override
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?
// 下面这个循环有两个作用，第一是计算总权重 totalWeight，
// 第二是检测每个服务提供者的权重是否相同
for (int i = 0; i < length; i++) {
int weight = getWeight(invokers.get(i), invocation);
totalWeight += weight; // Sum
// 检测当前服务提供者的权重与上一个服务提供者的权重是否相同，
// 不相同的话，则将 sameWeight 置为 false。
if (sameWeight && i > 0
&& weight != getWeight(invokers.get(i - 1), invocation)) {
sameWeight = false;
}
}
if (totalWeight > 0 && !sameWeight) {
// 随机获取一个 [0, totalWeight) 区间内的数字
int offset = random.nextInt(totalWeight);
// Return a invoker based on the random value.
// 循环让 offset 数减去服务提供者权重值，当 offset 小于0时，返回相应的 Invoker。
// 举例说明一下，我们有 servers = [A, B, C]，weights = [5, 3, 2]，offset = 7。
// 第一次循环，offset - 5 = 2 > 0，即 offset > 5，
// 表明其不会落在服务器 A 对应的区间上。
// 第二次循环，offset - 3 = -1 < 0，即 5 < offset < 8，
// 表明其会落在服务器 B 对应的区间上
for (int i = 0; i < length; i++) {
// 让随机值 offset 减去权重值
offset -= getWeight(invokers.get(i), invocation);
if (offset < 0) {
// 返回相应的 Invoker
return invokers.get(i);
}
}
}
// 如果所有服务提供者权重值相同，此时直接随机返回一个即可
return invokers.get(random.nextInt(length));
}
}

复制代码

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
// 具有相同“最小活跃数”的服务者提供者（以下用 Invoker 代称）数量
int leastCount = 0; // The number of invokers having the same least active value (leastActive)
// leastIndexs 用于记录具有相同“最小活跃数”的 Invoker 在 invokers 列表中的下标信息
int[] leastIndexs = new int[length]; // The index of invokers having the same least active value (leastActive)
int totalWeight = 0; // The sum of with warmup weights
// 第一个最小活跃数的 Invoker 权重值，用于与其他具有相同最小活跃数的 Invoker 的权重进行对比，
// 以检测是否“所有具有相同最小活跃数的 Invoker 的权重”均相等
int firstWeight = 0; // Initial value, used for comparision
boolean sameWeight = true; // Every invoker has the same weight value?
// 遍历 invokers 列表
for (int i = 0; i < length; i++) {
Invoker<T> invoker = invokers.get(i);
// 获取 Invoker 对应的活跃数
int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive(); // Active number
// 获取权重
int afterWarmup = getWeight(invoker, invocation); // Weight
// 发现更小的活跃数，重新开始
if (leastActive == -1 || active < leastActive) { // Restart, when find a invoker having smaller least active value.
// 使用当前活跃数 active 更新最小活跃数 leastActive
leastActive = active; // Record the current least active value
// 更新 leastCount 为 1
leastCount = 1; // Reset leastCount, count again based on current leastCount
// 记录当前下标值到 leastIndexs 中
leastIndexs[0] = i; // Reset
totalWeight = afterWarmup; // Reset
firstWeight = afterWarmup; // Record the weight the first invoker
sameWeight = true; // Reset, every invoker has the same weight value?
// 当前 Invoker 的活跃数 active 与最小活跃数 leastActive 相同
} else if (active == leastActive) { // If current invoker's active value equals with leaseActive, then accumulating.
// 在 leastIndexs 中记录下当前 Invoker 在 invokers 集合中的下标
leastIndexs[leastCount++] = i; // Record index number of this invoker
// 累加权重
totalWeight += afterWarmup; // Add this invoker's weight to totalWeight.
// If every invoker has the same weight?
// 检测当前 Invoker 的权重与 firstWeight 是否相等，
// 不相等则将 sameWeight 置为 false
if (sameWeight && i > 0
&& afterWarmup != firstWeight) {
sameWeight = false;
}
}
}
// assert(leastCount > 0)
// 当只有一个 Invoker 具有最小活跃数，此时直接返回该 Invoker 即可
if (leastCount == 1) {
// If we got exactly one invoker having the least active value, return this invoker directly.
return invokers.get(leastIndexs[0]);
}
// 有多个 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 = random.nextInt(totalWeight) + 1;
// Return a invoker based on the random value.
// 循环让随机数减去具有最小活跃数的 Invoker 的权重值，
// 当 offset 小于等于0时，返回相应的 Invoker
for (int i = 0; i < leastCount; i++) {
int leastIndex = leastIndexs[i];
// 获取权重值，并让随机数减去权重值
offsetWeight -= getWeight(invokers.get(leastIndex), invocation);
if (offsetWeight <= 0)
return invokers.get(leastIndex);
}
}
// If all invokers have the same weight value or totalWeight=0, return evenly.
// 如果权重相同或权重为0时，随机返回一个 Invoker
return invokers.get(leastIndexs[random.nextInt(leastCount)]);
}

复制代码

@Activate(group = Constants.CONSUMER, value = Constants.ACTIVES_KEY)
public class ActiveLimitFilter implements Filter

复制代码

<dubbo:service interface="com.alibaba.dubbo.demo.DemoService" ref="demoService" registry="remoteRegistry" actives="5" />
<dubbo:reference id="demoService" interface="com.alibaba.dubbo.demo.DemoService" loadbalance="leastactive" actives="5" />

复制代码

<dubbo:reference id="demoService" interface="com.alibaba.dubbo.demo.DemoService" filter="activelimit" />

复制代码