Dubbo负载均衡实现

概念

  dubbo的负载均衡,从本质上来说是客户端负载均衡，按照官网的文档说明，一共有四种负载均衡模式，缺省条件下为random。
策略(形貌为官网形貌)

  父类

org.apache.dubbo.rpc.cluster.loadbalance.AbstractLoadBalance
这里主要实现了一个通用的权重算法

(有一个预热的过程，服务刚启动过程蒙受的最终负载比力少，随着服务运行时间的增长，蒙受的负载徐徐逼近真实所需要蒙受的负载)

1.         /**     * Calculate the weight according to the uptime proportion of warmup time     * the new weight will be within 1(inclusive) to weight(inclusive)     *     * @param uptime the uptime in milliseconds     * @param warmup the warmup time in milliseconds     * @param weight the weight of an invoker     * @return weight which takes warmup into account     */    static int calculateWarmupWeight(int uptime, int warmup, int weight) {        int ww = (int) ( uptime / ((float) warmup / weight));        return ww < 1 ? 1 : (Math.min(ww, weight));    }        /**     * Get the weight of the invoker&#39;s invocation which takes warmup time into account     * if the uptime is within the warmup time, the weight will be reduce proportionally     *     * @param invoker    the invoker     * @param invocation the invocation of this invoker     * @return weight     */    int getWeight(Invoker invoker, Invocation invocation) {        int weight;        URL url = invoker.getUrl();        // 先获取服务的权重        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);    }

复制代码

各个实现

Random LoadBalance

• 随机，按权重设置随机概率。
  
• 在一个截面上碰撞的概率高，但调用量越大分布越匀称，而且按概率使用权重后也比力匀称，有利于动态调解提供者权重。
  

ex: 根据权重大小来生成概率区间，例如效果为 a,b,c对应的比例为1:2:3。那么随机区间为
[0,1),[1,3),[3,6)。
实现代码 org.apache.dubbo.rpc.cluster.loadbalance.RandomLoadBalance
比力简单的加权随机

1. @Override    protected  Invoker doSelect(List invokers, URL url, Invocation invocation) {        // Number of invokers        int length = invokers.size();        // Every invoker has the same weight?        boolean sameWeight = true;        // the maxWeight of every invokers, the minWeight = 0 or the maxWeight of the last invoker        int[] weights = new int[length];        // The sum of weights        int totalWeight = 0;        for (int i = 0; i < length; i++) {            int weight = getWeight(invokers.get(i), invocation);            // Sum            totalWeight += weight;            // save for later use            weights[i] = totalWeight;            if (sameWeight && totalWeight != weight * (i + 1)) {                sameWeight = false;            }        }        if (totalWeight > 0 && !sameWeight) {            // If (not every invoker has the same weight & at least one invoker&#39;s weight>0), select randomly based on totalWeight.            int offset = ThreadLocalRandom.current().nextInt(totalWeight);            // Return a invoker based on the random value.            for (int i = 0; i < length; i++) {                if (offset < weights[i]) {                    return invokers.get(i);                }            }        }        // If all invokers have the same weight value or totalWeight=0, return evenly.        return invokers.get(ThreadLocalRandom.current().nextInt(length));    }

复制代码

RoundRobin LoadBalance

• 轮询，按公约后的权重设置轮询比率。
  
• 存在慢的提供者累积请求的问题，好比：第二台呆板很慢，但没挂，当请求调到第二台时就卡在那，久而久之，所有请求都卡在调到第二台上。
  

1. @Override    protected  Invoker doSelect(List invokers, URL url, Invocation invocation) {        String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();        ConcurrentMap map = methodWeightMap.computeIfAbsent(key, k -> new ConcurrentHashMap());        int totalWeight = 0;        long maxCurrent = Long.MIN_VALUE;        long now = System.currentTimeMillis();        Invoker selectedInvoker = null;        WeightedRoundRobin selectedWRR = null;        for (Invoker invoker : invokers) {            String identifyString = invoker.getUrl().toIdentityString();            int weight = getWeight(invoker, invocation);            WeightedRoundRobin weightedRoundRobin = map.computeIfAbsent(identifyString, k -> {                WeightedRoundRobin wrr = new WeightedRoundRobin();                wrr.setWeight(weight);                return wrr;            });            if (weight != weightedRoundRobin.getWeight()) {                //weight changed                weightedRoundRobin.setWeight(weight);            }            long cur = weightedRoundRobin.increaseCurrent();            weightedRoundRobin.setLastUpdate(now);            if (cur > maxCurrent) {                maxCurrent = cur;                selectedInvoker = invoker;                selectedWRR = weightedRoundRobin;            }            totalWeight += weight;        }        if (invokers.size() != map.size()) {            map.entrySet().removeIf(item -> now - item.getValue().getLastUpdate() > RECYCLE_PERIOD);        }        if (selectedInvoker != null) {            selectedWRR.sel(totalWeight);            return selectedInvoker;        }        // should not happen here        return invokers.get(0);    }

复制代码

思路，在权重的最大公约数的次数中，按照权重比例进行轮询
ex: a,b,c三个相同provider，设置的权重为1:3:2，那么按照调用顺序6次内为 a,b,c,b,c,b
LeastActive LoadBalance

• 最少活泼调用数，相同活泼数的随机，活泼数指调用前后计数差。
  
• 使慢的提供者收到更少请求，因为越慢的提供者的调用前后计数差会越大。
  

实现代码 org.apache.dubbo.rpc.cluster.loadbalance.LeastActiveLoadBalance
[code]public class LeastActiveLoadBalance extends AbstractLoadBalance {    public static final String NAME = "leastactive";    private final Random random = new Random();    protected  Invoker doSelect(List 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; // 是否所有权重相同        for (int i = 0; i < length; i++) {            Invoker invoker = invokers.get(i);            int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive(); // 活泼数            int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT); // 权重            if (leastActive == -1 || active < leastActive) { // 发现更小的活泼数，重新开始                leastActive = active; // 纪录最小活泼数                leastCount = 1; // 重新统计相同最小活泼数的个数                leastIndexs[0] = i; // 重新纪录最小活泼数下标                totalWeight = weight; // 重新累计总权重                firstWeight = weight; // 纪录第一个权重                sameWeight = true; // 还原权重相同标识            } else if (active == leastActive) { // 累计相同最小的活泼数                leastIndexs[leastCount ++] = i; // 累计相同最小活泼数下标                totalWeight += weight; // 累计总权重                // 判定所有权重是否一样                if (sameWeight && i > 0                         && weight != firstWeight) {                    sameWeight = false;                }            }        }        // assert(leastCount > 0)        if (leastCount == 1) {            // 如果只有一个最小则直接返回            return invokers.get(leastIndexs[0]);        }        if (! sameWeight && totalWeight > 0) {            // 如果权重不相同且权重大于0则按总权重数随机            int offsetWeight = random.nextInt(totalWeight);            // 并确定随机值落在哪个片断上            for (int i = 0; i < leastCount; i++) {                int leastIndex = leastIndexs;                offsetWeight -= getWeight(invokers.get(leastIndex), invocation);                if (offsetWeight = 0 && i < args.length) {                    buf.append(args);                }            }            return buf.toString();        }        private Invoker selectForKey(long hash) {            //ceilingEntry(K key) 方法用来返回与该键至少大于或即是给定键，如果不存在这样的键的键 - 值映射，则返回null相关联。            Map.Entry 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)