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

protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
	
	// 获得所有invoker的数量
	int length = invokers.size();

	if (!needWeightLoadBalance(invokers,invocation)){
		return invokers.get(ThreadLocalRandom.current().nextInt(length));
	}

	// 用于判断是否所有的invoker都有相同的权重
	boolean sameWeight = true;
	// 存储每个invoker的最大权重，也即累积和
	int[] weights = new int[length];
	// 计算权重和
	int totalWeight = 0;
	for (int i = 0; i < length; i++) {
		// 计算每个invoker的权重
		int weight = getWeight(invokers.get(i), invocation);
		// 求和
		totalWeight += weight;
		// 存储每个invoker当前的累积和
		weights[i] = totalWeight;
		// 判断权重是否不同
		if (sameWeight && totalWeight != weight * (i + 1)) {
			sameWeight = false;
		}
	}
	
	if (totalWeight > 0 && !sameWeight) {
		// 如果各个invoker的权重不相同，则根据我们上一节介绍的算法随机选出一个invoker
		int offset = ThreadLocalRandom.current().nextInt(totalWeight);
		for (int i = 0; i < length; i++) {
			if (offset < weights[i]) {
				return invokers.get(i);
			}
		}
	}
	// 如果权重和为0或权重均相同，则随机返回一个
	return invokers.get(ThreadLocalRandom.current().nextInt(length));
}

1
2
3
4
5
6

servers = ['A', 'B', 'C']
index = 0
def select():
	global index
	index = (index+1) % len(servers)
	return servers[index]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

servers = [(2, 'A'), (8, 'B'), (1, 'C')]
index = 0

def select():
	global index
	total_weight = 0
	weights = [0] * len(servers)
	target_server = servers[0]
	for i in range(len(servers)):
		total_weight += servers[i][0]
		weights[i] = total_weight
	for i in range(len(servers)):
		if index < weights[i]:
			target_server = servers[i]
			break
	index = (index + 1) % total_weight
	return target_server

1
2
3
4
5
6
7
8

protected static class WeightedRoundRobin {
	// 节点的权重
	private int weight;
	// 节点的当前权重
	private AtomicLong current = new AtomicLong(0);
	// 节点权重的上次更新时间
	private long lastUpdate;
}

 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

@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
    // 负载均衡实现的粒度是方法
    String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
    // 根据方法key获取invoker的负载均衡map
    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;
	for (Invoker<T> invoker : invokers) {
		// 获取每个invoker的唯一标识
		String identifyString = invoker.getUrl().toIdentityString();
		int weight = getWeight(invoker, invocation);
		// 根据唯一标志获取该inoker的当前权重，如果没有就新建
		WeightedRoundRobin weightedRoundRobin = map.computeIfAbsent(identifyString, k -> {
			WeightedRoundRobin wrr = new WeightedRoundRobin();
			wrr.setWeight(weight);
			return wrr;
		});
		// 如果权重发生变化，那么就重新设置权重。这里会将当前权重置0，同时设置节点权重
		if (weight != weightedRoundRobin.getWeight()) {
			//weight changed
			weightedRoundRobin.setWeight(weight);
		}
		// 更新节点的当前权重，current = current + weight
		long cur = weightedRoundRobin.increaseCurrent();
		weightedRoundRobin.setLastUpdate(now);
		if (cur > maxCurrent) {
			maxCurrent = cur;
			selectedInvoker = invoker;
			selectedWRR = weightedRoundRobin;
		}
		totalWeight += weight;
	}
	// 移除已经不活跃的invker的权重，将在外下一次负载均衡生效
	if (invokers.size() != map.size()) {
		map.entrySet().removeIf(item -> now - item.getValue().getLastUpdate() > RECYCLE_PERIOD);
	}
	// 选出invoker后，将该invoker的当前权重减去权重和
	if (selectedInvoker != null) {
		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

int getWeight(Invoker<?> invoker, Invocation invocation) {
	int weight;
	URL url = invoker.getUrl();
	// Multiple registry scenario, load balance among multiple registries.
	if (REGISTRY_SERVICE_REFERENCE_PATH.equals(url.getServiceInterface())) {
		weight = url.getParameter(REGISTRY_KEY + "." + WEIGHT_KEY, DEFAULT_WEIGHT);
	} else {
		// 从Method中获取权重参数
		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;
				}
				// 获取warmup时间
				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);
}

static int calculateWarmupWeight(int uptime, int warmup, int weight) {
	// 随着uptime增大，uptime/warmup逐渐变大
	int ww = (int) ( uptime / ((float) warmup / weight));
	// 当uptime/warmup > 1时为weight
	return ww < 1 ? 1 : (Math.min(ww, weight));
}