Dubbo负载均衡包含4中算法
RandomLoadBalance 随机带权重（默认）
RoundRobinLoadBalance 轮询带权重
LeastActiveLoadBalance 最少活跃调用数
ConsistentHashLoadBalance 一致性Hash算法
AbstractLoadBalance 抽象类
外部调用入口，可根据Invoker获取权重，具有预热功能
public abstract class AbstractLoadBalance implements LoadBalance {
public AbstractLoadBalance() {
}
static int calculateWarmupWeight(int uptime, int warmup, int weight) {
// 提供者启动时间/(服务预热时间/服务权重) 比如 60sec/(600sec/10) = 1 每分钟增加1个权重 递进 类似慢启动
int ww = (int)((float)uptime / ((float)warmup / (float)weight));
return ww < 1 ? 1 : (ww > weight ? weight : ww);
}
// 供外部调用
public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) {
if (invokers != null && !invokers.isEmpty()) {
// 如果提供者只有1个 直接返回 否则执行对应loadBalance的doSelect方法
return invokers.size() == 1 ? (Invoker)invokers.get(0) : this.doSelect(invokers, url, invocation);
} else {
return null;
}
}
protected abstract <T> Invoker<T> doSelect(List<Invoker<T>> var1, URL var2, Invocation var3);
protected int getWeight(Invoker<?> invoker, Invocation invocation) {
// 获取方法的权重配置值 默认为100
int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), "weight", 100);
if (weight > 0) {
// 获取远程服务开始运行时间戳 默认为0
long timestamp = invoker.getUrl().getParameter("remote.timestamp", 0L);
if (timestamp > 0L) {
// 获取当前与远程开始提供服务的时间差
int uptime = (int)(System.currentTimeMillis() - timestamp);
// 获取方法预热时间 默认600sec 1min
int warmup = invoker.getUrl().getParameter("warmup", 600000);
if (uptime > 0 && uptime < warmup) {
// 如果服务没有预热完成
// 计算权重
weight = calculateWarmupWeight(uptime, warmup, weight);
}
}
}
// 权重为0 或者预热已完成
return weight;
}
}
RandomLoadBalance
比如 3个提供者abc，权重分别为3,2,1，则随机数范围0-5，如果随机数为5，则在c处，随机数被减至小于0，调用c
public class RandomLoadBalance extends AbstractLoadBalance {
public static final String NAME = "random";
private final Random random = new Random();
public RandomLoadBalance() {
}
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
// 提供者个数
int length = invokers.size();
// 总权重
int totalWeight = 0;
// 各提供者权重是否相同
boolean sameWeight = true;
int offset;
int i;
// 遍历提供者
for(offset = 0; offset < length; ++offset) {
// 获取当前提供者权重
i = this.getWeight((Invoker)invokers.get(offset), invocation);
// 总权重累加
totalWeight += i;
// 如果当前提供者权重不等于上一个权重
if (sameWeight && offset > 0 && i != this.getWeight((Invoker)invokers.get(offset - 1), invocation)) {
// 标记权重不是相同
sameWeight = false;
}
}
// 总权重大于0 并且个服务权重不相等
if (totalWeight > 0 && !sameWeight) {
// 生成指定范围0—总权重随机数
offset = this.random.nextInt(totalWeight);
// 遍历逐个提供者
for(i = 0; i < length; ++i) {
// 逐个权重做减法 直到offset小于0
offset -= this.getWeight((Invoker)invokers.get(i), invocation);
if (offset < 0) {
// 返回当前提供者
return (Invoker)invokers.get(i);
}
}
}
// 各提供者权重相等
return (Invoker)invokers.get(this.random.nextInt(length));
}
}
RoundRobinLoadBalance
比如 3个提供者abc，权重分别为3,2,1，则顺序为abcaba…
public class RoundRobinLoadBalance extends AbstractLoadBalance {
// 保存每一个方法的当前调用序号 自增
private final ConcurrentMap<String, AtomicPositiveInteger> sequences = new ConcurrentHashMap();
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
// 通过方法全路径定义key
String key = ((Invoker)invokers.get(0)).getUrl().getServiceKey() + "." + invocation.getMethodName();
// 服务提供者个数
int length = invokers.size();
// 定义最大和最小权重
int maxWeight = 0;
int minWeight = 2147483647;
// 保存每一个提供者的权重
LinkedHashMap<Invoker<T>, RoundRobinLoadBalance.IntegerWrapper> invokerToWeightMap = new LinkedHashMap();
// 总权重
int weightSum = 0;
int currentSequence;
for(int i = 0; i < length; ++i) {
// 获取当前提供者的权重
currentSequence = this.getWeight((Invoker)invokers.get(i), invocation);
// 比较并记录最大最小权重
maxWeight = Math.max(maxWeight, currentSequence);
minWeight = Math.min(minWeight, currentSequence);
if (currentSequence > 0) {
// 如果有权重 保存
invokerToWeightMap.put(invokers.get(i), new RoundRobinLoadBalance.IntegerWrapper(currentSequence));
// 总权重累加
weightSum += currentSequence;
}
}
// 获取当前方法的调用序号
AtomicPositiveInteger sequence = (AtomicPositiveInteger)this.sequences.get(key);
if (sequence == null) {
// 如果获取不到 新建并存入map
this.sequences.putIfAbsent(key, new AtomicPositiveInteger());
sequence = (AtomicPositiveInteger)this.sequences.get(key);
}
// 获取当前值 然后加1
currentSequence = sequence.getAndIncrement();
// 这个判断很有意义 如果最大权重大于0并且最大最小不相等才需要执行 如果权重都为0或者都相等就没必要执行了 可直接求余
if (maxWeight > 0 && minWeight < maxWeight) {
// 因为currentSequence是自增的 比如有5个提供者 权重和为8 序号一直自增 会大于8 我们对8求余 可判断出当前权重序列中的位置
int mod = currentSequence % weightSum;
for(int i = 0; i < maxWeight; ++i) { // 因为对总权重求余 最大为总权重
// 遍历所有的提供者
Iterator i$ = invokerToWeightMap.entrySet().iterator();
while(i$.hasNext()) {
// 获取当前遍历的提供者
Entry<Invoker<T>, RoundRobinLoadBalance.IntegerWrapper> each = (Entry)i$.next();
Invoker<T> k = (Invoker)each.getKey();
RoundRobinLoadBalance.IntegerWrapper v = (RoundRobinLoadBalance.IntegerWrapper)each.getValue();
// 如果当前权重序号被减到0 并且当前提供者权重没有被减到0 直接返回当前提供者
if (mod == 0 && v.getValue() > 0) {
return k;
}
// 正常情况下mod不会小于0的
// 把当前权重缩减1 继续下一个提供者
if (v.getValue() > 0) {
v.decrement();
--mod;
}
}
}
}
// 如果权重都为0或者都相等就没必要执行了 可直接求余
return (Invoker)invokers.get(currentSequence % length);
}
}
LeastActiveLoadBalance
先遍历所有提供者，获取最小活跃数的列表
如果只有1个直接返回，如果有多个随机选一个返回
此处随机和RandomLoadBalance的判断不一样，感觉有问题，后面再研究
public class LeastActiveLoadBalance extends AbstractLoadBalance {
public static final String NAME = "leastactive";
private final Random random = new Random();
public LeastActiveLoadBalance() {
}
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
// 服务提供者个数
int length = invokers.size();
// 记录最小活跃值
int leastActive = -1;
// 最小活跃值相同的个数
int leastCount = 0;
// 数组记录对应的提供者列表序号
int[] leastIndexs = new int[length];
int totalWeight = 0;
int firstWeight = 0;
// 各提供者权重是否相同
boolean sameWeight = true;
int offsetWeight;
int leastIndex;
// 遍历每一个提供者
for(offsetWeight = 0; offsetWeight < length; ++offsetWeight) {
Invoker<T> invoker = (Invoker)invokers.get(offsetWeight);
// 获取当前提供者状态
leastIndex = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive();
// 获取提供者权重
int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), "weight", 100);
if (leastActive != -1 && leastIndex >= leastActive) {
if (leastIndex == leastActive) { // 如果当前和最小活跃值相等
leastIndexs[leastCount++] = offsetWeight; // 在数组内记录提供者列表序号
// 最小活跃服务总权重累加
totalWeight += weight;
// 最小活跃服务权重不相同
if (sameWeight && offsetWeight > 0 && weight != firstWeight) {
sameWeight = false;
}
}
} else { // 如果开始遍历 或者发现当前提供者活跃值小于最小值 重置最小相关参数
leastActive = leastIndex;
leastCount = 1;
leastIndexs[0] = offsetWeight;
totalWeight = weight;
firstWeight = weight;
sameWeight = true;
}
}
// 如果最小活跃的提供者只有1个 直接返回
if (leastCount == 1) {
return (Invoker)invokers.get(leastIndexs[0]);
} else {
// 如果最少活跃各权重不相同
if (!sameWeight && totalWeight > 0) {
// 生成0-权重总和随机数
offsetWeight = this.random.nextInt(totalWeight);
// 遍历数组至最小活跃数个数
for(int i = 0; i < leastCount; ++i) {
leastIndex = leastIndexs[i];
// 随机数递减
offsetWeight -= this.getWeight((Invoker)invokers.get(leastIndex), invocation);
// 当随机数被减至0 返回提供者
if (offsetWeight <= 0) { // 此处和RandomLoadBalance的判断不一样 感觉这里有问题 如果3个提供者abc 权重为3,2,1 生成随机数为0-5 那c服务好像一直调不到
return (Invoker)invokers.get(leastIndex);
}
}
}
// 如果权重都相等 随机返回一个提供者
return (Invoker)invokers.get(leastIndexs[this.random.nextInt(leastCount)]);
}
}
}
ConsistentHashLoadBalance
一致性hash算法，就是根据key值生成0-2^32-1范围内的hash值，将节点散列到圆环上，获取时根据key的hash值，找到第一个大于该hash值的节点，如果找不到返回第一个。此处有两个参数 hash.nodes节点个数（默认160） hash.arguments参与hash参数（默认0）。
一致性hash算法目的在于将参数相同或者类似的请求分发到同一服务上，比如分布式会话，将同一用户的所有请求分发到同一设备上，就不会出现多服务会话不同步的情况。
public class ConsistentHashLoadBalance extends AbstractLoadBalance {
private final ConcurrentMap<String, ConsistentHashLoadBalance.ConsistentHashSelector<?>> selectors = new ConcurrentHashMap();
public ConsistentHashLoadBalance() {
}
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
// 获取方法名称
String methodName = RpcUtils.getMethodName(invocation);
// key为方法全路径名
String key = ((Invoker)invokers.get(0)).getUrl().getServiceKey() + "." + methodName;
// 获取hashCode
int identityHashCode = System.identityHashCode(invokers);
// 判断是否已经缓存
ConsistentHashLoadBalance.ConsistentHashSelector<T> selector = (ConsistentHashLoadBalance.ConsistentHashSelector)this.selectors.get(key);
if (selector == null || selector.identityHashCode != identityHashCode) {
// 添加缓存
this.selectors.put(key, new ConsistentHashLoadBalance.ConsistentHashSelector(invokers, methodName, identityHashCode));
selector = (ConsistentHashLoadBalance.ConsistentHashSelector)this.selectors.get(key);
}
return selector.select(invocation);
}
private static final class ConsistentHashSelector<T> {
private final TreeMap<Long, Invoker<T>> virtualInvokers = new TreeMap();
private final int replicaNumber;
private final int identityHashCode;
private final int[] argumentIndex;
ConsistentHashSelector(List<Invoker<T>> invokers, String methodName, int identityHashCode) {
this.identityHashCode = identityHashCode;
URL url = ((Invoker)invokers.get(0)).getUrl();
// 设置提供者节点 默认160
this.replicaNumber = url.getMethodParameter(methodName, "hash.nodes", 160);
// 指定hash参数 默认为第一个参数 参数相同的请求会被分发到同一个服务上去
String[] index = Constants.COMMA_SPLIT_PATTERN.split(url.getMethodParameter(methodName, "hash.arguments", "0"));
this.argumentIndex = new int[index.length];
for(int i = 0; i < index.length; ++i) {
this.argumentIndex[i] = Integer.parseInt(index[i]);
}
Iterator i$ = invokers.iterator();
while(i$.hasNext()) {
Invoker<T> invoker = (Invoker)i$.next();
String address = invoker.getUrl().getAddress();
for(int i = 0; i < this.replicaNumber / 4; ++i) {
byte[] digest = this.md5(address + i);
for(int h = 0; h < 4; ++h) {
long m = this.hash(digest, h);
this.virtualInvokers.put(m, invoker);
}
}
}
}
public Invoker<T> select(Invocation invocation) {
String key = this.toKey(invocation.getArguments());
byte[] digest = this.md5(key);
return this.selectForKey(this.hash(digest, 0));
}
// 根据各个参数生成key
private String toKey(Object[] args) {
StringBuilder buf = new StringBuilder();
int[] arr$ = this.argumentIndex;
int len$ = arr$.length;
for(int i$ = 0; i$ < len$; ++i$) {
int i = arr$[i$];
if (i >= 0 && i < args.length) {
buf.append(args[i]);
}
}
return buf.toString();
}
private Invoker<T> selectForKey(long hash) {
// 从treeMap中查找
Entry<Long, Invoker<T>> entry = this.virtualInvokers.tailMap(hash, true).firstEntry();
if (entry == null) {
// 如果未找到有效的提供者 返回第一个
entry = this.virtualInvokers.firstEntry();
}
return (Invoker)entry.getValue();
}
// 计算hash值 返回有32位 如果是int会有负数 所以转成long
private long hash(byte[] digest, int number) {
return ((long)(digest[3 + number * 4] & 255) << 24 | (long)(digest[2 + number * 4] & 255) << 16 | (long)(digest[1 + number * 4] & 255) << 8 | (long)(digest[number * 4] & 255)) & 4294967295L;
}
// 求md5值
private byte[] md5(String value) {
MessageDigest md5;
try {
md5 = MessageDigest.getInstance("MD5");
} catch (NoSuchAlgorithmException var6) {
throw new IllegalStateException(var6.getMessage(), var6);
}
md5.reset();
byte[] bytes;
try {
bytes = value.getBytes("UTF-8");
} catch (UnsupportedEncodingException var5) {
throw new IllegalStateException(var5.getMessage(), var5);
}
md5.update(bytes);
return md5.digest();
}
}
}
版权声明：本文为guandongsheng110原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接和本声明。