dubbo2.6源码-负载均衡

生产者集群启动时，才会使用负载均衡，否则跳过

    <!--use dubbo protocol to export service on port 20880修改端口，即可启动集群，此时负载均衡才会使用--><dubbo:protocol name="dubbo" port="20881"/>

LoadBalance

@SPI(RandomLoadBalance.NAME)
public interface LoadBalance {/*** select one invoker in list.** @param invokers   invokers.* @param url        refer url* @param invocation invocation.* @return selected invoker.*/@Adaptive("loadbalance")<T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException;}

AbstractLoadBalance

public abstract class AbstractLoadBalance implements LoadBalance {static int calculateWarmupWeight(int uptime, int warmup, int weight) {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())return null;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) {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);int warmup = invoker.getUrl().getParameter(Constants.WARMUP_KEY, Constants.DEFAULT_WARMUP);if (uptime > 0 && uptime < warmup) {weight = calculateWarmupWeight(uptime, warmup, weight);}}}return weight;}}

四种负载均衡策略

RandomLoadBalance随机加权，默认

public class RandomLoadBalance extends AbstractLoadBalance {public static final String NAME = "random";private final Random random = new Random();protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {int length = invokers.size(); // Number of invokersint totalWeight = 0; // The sum of weightsboolean sameWeight = true; // Every invoker has the same weight?for (int i = 0; i < length; i++) {int weight = getWeight(invokers.get(i), invocation);totalWeight += weight; // Sumif (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.int offset = random.nextInt(totalWeight);// Return a invoker based on the random value.for (int i = 0; i < length; i++) {offset -= getWeight(invokers.get(i), invocation);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));}

生产者改变权重

<dubbo:provider delay="-1" timeout="1000000" retries="0" weight="80"/>

默认权重值时100，

我们现在假设集群有四个节点分别对应的权重为｛A:1,B:2,C:3,D:4｝，分别将权重套入到代码中进行分析，该随机算法按总权重进行加权随机，A节点负载请求的概率为1/(1+2+3+4)，依次类推，B，C，D负载的请求概率分别是20%，30%，40%。在这种方式下，用户可以根据机器的实际性能动态调整权重比率，如果发现机器D负载过大，请求堆积过多，通过调整权重可以缓解机器D处理请求的压力。

轮询加权

修改负载均衡策略

    <dubbo:reference id="userService" check="false" interface="com.study.dubbo.userapi.service.UserService" loadbalance="roundrobin"/>

消费者添加 loadbalance=“roundrobin”，就是类中Name常量

RoundRobinLoadBalance

public class RoundRobinLoadBalance extends AbstractLoadBalance {public static final String NAME = "roundrobin";private final ConcurrentMap<String, AtomicPositiveInteger> sequences = new ConcurrentHashMap<String, AtomicPositiveInteger>();protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();int length = invokers.size(); // Number of invokersint maxWeight = 0; // The maximum weightint minWeight = Integer.MAX_VALUE; // The minimum weightfinal LinkedHashMap<Invoker<T>, IntegerWrapper> invokerToWeightMap = new LinkedHashMap<Invoker<T>, IntegerWrapper>();int weightSum = 0;for (int i = 0; i < length; i++) {int weight = getWeight(invokers.get(i), invocation);maxWeight = Math.max(maxWeight, weight); // Choose the maximum weightminWeight = Math.min(minWeight, weight); // Choose the minimum weightif (weight > 0) {invokerToWeightMap.put(invokers.get(i), new IntegerWrapper(weight));weightSum += weight;}}AtomicPositiveInteger sequence = sequences.get(key);if (sequence == null) {sequences.putIfAbsent(key, new AtomicPositiveInteger());sequence = sequences.get(key);}int currentSequence = sequence.getAndIncrement();if (maxWeight > 0 && minWeight < maxWeight) {int mod = currentSequence % weightSum;for (int i = 0; i < maxWeight; i++) {for (Map.Entry<Invoker<T>, IntegerWrapper> each : invokerToWeightMap.entrySet()) {final Invoker<T> k = each.getKey();final IntegerWrapper v = each.getValue();if (mod == 0 && v.getValue() > 0) {return k;}if (v.getValue() > 0) {v.decrement();mod--;}}}}// Round robinreturn invokers.get(currentSequence % length);}private static final class IntegerWrapper {private int value;public IntegerWrapper(int value) {this.value = value;}public int getValue() {return value;}public void setValue(int value) {this.value = value;}public void decrement() {this.value--;}}}

最少调用

LeastActiveLoadBalance

public class LeastActiveLoadBalance extends AbstractLoadBalance {public static final String NAME = "leastactive";private final Random random = new Random();protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {int length = invokers.size(); // Number of invokersint leastActive = -1; // The least active value of all invokersint leastCount = 0; // The number of invokers having the same least active value (leastActive)int[] leastIndexs = new int[length]; // The index of invokers having the same least active value (leastActive)int totalWeight = 0; // The sum of weightsint firstWeight = 0; // Initial value, used for comparisionboolean sameWeight = true; // Every invoker has the same weight value?for (int i = 0; i < length; i++) {Invoker<T> invoker = invokers.get(i);int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive(); // Active numberint weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT); // Weightif (leastActive == -1 || active < leastActive) { // Restart, when find a invoker having smaller least active value.leastActive = active; // Record the current least active valueleastCount = 1; // Reset leastCount, count again based on current leastCountleastIndexs[0] = i; // ResettotalWeight = weight; // ResetfirstWeight = weight; // Record the weight the first invokersameWeight = 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 invokertotalWeight += weight; // Add this invoker's weight to totalWeight.// If every invoker has the same weight?if (sameWeight && i > 0&& weight != firstWeight) {sameWeight = false;}}}// assert(leastCount > 0)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);// Return a invoker based on the random value.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.return invokers.get(leastIndexs[random.nextInt(leastCount)]);}
}

一致hash

使用consistenthash

<dubbo:reference id="userService" check="false" interface="com.study.dubbo.userapi.service.UserService" loadbalance="consistenthash"/>

public class ConsistentHashLoadBalance extends AbstractLoadBalance {private final ConcurrentMap<String, ConsistentHashSelector<?>> selectors = new ConcurrentHashMap<String, ConsistentHashSelector<?>>();@SuppressWarnings("unchecked")@Overrideprotected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();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, invocation.getMethodName(), 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) {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 = 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]);}for (Invoker<T> invoker : invokers) {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);}}}}public Invoker<T> select(Invocation invocation) {String key = toKey(invocation.getArguments());byte[] digest = md5(key);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) {Map.Entry<Long, Invoker<T>> entry = virtualInvokers.tailMap(hash, true).firstEntry();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;try {bytes = value.getBytes("UTF-8");} catch (UnsupportedEncodingException e) {throw new IllegalStateException(e.getMessage(), e);}md5.update(bytes);return md5.digest();}}}

构造函数中，每个实际的提供者均有160个（默认值，可调整）虚拟节点，每个提供者对应的虚拟节点将平均散列到哈希环上，当有请求时，先计算该请求参数对应的哈希值，然后顺时针寻找最近的虚拟节点，得到实际的提供者节点。

ConsistentHashLoadBalance: 一致性Hash，相同参数的请求总是发到同一提供者。
当某一台提供者挂时，原本发往该提供者的请求，基于虚拟节点，平摊到其它提供者，不会引起剧烈变动。