dubbo学习三-负载均衡策略_loadbalance dubboreference

1 负载均衡策略介绍

dubbo负载均衡包括4种，随机、轮询、最小活跃调用数、IP哈希
Dubbo负载均衡类图如下

Dubbo使用负载均衡方式很简单，只需要在服务调用者标签上配置loadbalance属性就行

1.1 随机(默认)

@Reference(loadbalance = "random")

如果配置了权重，则invoker权重越大，执行的概率越大；如果没有配置权重，则随机选择invoker执行

1.2 轮询

@Reference(loadbalance = "roundrobin")

如果配置了权重，则根据权重轮询；否则直接通过方法调用次数取模

1.3 最小活跃数

@Reference(loadbalance = "leastactive")

指的是当请求调用来临，有多个实例提供服务的时候，选择其中被调用活跃次数最少的实例来提供服务。通俗一点讲就是，当前有 3 个实例在提供服务，A 当前被 2 个服务调用，B 当前被 3 个服务调用，C 当前被 1 个服务调用，一个新的调用请求过来，会选择调用到 C 实例。
能动态的根据当前服务的调用情况，选择最小被调用的实例，调用越慢的机器，会接收到更少的请求。因为调用越慢的机器调用间隔越大，累积的请求越多

1.4 一致性哈希

@Reference(loadbalance = "consistenthash")

相同参数的请求总是发到同一提供者。当某一台提供者挂时，原本发往该提供者的请求，基于虚拟节点，平摊到其它提供者，不会引起剧烈变动。

2 实现原理

2.1 随机算法RandomLoadBalance

protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
    // 获取所有的服务提供者invoker数量
    int length = invokers.size();
    // 初始化总权重为0
    int totalWeight = 0;
    // 默认权重都相等
    boolean sameWeight = true;
 
    int offset;
    int i;
    for(offset = 0; offset < length; ++offset) {
        // 获取某个invoker的权重(可以在@Server注解上配置)
        i = this.getWeight((Invoker)invokers.get(offset), invocation);
        totalWeight += i;
        if (sameWeight && offset > 0 && i != this.getWeight((Invoker)invokers.get(offset - 1), invocation)) {
            // 如果任意相邻两个invoker权重不相等则把sameWeight设置为false
            sameWeight = false;
        }
    }
 
    // 下面是轮询权重算法
    if (totalWeight > 0 && !sameWeight) {
        // 从0-totalWeight中随机选取一个数
        offset = this.random.nextInt(totalWeight);
        for(i = 0; i < length; ++i) {
            // 这个数减去该invoker的offset
            offset -= this.getWeight((Invoker)invokers.get(i), invocation);
            if (offset < 0) {
                // 如果相减后的值<0则用该invoker，权重越大，相减后<0的概率越大
                return (Invoker)invokers.get(i);
            }
        }
    }
    // 如果权重都相同或者权重算法相减后都>0则随机选一个
    return (Invoker)invokers.get(this.random.nextInt(length));
}

2.2 轮询RoundRobinLoadBalance

protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
    // 用全类名和方法名做key
    String key = ((Invoker)invokers.get(0)).getUrl().getServiceKey() + "." + invocation.getMethodName();
    // 获取invoker的数量
    int length = invokers.size();
    int maxWeight = 0;
    int minWeight = 2147483647;
    // 做一个invoker-weight map
    LinkedHashMap<Invoker<T>, RoundRobinLoadBalance.IntegerWrapper> invokerToWeightMap = new LinkedHashMap();
    int weightSum = 0;
 
    int currentSequence;
    for(int i = 0; i < length; ++i) {
        // 当前invoker权重
        currentSequence = this.getWeight((Invoker)invokers.get(i), invocation);
        // 设置最大权重和最小权重
        maxWeight = Math.max(maxWeight, currentSequence);
        minWeight = Math.min(minWeight, currentSequence);
        if (currentSequence > 0) {
            // 如果权重>0则把invoker-weight放到map中
            invokerToWeightMap.put(invokers.get(i), new RoundRobinLoadBalance.IntegerWrapper(currentSequence));
            weightSum += currentSequence;
        }
    }
 
    // 获取这个方法的执行次数，如果为空则初始化
    AtomicPositiveInteger sequence = (AtomicPositiveInteger)this.sequences.get(key);
    if (sequence == null) {
        this.sequences.putIfAbsent(key, new AtomicPositiveInteger());
        sequence = (AtomicPositiveInteger)this.sequences.get(key);
    }
    // 调用次数自增1
    currentSequence = sequence.getAndIncrement();
    // 如果有权重，则考虑权重因素进行选取
    if (maxWeight > 0 && minWeight < maxWeight) {
        int mod = currentSequence % weightSum;
 
        for(int i = 0; i < maxWeight; ++i) {
            Iterator i$ = invokerToWeightMap.entrySet().iterator();
            while(i$.hasNext()) {
                Entry<Invoker<T>, RoundRobinLoadBalance.IntegerWrapper> each = (Entry)i$.next();
                Invoker<T> k = (Invoker)each.getKey();
                RoundRobinLoadBalance.IntegerWrapper v = (RoundRobinLoadBalance.IntegerWrapper)each.getValue();
                if (mod == 0 && v.getValue() > 0) {
                    return k;
                }
 
                if (v.getValue() > 0) {
                    v.decrement();
                    --mod;
                }
            }
        }
    }
 
    // 根据方法调用次数取模，一次次来
    return (Invoker)invokers.get(currentSequence % length);
}

2.3 最小活跃数LeastActiveLoadBalance

protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
    // 获取invoker个数
    int length = invokers.size();
    // 最小活跃值
    int leastActive = -1;
    // 最小活跃数量
    int leastCount = 0;
    // 做一个数组，记录最小活跃invoker的下标
    int[] leastIndexs = new int[length];
    int totalWeight = 0;
    int firstWeight = 0;
    boolean sameWeight = true;
 
    int offsetWeight;
    int leastIndex;
    for(offsetWeight = 0; offsetWeight < length; ++offsetWeight) {
        Invoker<T> invoker = (Invoker)invokers.get(offsetWeight);
        // 获取该实例的活跃值
        leastIndex = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive();
        int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), "weight", 100);
        if (leastActive != -1 && leastIndex >= leastActive) {
            if (leastIndex == leastActive) {
                // 如果该invoker == 最小活跃度则把该invoker下标记录到leastIndexs中
                leastIndexs[leastCount++] = offsetWeight;
                totalWeight += weight;
                if (sameWeight && offsetWeight > 0 && weight != firstWeight) {
                    sameWeight = false;
                }
            }
        } else {
            // 如果出现更小活跃度的invoker则重新设置最小活跃度的值
            leastActive = leastIndex;
            // 更新最小活跃度个数为1
            leastCount = 1;
            // 把该invoker下标记录到leastIndexs中
            leastIndexs[0] = offsetWeight;
            totalWeight = weight;
            firstWeight = weight;
            sameWeight = true;
        }
    }
 
    if (leastCount == 1) {
        // 如果最小活跃数量invoker个数为1则直接返回
        return (Invoker)invokers.get(leastIndexs[0]);
    } else {
        // 如果有多个最小活跃数量的invoker
        if (!sameWeight && totalWeight > 0) {
            // 如果考虑权重则根据权重做随机
            offsetWeight = this.random.nextInt(totalWeight);
 
            for(int i = 0; i < leastCount; ++i) {
                leastIndex = leastIndexs[i];
                offsetWeight -= this.getWeight((Invoker)invokers.get(leastIndex), invocation);
                if (offsetWeight <= 0) {
                    return (Invoker)invokers.get(leastIndex);
                }
            }
        }
        // 如果没有权重则从最小活跃度invoker中随机选择一个
        return (Invoker)invokers.get(leastIndexs[this.random.nextInt(leastCount)]);
    }
}

2.4 一致性哈希ConsistentHashLoadBalance

protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
    // 用全类名和方法名做key
    String key = ((Invoker)invokers.get(0)).getUrl().getServiceKey() + "." + invocation.getMethodName();
    // 基于 invokers 集合，根据对象内存地址来计算定义哈希值
    int identityHashCode = System.identityHashCode(invokers);
    // 获取根据key获取本次执行的选择器
    ConsistentHashLoadBalance.ConsistentHashSelector<T> selector = (ConsistentHashLoadBalance.ConsistentHashSelector)this.selectors.get(key);
    if (selector == null || selector.identityHashCode != identityHashCode) {
        this.selectors.put(key, new ConsistentHashLoadBalance.ConsistentHashSelector(invokers, invocation.getMethodName(), identityHashCode));
        selector = (ConsistentHashLoadBalance.ConsistentHashSelector)this.selectors.get(key);
    }
 
    return selector.select(invocation);
}
 
public Invoker<T> select(Invocation invocation) {
    // 根据方法参数做key
    String key = this.toKey(invocation.getArguments());
    // 加密
    byte[] digest = this.md5(key);
    // 选取执行的invoker
    return this.selectForKey(this.hash(digest, 0));
}