1. loadBalance的作用
它的职责是将网络请求，或者其他形式的负载“均摊”到不同的机器上。避免集群中部分服务器压力过大，而另一些服务器比较空闲的情况。通过负载均衡，可以让每台服务器获取到适合自己处理能力的负载
2. loadBalance的入口
入口是在AbstractClusterInvoker中，这个抽象类要上一篇中已经讲过，它有一个方法：
private Invoker<T> doSelect(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
if (invokers == null || invokers.isEmpty())
return null;
if (invokers.size() == 1)
return invokers.get(0);
// SPI选择loadbalance
if (loadbalance == null) {
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
}
// 负载均衡选择invoker
Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation);
//If the `invoker` is in the `selected` or invoker is unavailable && availablecheck is true, reselect.
if ((selected != null && selected.contains(invoker))
|| (!invoker.isAvailable() && getUrl() != null && availablecheck)) {
try {
Invoker<T> rinvoker = reselect(loadbalance, invocation, invokers, selected, availablecheck);
if (rinvoker != null) {
invoker = rinvoker;
} else {
//Check the index of current selected invoker, if it's not the last one, choose the one at index+1.
int index = invokers.indexOf(invoker);
try {
//Avoid collision
invoker = index < invokers.size() - 1 ? invokers.get(index + 1) : invokers.get(0);
} catch (Exception e) {
logger.warn(e.getMessage() + " may because invokers list dynamic change, ignore.", e);
}
}
} catch (Throwable t) {
logger.error("cluster reselect fail reason is :" + t.getMessage() + " if can not solve, you can set cluster.availablecheck=false in url", t);
}
}
return invoker;
}
3. loadBalance源码
AbstractLoadBalance抽象类的核心方法
public abstract class AbstractLoadBalance implements LoadBalance {
@Override
public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) {
// invokers列表如果为空，则直接返回空
if (invokers == null || invokers.isEmpty())
return null;
// 如果invokers列表只有1个，则返回第1个
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) {
// 获取权重 weight 默认是：100
int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT);
if (weight > 0) {
// 获取remote.timestamp
// 它表示：服务启动时的时间戳
long timestamp = invoker.getUrl().getParameter(Constants.REMOTE_TIMESTAMP_KEY, 0L);
if (timestamp > 0L) {
// 当前时间-服务启动时的时间戳
int uptime = (int) (System.currentTimeMillis() - timestamp);
// 获取服务预热时间，默认为10分钟
// warmup，10 * 60 * 1000
int warmup = invoker.getUrl().getParameter(Constants.WARMUP_KEY, Constants.DEFAULT_WARMUP);
// 如果服务启动时间还不足10分钟，则计算权重
if (uptime > 0 && uptime < warmup) {
weight = calculateWarmupWeight(uptime, warmup, weight);
}
}
}
return weight;
}
static int calculateWarmupWeight(int uptime, int warmup, int weight) {
// 当uptime越大的情况下，得到的ww值越大；
int ww = (int) ((float) uptime / ((float) warmup / (float) weight));
// ww如果大于weight，则返回weight，否则返回ww
return ww < 1 ? 1 : (ww > weight ? weight : ww);
}
3.1 RandomLoadBalance 加权随机算法
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?
for (int i = 0; i < length; i++) {
// 挨个获取每个invoker的权重；
int weight = getWeight(invokers.get(i), invocation);
// 累加到totalWeight里面；
totalWeight += weight; // Sum
// 权重发生改变的时候sameWeight=false
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.
// 生成一个小于totalWeight的随机值；
int offset = random.nextInt(totalWeight);
// Return a invoker based on the random value.
for (int i = 0; i < length; i++) {
// 随机值减去每个invoker的权重
offset -= getWeight(invokers.get(i), invocation);
// 当offset小于0时，返回当前invoker；
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));
}
}
3.2 LeastActiveLoadBalance 最小活跃数负载均衡
每个服务提供者对应一个活跃数 active。初始情况下，所有服务提供者活跃数均为0。每收到一个请求，活跃数加1，完成请求后则将活跃数减1。在服务运行一段时间后，性能好的服务提供者处理请求的速度更快，因此活跃数下降的也越快，此时这样的服务提供者能够优先获取到新的服务请求、这就是最小活跃数负载均衡算法的基本思想。除了最小活跃数，LeastActiveLoadBalance 在实现上还引入了权重值。
public class LeastActiveLoadBalance extends AbstractLoadBalance {
public static final String NAME = "leastactive";
private final Random random = new Random();
@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
// invokers 个数；
int length = invokers.size(); // Number of invokers
// 记录最小的活跃 invoker的活跃值；
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)
// 相同最小活跃值在invokers中的下标
int[] leastIndexs = new int[length]; // The index of invokers having the same least active value (leastActive)
// 相同最小活跃值的invoker权重累加；
int totalWeight = 0; // The sum of with warmup weights
// 当invoker最小活跃值一样时，判断权重是否一样，则需要记录最小活跃值的权重；
int firstWeight = 0; // Initial value, used for comparision
// false表示多个最小活值
boolean sameWeight = true; // Every invoker has the same weight value?
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
// 如果是第一个invoker或者当前invoker的活跃值小于了最小的invoker
if (leastActive == -1 || active < leastActive) { // Restart, when find a invoker having smaller least active value.
// 最小活跃值设置为当前invoker的活跃值
leastActive = active; // Record the current least active value
// 最小活跃数量设置；
leastCount = 1; // Reset leastCount, count again based on current leastCount
// 记录下标
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?
// 如果最小活跃值相同的话
} 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 += afterWarmup; // Add this invoker's weight to totalWeight.
// If every invoker has the same weight?
// 判断权重是否一样，如果不一样的话，下面的选择invoker不一样；
if (sameWeight && i > 0
&& afterWarmup != firstWeight) {
sameWeight = false;
}
}
}
// 如果只有一个最小活跃invoker的话，直接返回他；
if (leastCount == 1) {
// If we got exactly one invoker having the least active value, return this invoker directly.
return invokers.get(leastIndexs[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 = random.nextInt(totalWeight) + 1;
// Return a invoker based on the random value.
// 累减offsetWeight,当offsetWeight<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.
// 如果有多个最小活跃值相同的invoker，并且他们的权重都相同，则随机取一个；
return invokers.get(leastIndexs[random.nextInt(leastCount)]);
}
}
3.3 ConsistentHashLoadBalance 一致性 hash 算法
一致性hash负载是参考着一致性hash算法来的，大概意思就是在一个环，节点散落在这个环上，每次获取时，生成一个hash值，在这个节点上获取对应的节点；
为了让invoker更随机，一个invoker会生成很多虚拟的invoker散落在hash环上；
public class ConsistentHashLoadBalance extends AbstractLoadBalance {
public static final String NAME = "consistenthash";
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);
// 接口+方法名
String key = invokers.get(0).getUrl().getServiceKey() + "." + methodName;
// 如果invokers列表没有变化，那么可以将其缓存起来
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);
}
private static final class ConsistentHashSelector<T> {
private final TreeMap<Long, Invoker<T>> virtualInvokers;
private final int replicaNumber;
private final int identityHashCode;
private final int[] argumentIndex;
ConsistentHashSelector(List<Invoker<T>> invokers, String methodName, int identityHashCode) {
// 创建一个TreeMap
this.virtualInvokers = new TreeMap<Long, Invoker<T>>();
// 记录最新的invokers列表的hashCode值；
this.identityHashCode = identityHashCode;
// 获取invokers列表中的第一个invoker 的URL
URL url = invokers.get(0).getUrl();
// 获取 hash.nodes 属性，如果没有配置，则默认为160，表示每个invoker均分160个节点到hash环中；
this.replicaNumber = url.getMethodParameter(methodName, "hash.nodes", 160);
// 在select时，调用目标接口，目标接口时的，参数；没有配置默认为0；
// index=["0"]
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]);
}
// argumentIndex[0]=0
for (Invoker<T> invoker : invokers) {
// 获取ip+port的值
String address = invoker.getUrl().getAddress();
// 这里除4是因为 内部还有个循环，会循环4次；
for (int i = 0; i < replicaNumber / 4; i++) {
// 生成16位的md5值
byte[] digest = md5(address + i);
//
for (int h = 0; h < 4; h++) {
// 生成hash码
long m = hash(digest, h);
// 放入treeMap中；
virtualInvokers.put(m, invoker);
}
}
}
}
// 调用时
public Invoker<T> select(Invocation invocation) {
// 生成key
// invocation.getArguments()调用目标接口时，传入的参数；
// invocation.getArguments()只有参数值；不是参数名；
String key = toKey(invocation.getArguments());
// 生成md5
byte[] digest = md5(key);
// 生成hash，这里有个疑问，生成的hash是非常随机的，那怎么获取到对应的invoker呢？
// 也就是存入TreeMap的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();
}
private Invoker<T> selectForKey(long hash) {
// 返回在treeMap当前hash值之后map
// 如果返回为空，则表示生成的hash值过大，则选择treeMap中的第一个返回；
Map.Entry<Long, Invoker<T>> entry = virtualInvokers.ceilingEntry(hash);
if (entry == null) {
entry = virtualInvokers.firstEntry();
}
return entry.getValue();
}
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;
}
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();
}
}
}
3.4 RoundRobinLoadBalance 加权轮询负载均衡
dubbo的加权轮询算法有点复杂，普通的加权轮训都是搞个[A1,A2,A3,B1,C1 ]都是区间，然后循环这个区间，但这有个问题，就是连续的3个请求都请求到A节点，如果能再分散点是不是更好呢；
参考：http://www.manongjc.com/detail/50-ahzahzevmwfmvey.html文章来源：https://www.toymoban.com/news/detail-425748.html
4. loadBalance使用
@Service(loadbalance = "roundrobin",weight = 100,
methods = {@Method(name ="save",loadbalance = "roundrobin",timeout = 4000,retries = 5)})
@Reference(check = false,loadbalance = "roundrobin",cluster = "forking",
methods = {@Method(name="save",loadbalance = "roundrobin",timeout = 1000,retries = 5,async = true)})
Service和Reference都可以指定loadbalance属性，methods也可以指定，因为它都是在消费者调用时，对invokers列表做的一种选择目标invoker，但是weight只能在Service中指定，无法在methods中指定；文章来源地址https://www.toymoban.com/news/detail-425748.html
- random 加权随机，默认算法
- roundrobin 加权轮询
- leastactive 最少活跃调用数，慢的提供者会接收到更少的请求
- consistenthash 一致性哈希，相同参数请求同一提供者
到了这里，关于08 dubbo源码学习_LoadBalance的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
