Dubbo：負(fù)載均衡實(shí)現(xiàn)解析

<!-- 服務(wù)端服務(wù)級(jí)別 -->    <dubbo:service interface="..." loadbalance="roundrobin" />    <!-- 客戶端服務(wù)級(jí)別 -->    <dubbo:reference interface="..." loadbalance="roundrobin" />    <!-- 服務(wù)端方法級(jí)別 -->    <dubbo:service interface="...">        <dubbo:method name="hello" loadbalance="roundrobin"/>    </dubbo:service>    <!-- 客戶端方法級(jí)別 -->    <dubbo:reference interface="...">        <dubbo:method name="hello" loadbalance="roundrobin"/>    </dubbo:reference>

// 調(diào)用提供者服務(wù)入口    // org.apache.dubbo.rpc.cluster.support.AbstractClusterInvoker#invoke    @Override    public Result invoke(final Invocation invocation) throws RpcException {        checkWhetherDestroyed();
        // binding attachments into invocation.        Map<String, Object> contextAttachments = RpcContext.getContext().getAttachments();        if (contextAttachments != null && contextAttachments.size() != 0) {            ((RpcInvocation) invocation).addAttachments(contextAttachments);        }
        List<Invoker<T>> invokers = list(invocation);        // 根據(jù)可用的提供者列表和要調(diào)用的方法，決定選取的負(fù)載均衡器        LoadBalance loadbalance = initLoadBalance(invokers, invocation);        RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);        return doInvoke(invocation, invokers, loadbalance);    }
    // 實(shí)例化一個(gè)負(fù)載均衡器，以備后續(xù)使用    // org.apache.dubbo.rpc.cluster.support.AbstractClusterInvoker#initLoadBalance    /**     * Init LoadBalance.     * <p>     * if invokers is not empty, init from the first invoke's url and invocation     * if invokes is empty, init a default LoadBalance(RandomLoadBalance)     * </p>     *     * @param invokers   invokers     * @param invocation invocation     * @return LoadBalance instance. if not need init, return null.     */    protected LoadBalance initLoadBalance(List<Invoker<T>> invokers, Invocation invocation) {        if (CollectionUtils.isNotEmpty(invokers)) {            // 從provider 的 url 地址中取出 loadbalance=xxx 配置，如果沒(méi)有仍使用 random 策略            return ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()                    .getMethodParameter(RpcUtils.getMethodName(invocation), LOADBALANCE_KEY, DEFAULT_LOADBALANCE));        } else {            // 默認(rèn)是 random 策略            return ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(DEFAULT_LOADBALANCE);        }    }

// org.apache.dubbo.rpc.cluster.loadbalance.AbstractLoadBalance#select    @Override    public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) {        // 沒(méi)有可用的提供者，沒(méi)得選        if (CollectionUtils.isEmpty(invokers)) {            return null;        }        // 只有一個(gè)提供者，沒(méi)得均衡的，直接用        if (invokers.size() == 1) {            return invokers.get(0);        }        // 然后就是各自均衡算法的實(shí)現(xiàn)了        return doSelect(invokers, url, invocation);    }

/** * This class select one provider from multiple providers randomly. * You can define weights for each provider: * If the weights are all the same then it will use random.nextInt(number of invokers). * If the weights are different then it will use random.nextInt(w1 + w2 + ... + wn) * Note that if the performance of the machine is better than others, you can set a larger weight. * If the performance is not so good, you can set a smaller weight. */public class RandomLoadBalance extends AbstractLoadBalance {    // 標(biāo)識(shí)自身    public static final String NAME = "random";
    /**     * Select one invoker between a list using a random criteria     * @param invokers List of possible invokers     * @param url URL     * @param invocation Invocation     * @param <T>     * @return The selected invoker     */    @Override    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;            // 計(jì)算出所有權(quán)重和，以便在進(jìn)行隨機(jī)時(shí)設(shè)定范圍            totalWeight += weight;            if (sameWeight && weight != firstWeight) {                sameWeight = false;            }        }        // 針對(duì)各提供供者權(quán)重不一的情況，則找到第一個(gè)大于隨機(jī)數(shù)的提供者即可        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 = ThreadLocalRandom.current().nextInt(totalWeight);            // Return a invoker based on the random value.            for (int i = 0; i < length; i++) {                offset -= weights[i];                if (offset < 0) {                    return invokers.get(i);                }            }        }        // If all invokers have the same weight value or totalWeight=0, return evenly.        // 如果大家權(quán)重都一樣，則直接以個(gè)數(shù)進(jìn)行隨機(jī)即可得到提供者        return invokers.get(ThreadLocalRandom.current().nextInt(length));    }
}

/** * Round robin load balance. */public class RoundRobinLoadBalance extends AbstractLoadBalance {    // 自身標(biāo)識(shí)    public static final String NAME = "roundrobin";
    private static final int RECYCLE_PERIOD = 60000;
    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();
    /**     * get invoker addr list cached for specified invocation     * <p>     * <b>for unit test only</b>     *      * @param invokers     * @param invocation     * @return     */    protected <T> Collection<String> getInvokerAddrList(List<Invoker<T>> invokers, Invocation invocation) {        String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();        Map<String, WeightedRoundRobin> map = methodWeightMap.get(key);        if (map != null) {            return map.keySet();        }        return null;    }
    @Override    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {        String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();        ConcurrentMap<String, WeightedRoundRobin> map = methodWeightMap.get(key);        if (map == null) {            methodWeightMap.putIfAbsent(key, new ConcurrentHashMap<String, WeightedRoundRobin>());            map = methodWeightMap.get(key);        }        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();            WeightedRoundRobin weightedRoundRobin = map.get(identifyString);            int weight = getWeight(invoker, invocation);
            if (weightedRoundRobin == null) {                weightedRoundRobin = new WeightedRoundRobin();                weightedRoundRobin.setWeight(weight);                map.putIfAbsent(identifyString, weightedRoundRobin);            }            if (weight != weightedRoundRobin.getWeight()) {                //weight changed                weightedRoundRobin.setWeight(weight);            }            // 自增權(quán)重            long cur = weightedRoundRobin.increaseCurrent();            weightedRoundRobin.setLastUpdate(now);            // 獲取最大權(quán)重項(xiàng)，并以對(duì)應(yīng)的 invoker 作為本次選擇的實(shí)例            if (cur > maxCurrent) {                maxCurrent = cur;                selectedInvoker = invoker;                selectedWRR = weightedRoundRobin;            }            totalWeight += weight;        }        // 當(dāng)invoker數(shù)量發(fā)生變化時(shí)，需要能感知到，以便清理 map, 避免內(nèi)存泄露        if (!updateLock.get() && invokers.size() != map.size()) {            if (updateLock.compareAndSet(false, true)) {                try {                    // copy -> modify -> update reference                    // 超出計(jì)數(shù)周期，則清空原來(lái)的 WeightedRoundRobin                    ConcurrentMap<String, WeightedRoundRobin> newMap = new ConcurrentHashMap<>(map);                    newMap.entrySet().removeIf(item -> now - item.getValue().getLastUpdate() > RECYCLE_PERIOD);                    methodWeightMap.put(key, newMap);                } finally {                    updateLock.set(false);                }            }        }        if (selectedInvoker != null) {            // 將本次選中的invoker, 權(quán)重置為最低, 以便下次不會(huì)再被選中            selectedWRR.sel(totalWeight);            return selectedInvoker;        }        // should not happen here        return invokers.get(0);    }
}

/** * LeastActiveLoadBalance * <p> * Filter the number of invokers with the least number of active calls and count the weights and quantities of these invokers. * If there is only one invoker, use the invoker directly; * if there are multiple invokers and the weights are not the same, then random according to the total weight; * if there are multiple invokers and the same weight, then randomly called. */public class LeastActiveLoadBalance extends AbstractLoadBalance {    // 自身標(biāo)識(shí)    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;

        // Filter out all the least active invokers        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) {            // 如果只有一個(gè)最小則直接返回            // 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.            // 如果權(quán)重不相同且權(quán)重大于0則按總權(quán)重?cái)?shù)隨機(jī)            // 并確定隨機(jī)值落在哪個(gè)片斷上（第一個(gè)相減為負(fù)的值）            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)]);    }}

/** * ConsistentHashLoadBalance */public class ConsistentHashLoadBalance extends AbstractLoadBalance {    public static final String NAME = "consistenthash";
    /**     * Hash nodes name     * 通過(guò) 指定 hash.nodes=0,1,2... 可以自定義參與一致性hash的參數(shù)列表     */    public static final String HASH_NODES = "hash.nodes";
    /**     * Hash arguments name     */    public static final String HASH_ARGUMENTS = "hash.arguments";    // 使用selector 保存某個(gè)固定狀態(tài)時(shí) invoker 的映射關(guān)系    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;        int identityHashCode = System.identityHashCode(invokers);        ConsistentHashSelector<T> selector = (ConsistentHashSelector<T>) selectors.get(key);        // 第一次進(jìn)入或者 identityHashCode 不相等時(shí)（invoker環(huán)境發(fā)生了變化）        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) {            // 存放虛擬節(jié)點(diǎn)            this.virtualInvokers = new TreeMap<Long, Invoker<T>>();            this.identityHashCode = identityHashCode;            URL url = invokers.get(0).getUrl();            // hash.nodes 默認(rèn)是 160            this.replicaNumber = url.getMethodParameter(methodName, HASH_NODES, 160);            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]);            }            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) {            // 取出參與一致性hash計(jì)算的參數(shù)信息            String key = toKey(invocation.getArguments());            byte[] digest = md5(key);            // 根據(jù)hash值選取 invoker            return selectForKey(hash(digest, 0));        }
        private String toKey(Object[] args) {            StringBuilder buf = new StringBuilder();            // 按照指定的參與hash的參數(shù)，調(diào)用 toString() 方法，得到參數(shù)標(biāo)識(shí)信息            for (int i : argumentIndex) {                if (i >= 0 && i < args.length) {                    buf.append(args[i]);                }            }            return buf.toString();        }
        private Invoker<T> selectForKey(long hash) {            // ceilingEntry            Map.Entry<Long, Invoker<T>> entry = virtualInvokers.ceilingEntry(hash);            // 如果沒(méi)有找到，取第一個(gè)值            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();        }
    }
}

demo=my=com.demo.dubbo.DemoLoadBalance