Dubbo 负载均衡的实现

发表于 4年以前 | 总阅读数：370 次

前言

负载均衡是指在集群中，将多个数据请求分散在不同单元上进行执行，主要为了提高系统容错能力和加强系统对数据的处理能力。

在 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 ~ weight
return ww < 1 ? 1 : (ww > weight ? weight : ww);
    }
@Override
public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) {
// 校验 invokers 是否为空
if (CollectionUtils.isEmpty(invokers)) {
return null;
        }
// 当到达负载均衡流程时，invokers 中只有一个 Invoker 时，直接返回该 Invoker
if (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);
                }
            }
        }
// 保证最后返回的权重值不小于0
return 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 = false
if (sameWeight && weight != firstWeight) {
                sameWeight = false;
            }
        }
// 计算随机数取到的 Invoker，条件是必须总权重大于0，并且每个 Invoker 的权重都不一样
if (totalWeight > 0 && !sameWeight)
        {
// 基于 0~总数 范围内生成随机数
int offset = ThreadLocalRandom.current().nextInt(totalWeight );
// 计算随机数对应的 Invoker
for (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));
    }
}

以上就是加权随机策略的实现，这里比较主要关注计算生成的随机数对应的 Invoker。

通过遍历权重数组，生成的数累减当前权重值，当 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";
@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
// Invoker 数量
int length = invokers.size();
// 所有 Invoker 中的最小活跃值都是 -1
int 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为最小值）或小于leastActive
if (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 改为 false
if (sameWeight && i > 0
                        && afterWarmup != firstWeight) {
                    sameWeight = false;
                }
            }
        }
// 如果最小活跃数 Invoker 只有一个，直接返回该 Invoker
if (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;
    }
    @Override
protected <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);
                }
            }
        }
// 如果存在被选中的 Invoker
if (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")
@Override
protected <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 中实现。

private static final class ConsistentHashSelector<T> {
// 存储虚拟节点
private final TreeMap<Long, Invoker<T>> virtualInvokers;
// 节点数
private final int replicaNumber;
// invoker 列表的 hashcode，用来判断 Invoker 列表是否变化
private final int identityHashCode;
// 请求中用来作Hash映射的参数的索引
private final int[] argumentIndex;
    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();
// 获取节点数
this.replicaNumber = url.getMethodParameter(methodName, HASH_NODES, 160);
// 获取配置中的 参数索引
        String[] index = 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) {
// 获取 Invoker 中的地址，包括端口号
            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);
                }
            }
        }
    }
// 找出 Invoker
public Invoker<T> select(Invocation invocation) {
// 将参数转为字符串
        String key = toKey(invocation.getArguments());
// 字符串参数转换为 md5
byte[] digest = md5(key);
// 根据 md5 找出 Invoker
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();
    }
// 利用 md5 匹配到对应的 Invoker
private Invoker<T> selectForKey(long hash) {
// 找到第一个大于当前 hash 的 Invoker
        Map.Entry<Long, Invoker<T>> entry = virtualInvokers.ceilingEntry(hash);
if (entry == null) {
            entry = virtualInvokers.firstEntry();
        }
return entry.getValue();
    }
// hash 运算
private long hash(byte[] digest, int number) {
return (((long) (digest[3 + number * 4] & 0xFF) << 24)
                | ((long) (digest[2 + number * 4] & 0xFF) << 16)
                | ((long) (digest[1 + number * 4] & 0xFF) << 8)
                | (digest[number * 4] & 0xFF))
                & 0xFFFFFFFFL;
    }
// md5 运算
private byte[] md5(String value) {
        MessageDigest md5;
try {
            md5 = MessageDigest.getInstance("MD5");
        } catch (NoSuchAlgorithmException e) {
throw new IllegalStateException(e.getMessage(), e);
        }
        md5.reset();
byte[] bytes = value.getBytes(StandardCharsets.UTF_8);
        md5.update(bytes);
return md5.digest();
    }
}

一致 hash 实现过程就是先创建好虚拟节点，虚拟节点保存在 TreeMap 中。TreeMap 的 key 为配置的参数先进行 md5 运算，然后将 md5 值进行 hash 运算。TreeMap 的 value 为被选中的 Invoker。

最后请求时，计算参数的 hash 值，去从 TreeMap 中获取 Invoker。