dubbo2.6源码-负载均衡

秒速五厘米 2021-12-10 05:37 269阅读 0赞

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

<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 invokers
            int totalWeight = 0; // The sum of weights
            boolean sameWeight = true; // Every invoker has the same weight?
            for (int i = 0; i < length; i++) {
                int weight = getWeight(invokers.get(i), invocation);
                totalWeight += weight; // Sum
                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.
                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 invokers
            int maxWeight = 0; // The maximum weight
            int minWeight = Integer.MAX_VALUE; // The minimum weight
            final 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 weight
                minWeight = Math.min(minWeight, weight); // Choose the minimum weight
                if (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 robin
            return 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 invokers
            int leastActive = -1; // The least active value of all invokers
            int 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 weights
            int firstWeight = 0; // Initial value, used for comparision
            boolean 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 number
                int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT); // Weight
                if (leastActive == -1 || active < leastActive) { // Restart, when find a invoker having smaller least active value.
                    leastActive = active; // Record the current least active value
                    leastCount = 1; // Reset leastCount, count again based on current leastCount
                    leastIndexs[0] = i; // Reset
                    totalWeight = weight; // Reset
                    firstWeight = weight; // 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 += 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")
        @Override
        protected <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，相同参数的请求总是发到同一提供者。  
当某一台提供者挂时，原本发往该提供者的请求，基于虚拟节点，平摊到其它提供者，不会引起剧烈变动。