搞定 Dubbo 系列（三）：负载均衡策略和源码解析

Dear 丶 2023-06-26 05:37 14阅读 0赞

loadBalance是现在服务端框架的标配，dubbo作为在互联网端流行这么多年的rpc框架，自然也少不了。现在的服务端一般都是集群部署，dubbo 需要对服务消费者的调用请求进行分配，避免少数服务提供者负载过大。

## 一、四种负载均衡策略 ##

dubbo 提供了四种负载均衡策略，分别是，

1.  Random LoadBalance 随机（**默认负载均衡策略**），可以完全随机，也可以按权重设置随机概率。  
    在一个截面上碰撞的概率高，但调用量越大分布越均匀，而且按概率使用权重后也比较均匀，有利于动态调整提供者权重。
2.  RoundRobin LoadBalance 轮循，按公约后的权重设置轮循比率。  
    存在响应慢的提供者会累积请求的问题，比如：第二台机器很慢，但没挂，当请求调到第二台时就卡在那，久而久之，所有请求都卡在调到第二台上。
3.  LeastActive LoadBalance 最少活跃调用数，相同活跃数的随机，活跃数指调用前后计数差。  
    使慢的提供者收到更少请求，因为越慢的提供者的调用前后计数差会越大。
4.  ConsistentHash LoadBalance 一致性Hash，相同参数的请求总是发到同一提供者。  
    当某一台提供者挂时，原本发往该提供者的请求，基于虚拟节点，平摊到其它提供者，不会引起剧烈变动。

## 二、负载均衡配置 ##

**1、服务端（提供者）**

@Service(interfaceClass = IBookService.class, loadbalance = "roundrobin", retries = 3, version = "1.0")
    public class BookServiceImpl implements IBookService {
        // 省略实现
    }

<dubbo:service interface="..." loadbalance="roundrobin" />

**2、调用端（消费者）**

@Service
    public class DubboService {
    
        @Reference(interfaceClass = IBookService.class, loadbalance = "random", retries = 3, version = "1.0")
        IBookService bookService;
    
        // 省略逻辑实现
    }

<dubbo:reference interface="..." loadbalance="roundrobin" />

## 三、源码解析 ##

### 负载均衡类的依赖 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NoaXBmZWlfY3Nkbg_size_16_color_FFFFFF_t_70][]

AbstractLoadBalance 抽象类是所有负载均衡策略实现类的父类，实现了LoadBalance接口 的方法，同时提供抽象方法交由子类实现。

@Override
        public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) {
            if (CollectionUtils.isEmpty(invokers)) {
                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);

### 1、RandomLoadBalance，随机 ###

RandomLoadBalance 的实现逻辑分为权重相同和权重不同的情况。如果每个提供者的权重都相同，那么根据列表长度直接随机选取一个，如果权重不同，累加权重值。根据0~累加的权重值 选取一个随机数，然后判断该随机数落在那个提供者上。这种策略大概率能够让权重大的provider获取更多机会同时兼顾了权重小的接口。

@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;
                // Sum
                totalWeight += weight;
                if (sameWeight && weight != firstWeight) {
                    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 = 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.
            return invokers.get(ThreadLocalRandom.current().nextInt(length));
        }

### 2、RoundRobinLoadBalance，轮询 ###

轮询的实现逻辑分为权重相同和权重不同两种情况。先判断权重是否一致，如果一致，通过维护一个 AtomicLong 的增长 进行取模来轮询。如果权重不一致，通过维护一个 AtomicLong 的增长与最大权重取模作为当前权重，然后获取大于当前权重的列表作为调用者列表，然后进行取模轮询。

@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);
                }
                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 {
                        // copy -> modify -> update reference
                        ConcurrentMap<String, WeightedRoundRobin> newMap = new ConcurrentHashMap<String, WeightedRoundRobin>();
                        newMap.putAll(map);
                        Iterator<Entry<String, WeightedRoundRobin>> it = newMap.entrySet().iterator();
                        while (it.hasNext()) {
                            Entry<String, WeightedRoundRobin> item = it.next();
                            if (now - item.getValue().getLastUpdate() > RECYCLE_PERIOD) {
                                it.remove();
                            }
                        }
                        methodWeightMap.put(key, newMap);
                    } finally {
                        updateLock.set(false);
                    }
                }
            }
            if (selectedInvoker != null) {
                selectedWRR.sel(totalWeight);
                return selectedInvoker;
            }
            // should not happen here
            return invokers.get(0);
        }

### 3、LeastActiveLoadBalance，最少活跃调用数 ###

最少活跃调用数的思想就是：获取最小的活跃数，把活跃数等于最小活跃数的调用者维护成一个数组。如果权重一致随机取出，如果不同则跟 RandomLoadBalance 一致，累加权重，然后随机取出。

@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 invokes
            int totalWeight = 0;
            // The weight of the first least active invoke
            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 invoke
                int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive();
                // Get the weight of the invoke 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)]);
        }

### 4、ConsistentHashLoadBalance，一致性hash ###

一致性hash，主要是通过内部类 ConsistentHashSelector 来实现（ConsistentHashSelector 的源码在这里就不列出了）：

*  首先根据invokers的url获取分片个数，创建相同大小的虚拟节点。
 *  然后根据参数的MD5值 获取对应的提供者。

@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);
            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);
        }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NoaXBmZWlfY3Nkbg_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191230135643322.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NoaXBmZWlfY3Nkbg==,size_16,color_FFFFFF,t_70