在集群負載均衡時,Dubbo 提供了多種均衡策略,缺省為 random 隨機調用。
我們還可以擴展自己的負責均衡策略,前提是你已經從一個小白變成了大牛,嘻嘻
1、Random LoadBalance
1.1 隨機,按權重設置隨機概率。
1.2 在一個截面上碰撞的概率高,但調用量越大分佈越均勻,而且按概率使用權重後也比較均勻,有利於動態調整提供者權重。
1.3 源碼分析
<code> package com.alibaba.dubbo.rpc.cluster.loadbalance;
import java.util.List;
import java.util.Random;
import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.rpc.Invocation;
import com.alibaba.dubbo.rpc.Invoker;
/**
* random load balance.
*
* @author qianlei
* @author william.liangf
*/
public class RandomLoadBalance extends AbstractLoadBalance {
public static final String NAME = "random";
private final Random random = new Random();
protectedInvoker /<code>doSelect(List<invoker>> invokers, URL url, Invocation invocation) {
int length = invokers.size(); // 總個數
int totalWeight = 0; // 總權重
boolean sameWeight = true; // 權重是否都一樣
for (int i = 0; i < length; i++) {
int weight = getWeight(invokers.get(i), invocation);
totalWeight += weight; // 累計總權重
if (sameWeight && i > 0
&& weight != getWeight(invokers.get(i - 1), invocation)) {
sameWeight = false; // 計算所有權重是否一樣
}
}
if (totalWeight > 0 && ! sameWeight) {
// 如果權重不相同且權重大於0則按總權重數隨機
int offset = random.nextInt(totalWeight);
// 並確定隨機值落在哪個片斷上
for (int i = 0; i < length; i++) {
offset -= getWeight(invokers.get(i), invocation);
if (offset < 0) {
return invokers.get(i);
}
}
}
// 如果權重相同或權重為0則均等隨機
return invokers.get(random.nextInt(length));
}
}/<invoker>
說明:從源碼可以看出隨機負載均衡的策略分為兩種情況
a. 如果總權重大於0並且權重不相同,就生成一個1~totalWeight(總權重數)的隨機數,然後再把隨機數和所有的權重值一一相減得到一個新的隨機數,直到隨機 數小於0,那麼此時訪問的服務器就是使得隨機數小於0的權重所在的機器
b. 如果權重相同或者總權重數為0,就生成一個1~length(權重的總個數)的隨機數,此時所訪問的機器就是這個隨機數對應的權重所在的機器
2、RoundRobin LoadBalance
2.1 輪循,按公約後的權重設置輪循比率。
2.2 存在慢的提供者累積請求的問題,比如:第二臺機器很慢,但沒掛,當請求調到第二臺時就卡在那,久而久之,所有請求都卡在調到第二臺上。
2.3 源碼分析
<code>package com.alibaba.dubbo.rpc.cluster.loadbalance;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.common.utils.AtomicPositiveInteger;
import com.alibaba.dubbo.rpc.Invocation;
import com.alibaba.dubbo.rpc.Invoker;
/**
* Round robin load balance.
*
* @author qian.lei
* @author william.liangf
*/
public class RoundRobinLoadBalance extends AbstractLoadBalance {
public static final String NAME = "roundrobin";
private final ConcurrentMap<string> sequences = new ConcurrentHashMap<string>();
private final ConcurrentMap<string> weightSequences = new ConcurrentHashMap<string>();
protectedInvoker /<string>/<string>/<string>/<string>/<code>doSelect(List<invoker>> invokers, URL url, Invocation invocation) {
String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
int length = invokers.size(); // 總個數
int maxWeight = 0; // 最大權重
int minWeight = Integer.MAX_VALUE; // 最小權重
for (int i = 0; i < length; i++) {
int weight = getWeight(invokers.get(i), invocation);
maxWeight = Math.max(maxWeight, weight); // 累計最大權重
minWeight = Math.min(minWeight, weight); // 累計最小權重
}
if (maxWeight > 0 && minWeight < maxWeight) { // 權重不一樣
AtomicPositiveInteger weightSequence = weightSequences.get(key);
if (weightSequence == null) {
weightSequences.putIfAbsent(key, new AtomicPositiveInteger());
weightSequence = weightSequences.get(key);
}
int currentWeight = weightSequence.getAndIncrement() % maxWeight;
List<invoker>> weightInvokers = new ArrayList<invoker>>();
for (Invokerinvoker : invokers) { // 篩選權重大於當前權重基數的Invoker /<invoker>/<invoker>/<invoker>
if (getWeight(invoker, invocation) > currentWeight) {
weightInvokers.add(invoker);
}
}
int weightLength = weightInvokers.size();
if (weightLength == 1) {
return weightInvokers.get(0);
} else if (weightLength > 1) {
invokers = weightInvokers;
length = invokers.size();
}
}
AtomicPositiveInteger sequence = sequences.get(key);
if (sequence == null) {
sequences.putIfAbsent(key, new AtomicPositiveInteger());
sequence = sequences.get(key);
}
// 取模輪循
return invokers.get(sequence.getAndIncrement() % length);
}
}
說明:從源碼可以看出輪循負載均衡的算法是:
a. 如果權重不一樣時,獲取一個當前的權重基數,然後從權重集合中篩選權重大於當前權重基數的集合,如果篩選出的集合的長度為1,此時所訪問的機器就是集合裡面的權重對應的機器
b. 如果權重一樣時就取模輪循
3、LeastActive LoadBalance
3.1 最少活躍調用數,相同活躍數的隨機,活躍數指調用前後計數差(調用前的時刻減去響應後的時刻的值)。
3.2 使慢的提供者收到更少請求,因為越慢的提供者的調用前後計數差會越大
3.3 對應的源碼
<code>package com.alibaba.dubbo.rpc.cluster.loadbalance;
import java.util.List;
import java.util.Random;
import com.alibaba.dubbo.common.Constants;
import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.rpc.Invocation;
import com.alibaba.dubbo.rpc.Invoker;
import com.alibaba.dubbo.rpc.RpcStatus;
/**
* LeastActiveLoadBalance
*
* @author william.liangf
*/
public class LeastActiveLoadBalance extends AbstractLoadBalance {
public static final String NAME = "leastactive";
private final Random random = new Random();
protectedInvoker /<code>doSelect(List<invoker>> invokers, URL url, Invocation invocation) {
int length = invokers.size(); // 總個數
int leastActive = -1; // 最小的活躍數
int leastCount = 0; // 相同最小活躍數的個數
int[] leastIndexs = new int[length]; // 相同最小活躍數的下標
int totalWeight = 0; // 總權重
int firstWeight = 0; // 第一個權重,用於於計算是否相同
boolean sameWeight = true; // 是否所有權重相同
for (int i = 0; i < length; i++) {
Invokerinvoker = invokers.get(i); /<invoker>
int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive(); // 活躍數
int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT); // 權重
if (leastActive == -1 || active < leastActive) { // 發現更小的活躍數,重新開始
leastActive = active; // 記錄最小活躍數
leastCount = 1; // 重新統計相同最小活躍數的個數
leastIndexs[0] = i; // 重新記錄最小活躍數下標
totalWeight = weight; // 重新累計總權重
firstWeight = weight; // 記錄第一個權重
sameWeight = true; // 還原權重相同標識
} else if (active == leastActive) { // 累計相同最小的活躍數
leastIndexs[leastCount ++] = i; // 累計相同最小活躍數下標
totalWeight += weight; // 累計總權重
// 判斷所有權重是否一樣
if (sameWeight && i > 0
&& weight != firstWeight) {
sameWeight = false;
}
}
}
// assert(leastCount > 0)
if (leastCount == 1) {
// 如果只有一個最小則直接返回
return invokers.get(leastIndexs[0]);
}
if (! sameWeight && totalWeight > 0) {
// 如果權重不相同且權重大於0則按總權重數隨機
int offsetWeight = random.nextInt(totalWeight);
// 並確定隨機值落在哪個片斷上
for (int i = 0; i < leastCount; i++) {
int leastIndex = leastIndexs[i];
offsetWeight -= getWeight(invokers.get(leastIndex), invocation);
if (offsetWeight <= 0)
return invokers.get(leastIndex);
}
}
// 如果權重相同或權重為0則均等隨機
return invokers.get(leastIndexs[random.nextInt(leastCount)]);
}
}
說明:源碼裡面的註釋已經很清晰了,大致的意思就是活躍數越小的的機器分配到的請求越多
4、ConsistentHash LoadBalance
4.1 一致性 Hash,相同參數的請求總是發到同一提供者。
4.2 當某一臺提供者掛時,原本發往該提供者的請求,基於虛擬節點,平攤到其它提供者,不會引起劇烈變動。
4.3 缺省只對第一個參數 Hash,如果要修改,請配置 <parameter>
4.4 缺省用 160 份虛擬節點,如果要修改,請配置 <parameter>
4.5 源碼分析
<code>/*
* Copyright 1999-2012 Alibaba Group.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.alibaba.dubbo.rpc.cluster.loadbalance;
import java.io.UnsupportedEncodingException;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.List;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import com.alibaba.dubbo.common.Constants;
import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.rpc.Invocation;
import com.alibaba.dubbo.rpc.Invoker;
/**
* ConsistentHashLoadBalance
*
* @author william.liangf
*/
public class ConsistentHashLoadBalance extends AbstractLoadBalance {
private final ConcurrentMap<string>> selectors = new ConcurrentHashMap<string>>();
@SuppressWarnings("unchecked")
@Override
protectedInvoker /<string>/<string>/<code>doSelect(List<invoker>> invokers, URL url, Invocation invocation) {
String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
int identityHashCode = System.identityHashCode(invokers);
ConsistentHashSelectorselector = (ConsistentHashSelector /<invoker>) selectors.get(key);
if (selector == null || selector.getIdentityHashCode() != identityHashCode) {
selectors.put(key, new ConsistentHashSelector(invokers, invocation.getMethodName(), identityHashCode));
selector = (ConsistentHashSelector) selectors.get(key);
}
return selector.select(invocation);
}
private static final class ConsistentHashSelector{
private final TreeMap<long>> virtualInvokers;
private final int replicaNumber;
private final int identityHashCode;
private final int[] argumentIndex;
public ConsistentHashSelector(List<invoker>> invokers, String methodName, int identityHashCode) {
this.virtualInvokers = new TreeMap<long>>();
this.identityHashCode = System.identityHashCode(invokers);
URL url = invokers.get(0).getUrl();
this.replicaNumber = url.getMethodParameter(methodName, "hash.nodes", 160);
String[] index = Constants.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 (Invokerinvoker : invokers) { /<long>/<invoker>/<long>
for (int i = 0; i < replicaNumber / 4; i++) {
byte[] digest = md5(invoker.getUrl().toFullString() + i);
for (int h = 0; h < 4; h++) {
long m = hash(digest, h);
virtualInvokers.put(m, invoker);
}
}
}
}
public int getIdentityHashCode() {
return identityHashCode;
}
public Invokerselect(Invocation invocation) {
String key = toKey(invocation.getArguments());
byte[] digest = md5(key);
Invokerinvoker = sekectForKey(hash(digest, 0));
return invoker;
}
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 InvokersekectForKey(long hash) {
Invokerinvoker;
Long key = hash;
if (!virtualInvokers.containsKey(key)) {
SortedMap<long>> tailMap = virtualInvokers.tailMap(key);
if (tailMap.isEmpty()) {
key = virtualInvokers.firstKey();
} else {
key = tailMap.firstKey();
}
}
invoker = virtualInvokers.get(key);
return invoker;
}
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[0 + 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 = null;
try {
bytes = value.getBytes("UTF-8");
} catch (UnsupportedEncodingException e) {
throw new IllegalStateException(e.getMessage(), e);
}
md5.update(bytes);
return md5.digest();
}
}
}/<long>
說明:根據傳遞的參數進行hash然後調用服務,如果兩次傳遞的參數一樣就調用的是同一個機器上的服務.
5、dubbo官方的文檔的負載均衡配置示例
服務端服務級別.
<code> <service>/<code>
客戶端服務級別.
<code> <reference>/<code>
服務端方法級別.
<code> <service>
<method>
/<service>/<code>
客戶端方法級別.
<code> <reference>
<method>
/<reference>/<code>
以上就是dubbo負載均衡策略及對應源碼分析,希望可以幫助到你,下面展示了部分資料,也希望也能幫助到大家,對編程感興趣想進階的朋友,如果能幫到你請點贊、點贊、點贊:
整理的 pdf 文檔:
源碼分析專題部分課程:
獲取方式
點贊,收藏並轉發文章後點擊小編頭像或暱稱,關注後私信回覆:【11】 即可
舉手之勞,非常感謝!!!
來源:https://www.cnblogs.com/leeSmall/p/7620467.html
閱讀更多 java架構師小旭 的文章