风在路上
zdk
2022-08-01
目录

负载均衡机制

Table of Contents generated with DocToc (opens new window)

# Dubbo自带的负载均衡机制

负载均衡改善了跨多个计算资源（例如计算机、计算机集群、网络连接、中央处理单元或磁盘驱动）的工作负载分布。负载平衡旨在优化资源使用，最大化吞吐量，最小化响应时间，并避免任何单个资源的过载。使用具有负载平衡而不是单个组件的多个组件可以通过冗余提高可靠性和可用性。负载均衡通常涉及专用软件或硬件。

Dubbo提供了四种自带的负载均衡机制

  1. RandomLoadBalance：根据权重随机选择，Dubbo默认采用的策略；
  2. RoundRobinLoadBalance：加权轮询负载均衡；
  3. LeastActiveLoadBalance：最小活跃数负载均衡；
  4. ConsistentHashLoadBalance：一致性Hash负载均衡策略；

# 源码 LoadBalance接口

负载均衡核心，定义了从若干个服务中选择特定的服务进行调用的接口

@SPI(RandomLoadBalance.NAME)
public interface LoadBalance {

    /**
     * select one invoker in list.
     *
     * @param invokers   invokers.
     * @param url        refer url
     * @param invocation invocation.
     * @return selected invoker.
     */
    @Adaptive("loadbalance")
    <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

# AbstractLoadBalance抽象类

负载均衡接口的抽象实现，各种负载均衡策略实现类的父类，封装了公共的逻辑，实现了select方法的算法模板；

public abstract class AbstractLoadBalance implements LoadBalance {
    // 根据正常运行时间和预热时间比例计算新的权重值
    static int calculateWarmupWeight(int uptime, int warmup, int weight) {
		// 计算权重，下面代码逻辑上形似于 (uptime / warmup) * weight。
		// 随着服务运行时间 uptime 增大，权重计算值 ww 会慢慢接近配置值 weight
        int ww = (int) ( uptime / ((float) warmup / weight));
        return ww < 1 ? 1 : (Math.min(ww, weight));
    }

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


    //考虑预热时间的权重
    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 {
			// 从url中获取权重weight配置
            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;
                    }
                    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);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

# RandomLoadBalance

# 简介

根据权重随机选择，是Dubbo默认采用的策略

算法实现

加权随机算法的具体实现，它的算法思想很简单。

假设我们有一组服务器servers = [A, B, C]，他们对应的权重为weights = [5, 3, 2]，权重总和为10。现在把这些权重值平铺在一维坐标值上，[0, 5)区间属于服务器A，[5, 8)区间属于服务器B，[8, 10)区间属于服务器C。接下来通过随机数生成器生成一个范围在[0, 10)之间的随机数，然后计算这个随机数会落到哪个区间上。

比如数字3会落到服务器A对应的区间上，此时返回服务器A即可。权重越大的机器，在坐标轴上对应的区间范围就越大，因此随机数生成器生成的数字就会有更大的概率落到此区间内。只要随机数生成器产生的随机数分布性很好，在经过多次选择后，每个服务器被选中的次数比例接近其权重比例。

比如，经过一万次选择后，服务器A被选中的次数大约为5000次，服务器B被选中的次数约为3000次，服务器C被选中的次数约为2000次。 RandomLoadBalance是一个简单、高效的负载均衡实现，因此Dubbo选择它作为缺省实现。

# 核心代码

public class RandomLoadBalance extends AbstractLoadBalance {

    public static final String NAME = "random";

    @Override
    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        // 所有invoker的数量
        int length = invokers.size();
        // 每个invoker是否有相同的权重
        boolean sameWeight = true;
        // 每个invoker的权重数组
        int[] weights = new int[length];
        // 获取第一个invoker的权重
        int firstWeight = getWeight(invokers.get(0), invocation);
        weights[0] = firstWeight;
        // 统计权重总值
        int totalWeight = firstWeight;
        for (int i = 1; i < length; i++) {
			// 获取每个invoker的权重
            int weight = getWeight(invokers.get(i), invocation);
            weights[i] = weight;
            // 统计权重总值
            totalWeight += weight;
            if (sameWeight && weight != firstWeight) {
				// 存在权重不一致的invoker
                sameWeight = false;
            }
        }
		// 存在权重值，且所有的invoker的权重值有差异
        if (totalWeight > 0 && !sameWeight) {
			
			// 基于总权重值产生随机值
            int offset = ThreadLocalRandom.current().nextInt(totalWeight);
            // 挑选出随机值落在对应权重范围的invoker
            for (int i = 0; i < length; i++) {
                offset -= weights[i];
                if (offset < 0) {
                    return invokers.get(i);
                }
            }
        }
        // 这里是如果每个invoker权重相同，那么无需考虑权重，直接随机挑选
        return invokers.get(ThreadLocalRandom.current().nextInt(length));
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

# RoundRobinLoadBalance

# 简介

RoundRobinLoadBalance是加权轮询负载均衡的实现

何为加权轮询？

所谓轮询是指将请求轮流分配给每台服务器。

举个例子，我们有三台服务器A、B、C。我们将第一个请求分配给服务器A，第二个请求分配给服务器B，第三个请求分配给服务器C，第四个请求再次分配给服务器A；这个过程就叫做轮询。

轮询是一种无状态负载均衡算法，实现简单，适用于每台服务器性能相近的场景下。但现实情况下，我们并不能保证每台服务器性能均相近。如果我们将等量的请求分配给性能较差的服务器，这显然是不合理的。

因此，这个时候我们需要对轮询过程进行加权，以调控每台服务器的负载。经过加权后，每台服务器能够得到的请求数比例，接近或等于他们的权重比。

比如服务器 A、B、C 权重比为：5:2:1。那么在8次请求中，服务器A将收到其中的5次请求，服务器B会收到其中的2次请求，服务器C则收到其中的1次请求。

# 核心代码

public class RoundRobinLoadBalance extends AbstractLoadBalance {
    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();
    
    
    @Override
    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
		// 从url获取到WeightedRoundRobin的映射map，如果为null，就put一个新的
		// computeIfAbsent方法详情见ConcurrentHashMap中
        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;
		// 下面这个循环主要做了这样几件事情：
		/**
		 1. 遍历 Invoker 列表，检测当前 Invoker 是否有对应的 WeightedRoundRobin，没有则创建
		 2. 检测 Invoker 权重是否发生了变化，若变化了，则更新 WeightedRoundRobin 的 weight 字段
		 3. 让 current 字段加上自身权重，等价于 current += weight
		 4. 设置 lastUpdate 字段，即 lastUpdate = now
		 5. 寻找具有最大 current 的 Invoker，以及 Invoker 对应的WeightedRoundRobin，暂存起来，留作后用
		 6. 计算权重总和
		*/
        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加上自身的权重
			// 因为上面给weightedRoundRobin的weight设置了新值
			// 这里就是current+=weight 只不过加操作是原子的而已
            long cur = weightedRoundRobin.increaseCurrent();
			// 更新权重上次更新时间为 现在时间
            weightedRoundRobin.setLastUpdate(now);
			// 找出最大的current 
            if (cur > maxCurrent) {
                maxCurrent = cur;
				// 将具有最大权重的invoker赋值给筛选结果
                selectedInvoker = invoker;
				// 将此invoker对应的WeightRoundRobin赋值给筛选结果的WRR 留作后用
                selectedWRR = weightedRoundRobin;
            }
			//计算权重总和
            totalWeight += weight;
        }
		/**
		对methodWeightMap进行检查，过滤掉长时间未被更新的节点
		因为该节点可能挂了，invokers中不包含该节点，所以节点的lastUpdate长时间无法被更新
		未更新时长超过设定的阈值后，节点就会被移除，默认阈值为60秒
		*/
        if (!updateLock.get() && invokers.size() != map.size()) {
			// 这里保证更新节点的线程安全 使用原子操作作为锁
            if (updateLock.compareAndSet(false, true)) {
                try {
                    // copy -> modify -> update reference
                    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) {
			// 让 current 减去权重总和，等价于 current -= totalWeight
            selectedWRR.sel(totalWeight);
            return selectedInvoker;
        }
        // should not happen here
        return invokers.get(0);
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115

# 算法原理及示例

算法实现参考自Nginx的平滑加权轮询负载均衡，每个服务器对应两个权重，分别为weight和currentWeight，其中weight是固定的，currentWeight会动态调

整，初始值为0。当有新的请求进来时，遍历服务器列表，让它的currentWeight加上自身权重；遍历完成后，找到最大的currentWeight，并将其减去权重总和，

然后返回相应的服务器即可。

假设我们有一组服务器servers = [A, B, C]，他们对应的权重为weights = [5, 1, 1]，权重总和为7，现在有7个请求依次进入负载均衡逻辑，选择过程如下：

请求编号 currentWeight数组 选择结果 减去权重总和后的currentWeight数组
1 [5, 1, 1] A [-2, 1, 1]
2 [3, 2, 2] A [-4, 2, 2]
3 [1, 3, 3] B [ 1, -4, 3]
4 [6,-3, 4] A [-1, -3, 4]
5 [4,-2, 5] C [ 4, -2, -2]
6 [9,-1,-1] A [ 2, -1, -1]
7 [7, 0, 0] A [ 0, 0, 0]

# LeastActiveLoadBalance

# 简介

简介

最小活跃数负载均衡。活跃调用数越小，表明该提供者效率越高，单位时间内可处理更多的请求，此时应先将请求分配给该服务提供者。

在具体实现中，每个服务提供者对应一个活跃数active。初始情况下，所有服务提供者活跃数均为0。每收到一个请求，活跃数+1，完成请求后则将活跃数-1。在服务运行一段时间后，性能好的服务提供者处理请求的速度更快，因此活跃数下降的也越快，此时这样的服务提供者能够优先获取到新的服务请求，这就是最小活跃数负载均衡算法的基本思想。

除了最小活跃数，LeastActiveLoadBalance在实现上还引入了权重值，所以准确的来说LeastActiveLoadBalance是基于加权最小活跃数算法实现的。

举个例子说明一下：在一个服务提供者集群中，有两个性能优异的服务提供者。某一时刻它们的活跃数相同，此时Dubbo会根据它们的权重去分配请求，权重越大，获取到新请求的概率就越大；如果两个服务提供者权重相同，此时随机选择一个即可。

# 核心代码

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) {
        // invoker的总数
        int length = invokers.size();
        // 所有invoker的最小活跃数的默认值 -1
        int leastActive = -1;
        // 具有相同“最小活跃数”的服务者提供者（以下用 invoker 代称）数量
        int leastCount = 0;
        // leastIndexs 用于记录具有相同“最小活跃数”的 invoker 在 invokers 列表中的下标信息
        int[] leastIndexes = new int[length];
        // 所有invoker的权重数组
        int[] weights = new int[length];
        // The sum of the warmup weights of all the least active invokers
        int totalWeight = 0;
        // 第一个最小活跃数的 invoker 权重值，用于与其他具有相同最小活跃数的 invoker 的权重进行对比，
        int firstWeight = 0;
        // 检测是否"所有具有相同最小活跃数的 invoker 的权重" 均相等
        boolean sameWeight = true;


		// 过滤出所有最小活跃数的invoker
        for (int i = 0; i < length; i++) {
            Invoker<T> invoker = invokers.get(i);
            // 获取这个invoker的活跃数
            int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive();
            // 获取这个invoker配置的权重 默认为100.
            int afterWarmup = getWeight(invoker, invocation);
            // 缓存权重值以便后续使用
            weights[i] = afterWarmup;
			// 发现更小的活跃数，重新开始
            if (leastActive == -1 || active < leastActive) {
				// 更新最小活跃数为当前invoker的活跃数
                leastActive = active;
                // 重设具有最小活跃数的invoker的数量为1
                leastCount = 1;
                // 把这个具有最小活跃数的invoker的下标放入leastIndexes的第一个
                leastIndexes[0] = i;
                // 重设总权重值
                totalWeight = afterWarmup;
                // Record the weight the first least active invoker
                firstWeight = afterWarmup;
                // 设置每个invoker都有相同的权重(因为此时具备最小活跃数的invoker只有一个，所以是true)
                sameWeight = true;
                // 如果当前 Invoker 的活跃数 active 与最小活跃数 leastActive 相同 
            } else if (active == leastActive) {
                // 在leastIndexes中记录下相等最小活跃数的这个invoker的下标
				// 且对应的拥有最小活跃数的invoker的数量++
                leastIndexes[leastCount++] = i;
                // 累加权重
                totalWeight += afterWarmup;
                // 检测当前 invoker 的权重与 firstWeight 是否相等
                if (sameWeight && i > 0
                        && afterWarmup != firstWeight) {
					// 不相等则将 sameWeight 置为 false
                    sameWeight = false;
                }
            }
			// 当前invoker权重值大于最小权重，则无需处理
        }
        // 当只有一个 invoker 具有最小活跃数，此时直接返回该 invoker 即可
        if (leastCount == 1) {
            return invokers.get(leastIndexes[0]);
        }
		// 如果有多个invoker的具有相同的最小活跃数，但它们的权重不相同
        if (!sameWeight && totalWeight > 0) {
			// 随机生成一个 [0, totalWeight) 之间的数字
            int offsetWeight = ThreadLocalRandom.current().nextInt(totalWeight);
            // 循环让随机数减去具有最小活跃数的 Invoker 的权重值，
			// 当 offset 小于等于0时，返回相应的 Invoker
            for (int i = 0; i < leastCount; i++) {
                int leastIndex = leastIndexes[i];
                offsetWeight -= weights[leastIndex];
                if (offsetWeight < 0) {
                    return invokers.get(leastIndex);
                }
            }
        }
        // 如果权重相同或总权重为0时，随机返回一个Invoker即可
        return invokers.get(leastIndexes[ThreadLocalRandom.current().nextInt(leastCount)]);
    }
}

// RpcStatus.java
private static final ConcurrentMap<String, ConcurrentMap<String, RpcStatus>> METHOD_STATISTICS = new ConcurrentHashMap<String, ConcurrentMap<String, RpcStatus>>();

public static RpcStatus getStatus(URL url, String methodName) {
    String uri = url.toIdentityString();
    ConcurrentMap<String, RpcStatus> map = METHOD_STATISTICS.computeIfAbsent(uri, k -> new ConcurrentHashMap<>());
    return map.computeIfAbsent(methodName, k -> new RpcStatus());
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

# 算法主要逻辑

算法实现

  1. 遍历invokers列表，寻找活跃数最小的invoker；
  2. 如果有多个invoker具有相同的最小活跃数，此时记录下这些invoker在invokers集合中的下标，并累加它们的权重，比较它们的权重值是否相等；
  3. 如果只有一个invoker具有最小的活跃数，此时直接返回该invoker即可；
  4. 如果有多个invoker具有最小活跃数，且它们的权重不相等，此时处理方式和RandomLoadBalance一致；
  5. 如果有多个invoker具有最小活跃数，但它们的权重相等，此时随机返回一个即可；

# ConsistentHashLoadBalance

# 简介

概述

一致性hash算法提出之初是用于大规模缓存系统的负载均衡。

它的工作过程是这样的，首先根据ip或者其他的信息为缓存节点生成一个hash，并将这个hash投射到[0, 2^32 - 1]的圆环上；当有查询或写入请求时，则为缓存项的key生成一个hash值。然后查找第一个大于或等于该hash值的缓存节点，并到这个节点中查询或写入缓存项。如果当前节点挂了，则在下一次查询或写入缓存时，为缓存项查找另一个大于其hash值的缓存节点即可。

大致效果如下图所示，每个缓存节点在圆环上占据一个位置。

如果缓存项的key的hash值小于缓存节点hash值，则到该缓存节点中存储或读取缓存项。比如下面绿色点对应的缓存项将会被存储到 cache-2节点中。由于cache-3挂了，原本应该存到该节点中的缓存项最终会存储到cache-4节点中。

image.png

在Dubbo中，把上图的缓存节点替换成Dubbo的服务提供者，于是得到了下图：

image.png

这里相同颜色的节点均属于同一个服务提供者，比如Invoker1-1、Invoker1-2、……, Invoker1-160，这样做的目的是通过引入虚拟节点，让Invoker在圆环上分散开来，避免数据倾斜问题。所谓数据倾斜是指，由于节点不够分散，导致大量请求落到了同一个节点上，而其他节点只会接收到了少量请求的情况。

# 核心代码

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;
    // 获取 invokers 原始的 hashcode
    int invokersHashCode = getCorrespondingHashCode(invokers);
    ConsistentHashSelector<T> selector = (ConsistentHashSelector<T>) selectors.get(key);
    // 如果 invokers 是一个新的 List 对象，意味着服务提供者数量发生了变化，可能新增也可能减少了。
    // 此时 selector.identityHashCode != identityHashCode 条件成立
    if (selector == null || selector.identityHashCode != invokersHashCode) {
        // 创建新的 ConsistentHashSelector
        selectors.put(key, new ConsistentHashSelector<T>(invokers, methodName, invokersHashCode));
        selector = (ConsistentHashSelector<T>) selectors.get(key);
    }
    // 调用 ConsistentHashSelector 的 select 方法选择 Invoker
    return selector.select(invocation);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

ConsistentHashLoadBalance使用了内部类ConsistentHashSelector，先分析其初始化过程，代码如下：

private static final class ConsistentHashSelector<T> {

    // 使用 TreeMap 存储 Invoker 虚拟节点
    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
        this.replicaNumber = url.getMethodParameter(methodName, HASH_NODES, 160);
        // 获取参与 hash 计算的参数下标值，默认对第一个参数进行 hash 运算
        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++) {
                // 对 address + i 进行 md5 运算，得到一个长度为16的字节数组
                byte[] digest = Bytes.getMD5(address + i);
                // 对 digest 部分字节进行4次 hash 运算，得到四个不同的 long 型正整数
                for (int h = 0; h < 4; h++) {
                    // h = 0 时，取 digest 中下标为 0 ~ 3 的4个字节进行位运算
                    // h = 1 时，取 digest 中下标为 4 ~ 7 的4个字节进行位运算
                    // h = 2, h = 3 时过程同上
                    long m = hash(digest, h);
                    // 将 hash 到 invoker 的映射关系存储到 virtualInvokers 中，
                    // virtualInvokers 需要提供高效的查询操作，因此选用 TreeMap 作为存储结构
                    virtualInvokers.put(m, invoker);
                }
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

提示

ConsistentHashSelector的构造方法执行了一系列的初始化逻辑，比如从配置中获取虚拟节点数以及参与hash计算的参数下标，默认情况下只使用第一个参数进行hash。

需要特别说明的是，ConsistentHashLoadBalance的负载均衡逻辑只受参数值影响，具有相同参数值的请求将会被分配给同一个服务提供者。

在获取虚拟节点数和参数下标配置后，接下来要做的事情是计算虚拟节点hash值，并将虚拟节点存储到TreeMap中。

到此，ConsistentHashSelector初始化工作就完成了，接下来，分析select方法，代码如下：

		public Invoker<T> select(Invocation invocation) {
			// 将参数转为 key
            String key = toKey(invocation.getArguments());
			// 对key进行md5计算
            byte[] digest = md5(key);
			// 取 digest 数组的前四个字节进行 hash 运算，再将 hash 值传给 selectForKey 方法，
			// 寻找合适的 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) {
			// 到 TreeMap 中查找第一个节点值大于或等于当前 hash 的 Invoker
            Map.Entry<Long, Invoker<T>> entry = virtualInvokers.ceilingEntry(hash);
			// 如果 hash 大于 Invoker 在圆环上最大的位置，此时 entry = null，
			// 需要将 TreeMap 的头节点赋值给 entry
            if (entry == null) {
                entry = virtualInvokers.firstEntry();
            }
			//最后返回invoker
            return entry.getValue();
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

# ShortestResponseLoadBalance

# 简介

ShortestResponseLoadBalance（最短响应时间算法），监控服务的响应时间，并根据响应时间排序，选择响应时间最短的服务器。(2.7.7 release 版本新加的一种负载均衡)

# 核心代码

//org.apache.dubbo.rpc.cluster.loadbalance.ShortestResponseLoadBalance#doSelect
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
    //...... 删除其他代码 基本同最小活跃数算法
    for (int i = 0; i < length; i++) {
        Invoker<T> invoker = invokers.get(i);
        RpcStatus rpcStatus = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName());
        //每个请求成功时响应的平均时间 [每个请求开始有个时间戳 
		//结束时的时间减去开始时间加入rpcStatus]
        long succeededAverageElapsed = rpcStatus.getSucceededAverageElapsed();
        //获取当前在活跃调用的请求数量
        int active = rpcStatus.getActive();
        //平均请求时间*活跃数表示最小响应时间
        long estimateResponse = succeededAverageElapsed * active;
        int afterWarmup = getWeight(invoker, invocation);
        weights[i] = afterWarmup;
        //计算处理
        if (estimateResponse < shortestResponse) {
            shortestResponse = estimateResponse;
            shortestCount = 1;
            shortestIndexes[0] = i;
            totalWeight = afterWarmup;
            firstWeight = afterWarmup;
            sameWeight = true;
        } else if (estimateResponse == shortestResponse) {
            shortestIndexes[shortestCount++] = i;
            totalWeight += afterWarmup;
            if (sameWeight && i > 0
                    && afterWarmup != firstWeight) {
                sameWeight = false;
            }
        }
    }
    //最短响应时间相同的Invoker只有一个
    if (shortestCount == 1) {
        return invokers.get(shortestIndexes[0]);
    }
    //最短响应时间相同的Invoker有多个,且存在权重不同 加权获取
    if (!sameWeight && totalWeight > 0) {
        int offsetWeight = ThreadLocalRandom.current().nextInt(totalWeight);
        for (int i = 0; i < shortestCount; i++) {
            int shortestIndex = shortestIndexes[i];
            offsetWeight -= weights[shortestIndex];
            if (offsetWeight < 0) {
                return invokers.get(shortestIndex);
            }
        }
    }
    //最短响应时间相同的Invoker有多个,且权重相同 随机获取
    return invokers.get(shortestIndexes[ThreadLocalRandom.current().nextInt(shortestCount)]);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

# 算法主要逻辑

基本和最小活跃数算法是一样的

算法实现

  1. 遍历invokers列表，寻找最小响应时间最小的invoker；
  2. 如果有多个invoker具有相同的最小响应时间，此时记录下这些invoker在invokers集合中的下标，并累加它们的权重，比较它们的权重值是否相等；
  3. 如果只有一个invoker具有最小响应时间，此时直接返回该invoker即可；
  4. 如果有多个invoker具有最小响应时间，且它们的权重不相等，此时处理方式和RandomLoadBalance一致；
  5. 如果有多个invoker具有最小响应时间，但它们的权重相等，此时随机返回一个即可；
在 GitHub 上编辑此页 (opens new window)
#Dubbo#负载均衡
最后更新: 2022/10/04, 16:10:00
SPI机制
服务容错、降级

← 服务容错、降级→

Theme by Vdoing | Copyright © 2022-2024 zdk | notes
湘ICP备2022001117号-1
川公网安备 51142102511562号
提供加速服务