dubbo——负载均衡

dubbo提供四种负载均衡策略：随机、轮询、最少活动、一致性hash

一、RandomLoadBalance——随机

    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        // Number of invokers
        int length = invokers.size();
        // Every invoker has the same weight?
        boolean sameWeight = true;
        // the weight of every invokers
        int[] weights = new int[length];
        // the first invoker's weight
        int firstWeight = getWeight(invokers.get(0), invocation);
        weights[0] = firstWeight;
        // The sum of weights
        int totalWeight = firstWeight;
        for (int i = 1; i < length; i++) {
            int weight = getWeight(invokers.get(i), invocation);
            // save for later use
            weights[i] = weight;
            // Sum
            totalWeight += weight;
            if (sameWeight && weight != firstWeight) {
                sameWeight = false;
            }
        }
        //有权重，按权重随机
        if (totalWeight > 0 && !sameWeight) {
            // 0——totalweight(不包含)中随机一个数
            int offset = ThreadLocalRandom.current().nextInt(totalWeight);
            // 返回随机数对应数组的invoker
            for (int i = 0; i < length; i++) {
                offset -= weights[i];
                if (offset < 0) {
                    return invokers.get(i);
                }
            }
        }
        // 所有节点权重相等或为0，从数组中随机返回一个invoker
        return invokers.get(ThreadLocalRandom.current().nextInt(length));
    }

总结：

随机负载均衡：根据每个节点权重，进行随机（使用ThreadLocalRandom保证线程安全），具体分为了两种情况：

1、每个节点权重相同，随机返回一个invoker。

2、权重不相同，根据总权重生成一个随机数，然后判断随机数所处区间，返回对应的invoker。

栗子：3个节点A、B、C权重分别为1、2 、3，取0-6（不包含）中一个随机数n，n-1<0位于A节点，否则n-1-2<0位于B节点，否则n-1-2-3<0位于C节点。

特点：少量请求，可能会发生倾斜，当请求变多时，趋向均衡。

二、RoundRobinLoadBalance——轮询

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();
    
    @Override
    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
        //建立容器，以方法为单位，记录每个节点的权重（支持动态权重），原子变量current(用于实现轮询)
        ConcurrentMap<String, WeightedRoundRobin> map = methodWeightMap.computeIfAbsent(key, k -> new ConcurrentHashMap<>());
        int totalWeight = 0;
        long maxCurrent = Long.MIN_VALUE;
        long now = System.currentTimeMillis();
        Invoker<T> selectedInvoker = null;
        WeightedRoundRobin selectedWRR = null;
        for (Invoker<T> invoker : invokers) {
            String identifyString = invoker.getUrl().toIdentityString();
            int weight = getWeight(invoker, invocation);
            WeightedRoundRobin weightedRoundRobin = map.get(identifyString);

            if (weightedRoundRobin == null) {
                weightedRoundRobin = new WeightedRoundRobin();
                weightedRoundRobin.setWeight(weight);
                map.putIfAbsent(identifyString, weightedRoundRobin);
                weightedRoundRobin = map.get(identifyString);
            }
            if (weight != weightedRoundRobin.getWeight()) {
                //weight changed
                weightedRoundRobin.setWeight(weight);
            }
            //current.addAndGet(weight) ， 选中current最大的节点，然后current.addAndGet(-totalWeight)
            //这里实现轮询的逻辑有点看不懂
            long cur = weightedRoundRobin.increaseCurrent();
            weightedRoundRobin.setLastUpdate(now);
            if (cur > maxCurrent) {
                maxCurrent = cur;
                selectedInvoker = invoker;
                selectedWRR = weightedRoundRobin;
            }
            totalWeight += weight;
        }
        if (!updateLock.get() && invokers.size() != map.size()) {
            if (updateLock.compareAndSet(false, true)) {
                try {
                    // 写时复制策略
                    ConcurrentMap<String, WeightedRoundRobin> newMap = new ConcurrentHashMap<>(map);
                    // 解决倾斜，invoker超过60s未调用，提高优先级
                    newMap.entrySet().removeIf(item -> now - item.getValue().getLastUpdate() > RECYCLE_PERIOD);
                    methodWeightMap.put(key, newMap);
                } finally {
                    updateLock.set(false);
                }
            }
        }
        if (selectedInvoker != null) {
            selectedWRR.sel(totalWeight);
            return selectedInvoker;
        }
        // should not happen here
        return invokers.get(0);
    }

新版的轮询逻辑有点看不懂：举个栗子，有三个节点，权重为1 2 3

000-->123(选中③后12-3)-->240(选中②后2-20)-->303(选中①后-303)-->-226(选中③后-220)-->-143(选中②后-1-23)-->006(选中③后000)之后循环。

调用顺序：③②①③②③，然后循环。

轮询模式存在响应慢的提供者会累积请求的问题。

三、LeastActiveLoadBalance——最少活跃

/* org.apache.dubbo.rpc.cluster.loadbalance.LeastActiveLoadBalance#doSelect */
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) {
        // Number of invokers
        int length = invokers.size();
        // The least active value of all invokers
        int leastActive = -1;
        // The number of invokers having the same least active value (leastActive)
        int leastCount = 0;
        // The index of invokers having the same least active value (leastActive)
        int[] leastIndexes = new int[length];
        // the weight of every invokers
        int[] weights = new int[length];
        // The sum of the warmup weights of all the least active invokers
        int totalWeight = 0;
        // The weight of the first least active invoker
        int firstWeight = 0;
        // Every least active invoker has the same weight value?
        boolean sameWeight = true;


        //筛选出最不活跃的节点，
        //给每个节点方法创建一个RpcStatus实例，用于记录节点方法活跃性Map<url,<methodName,rpcStatus>>
        //找到最小活跃的节点，将它的数组下标，放入数组leastIndexes中
        //leastIndexes大小为1时，最小活跃节点仅一个，直接返回
        //leastIndexes大小大于1时，最小活跃节点多个，然后用Random类似方法，从多个最小活跃节点中，随机返回一个节点
        //所有节点活跃性相同时，Random随机返回一个节点
        for (int i = 0; i < length; i++) {
            Invoker<T> invoker = invokers.get(i);
            // Get the active number of the invoker
            int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive();
            // Get the weight of the invoker's configuration. The default value is 100.
            int afterWarmup = getWeight(invoker, invocation);
            // save for later use
            weights[i] = afterWarmup;
            // If it is the first invoker or the active number of the invoker is less than the current least active number
            if (leastActive == -1 || active < leastActive) {
                // Reset the active number of the current invoker to the least active number
                leastActive = active;
                // Reset the number of least active invokers
                leastCount = 1;
                // Put the first least active invoker first in leastIndexes
                leastIndexes[0] = i;
                // Reset totalWeight
                totalWeight = afterWarmup;
                // Record the weight the first least active invoker
                firstWeight = afterWarmup;
                // Each invoke has the same weight (only one invoker here)
                sameWeight = true;
                // If current invoker's active value equals with leaseActive, then accumulating.
            } else if (active == leastActive) {
                // Record the index of the least active invoker in leastIndexes order
                leastIndexes[leastCount++] = i;
                // Accumulate the total weight of the least active invoker
                totalWeight += afterWarmup;
                // If every invoker has the same weight?
                if (sameWeight && i > 0
                        && afterWarmup != firstWeight) {
                    sameWeight = false;
                }
            }
        }
        // Choose an invoker from all the least active invokers
        if (leastCount == 1) {
            // If we got exactly one invoker having the least active value, return this invoker directly.
            return invokers.get(leastIndexes[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 = ThreadLocalRandom.current().nextInt(totalWeight);
            // Return a invoker based on the random value.
            for (int i = 0; i < leastCount; i++) {
                int leastIndex = leastIndexes[i];
                offsetWeight -= weights[leastIndex];
                if (offsetWeight < 0) {
                    return invokers.get(leastIndex);
                }
            }
        }
        // If all invokers have the same weight value or totalWeight=0, return evenly.
        return invokers.get(leastIndexes[ThreadLocalRandom.current().nextInt(leastCount)]);
    }

总结：

最少活跃负载均衡指的是响应慢的提供者收到更少的请求，如果活跃性相同，跟Random负载均衡一致。

活跃数指的是方法调用前后的计数差，是一个简单的计数器，调用前+1，调用后-1，当某节点响应慢时，单位时间-1比较慢，活跃数就比较大。这个时候会请求那些活跃数小的，响应快的应用。

四、ConsistentHashLoadBalance——一致性Hash

protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        String methodName = RpcUtils.getMethodName(invocation);
        //key = group+interface+version+methodName
        String key = invokers.get(0).getUrl().getServiceKey() + "." + methodName;
        // using the hashcode of list to compute the hash only pay attention to the elements in the list
        int invokersHashCode = invokers.hashCode();
        //为每个key创建一个选择器（实际就是一个方法对应一个选择器）
        //<group.interface.version.method,consistentHashSelector>
        ConsistentHashSelector<T> selector = (ConsistentHashSelector<T>) selectors.get(key);
        if (selector == null || selector.identityHashCode != invokersHashCode) {
            //① selector为空时创建
            //② 原语节点更改，导致invokers.hashCode变动，重新分配
            selectors.put(key, new ConsistentHashSelector<T>(invokers, methodName, invokersHashCode));
            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();
            //默认160个槽位，<dubbo:parameter key="hash.codes" value="160" />
            this.replicaNumber = url.getMethodParameter(methodName, HASH_NODES, 160);
            //默认只对方法第一个参数去hash，<dubbo:parameter key="hash.arguments" value="0" />
            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]);
            }
            //① 根据address+replica+h确定invoker的<key,invoker>，与其他invoker的属性无关，所以其他invoker挂掉，<key,invoker>不变
            //② 为了解决数据倾斜问题dubbo默认160个虚拟节点是每个invoker都有160个虚拟节点，即一致性hash上会有160*invokers.size个服务节点
            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());
            //求得参数的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();
        }

        private Invoker<T> selectForKey(long hash) {
            //一致性hash的实现：hash对应的下一个节点
            //例如TreeMap现在有节点  3、6、9
            //hash=1时取TreeMap.get(3)
            //hash=3时取TreeMap.get(3)
            //hash=4时取TreeMap.get(6)
            //dubbo会有160*invokers.size个服务节点(value对应实际的invoker)
            Map.Entry<Long, Invoker<T>> entry = virtualInvokers.ceilingEntry(hash);
            if (entry == null) {
                entry = virtualInvokers.firstEntry();
            }
            return entry.getValue();
        }

        //CRC24生成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加密：将参数转化一个byte数组
        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负载均衡：是将带有相同参数（默认方法的第一个参数）的请求总是发送给同一个提供者。当某台提供者挂掉时，原本发往该提供者的请求会基于虚拟节点（默认160个）平摊到其他提供者上（具体就是挂点虚拟节点的下一个节点），不会引起剧烈变动

五、预热处理——getWeight()

权重处理主要有一个机器预热处理：越热时间内，根据 运行时间/预热时间 的值控制权重。

/* org.apache.dubbo.rpc.cluster.loadbalance.AbstractLoadBalance */
    int getWeight(Invoker<?> invoker, Invocation invocation) {
        int weight;
        URL url = invoker.getUrl();
        // Multiple registry scenario, load balance among multiple registries.
        if (REGISTRY_SERVICE_REFERENCE_PATH.equals(url.getServiceInterface())) {
            weight = url.getParameter(REGISTRY_KEY + "." + WEIGHT_KEY, DEFAULT_WEIGHT);
        } else {
            weight = url.getMethodParameter(invocation.getMethodName(), WEIGHT_KEY, DEFAULT_WEIGHT);
            if (weight > 0) {
                long timestamp = invoker.getUrl().getParameter(TIMESTAMP_KEY, 0L);
                if (timestamp > 0L) {
                    long uptime = System.currentTimeMillis() - timestamp;
                    if (uptime < 0) {
                        return 1;
                    }
                    //获取预热时间，默认10分钟
                    int warmup = invoker.getUrl().getParameter(WARMUP_KEY, DEFAULT_WARMUP);
                    if (uptime > 0 && uptime < warmup) {
                        //预热时间内，降低权重，避免刚启动，请求负载导致启动失败
                        weight = calculateWarmupWeight((int)uptime, warmup, weight);
                    }
                }
            }
        }
        return Math.max(weight, 0);
    }

    static int calculateWarmupWeight(int uptime, int warmup, int weight) {
        int ww = (int) ( uptime / ((float) warmup / weight)); //uptime/warmup * weight,与uptime成正比
        return ww < 1 ? 1 : (Math.min(ww, weight));
    }

 六、负载均衡的配置方法

默认random，

<!-- 接口层面，下面配置一个就可以生效 -->
<dubbo:service interface="..." loadbalance="roundrobin" />
<dubbo:reference interface="..." loadbalance="roundrobin" />
<!-- 方法层面，下面配置一个就可以生效 -->
<dubbo:service interface="...">
    <dubbo:method name="..." loadbalance="roundrobin" />
</dubbo:service>
<dubbo:reference interface="...">
    <dubbo:method name="..." loadbalance="roundrobin" />
</dubbo:reference>