dubbo的負載均衡所有由AbstractLoadBalance的子類來實現java
在一個截面上碰撞的機率高,但調用量越大分佈越均勻,並且按機率使用權重後也比較均勻,有利於動態調整提供者權重。node
@Override protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) { int length = invokers.size(); // Number of invokers boolean sameWeight = true; // Every invoker has the same weight? int firstWeight = getWeight(invokers.get(0), invocation); int totalWeight = firstWeight; // The sum of weights for (int i = 1; i < length; i++) { int weight = getWeight(invokers.get(i), invocation); totalWeight += weight; // Sum 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 -= getWeight(invokers.get(i), invocation); 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)); }
存在慢的提供者累積請求的問題,好比:第二臺機器很慢,但沒掛,當請求調到第二臺時就卡在那,長此以往,全部請求都卡在調到第二臺上。算法
在老的版本上,dubbo會求出最大權重和最小權重,若是權重相等,那麼就直接按取模的方式,每次取完後值加一;若是權重不相等,順序根據權重分配。數組
在新的版本上,對這個類進行了重構。app
這樣看顯然是不夠清晰的,咱們來舉個例子:負載均衡
假定有3臺dubbo provider: 10.0.0.1:20884, weight=2 10.0.0.1:20886, weight=3 10.0.0.1:20888, weight=4 totalWeight=9; 那麼第一次調用的時候: 10.0.0.1:20884, weight=2 selectedWRR -> current = 2 10.0.0.1:20886, weight=3 selectedWRR -> current = 3 10.0.0.1:20888, weight=4 selectedWRR -> current = 4 selectedInvoker-> 10.0.0.1:20888 調用 selectedWRR.sel(totalWeight); 10.0.0.1:20888, weight=4 selectedWRR -> current = -5 返回10.0.0.1:20888這個實例 那麼第二次調用的時候: 10.0.0.1:20884, weight=2 selectedWRR -> current = 4 10.0.0.1:20886, weight=3 selectedWRR -> current = 6 10.0.0.1:20888, weight=4 selectedWRR -> current = -1 selectedInvoker-> 10.0.0.1:20886 調用 selectedWRR.sel(totalWeight); 10.0.0.1:20886 , weight=4 selectedWRR -> current = -3 返回10.0.0.1:20886這個實例 那麼第三次調用的時候: 10.0.0.1:20884, weight=2 selectedWRR -> current = 6 10.0.0.1:20886, weight=3 selectedWRR -> current = 0 10.0.0.1:20888, weight=4 selectedWRR -> current = 3 selectedInvoker-> 10.0.0.1:20884 調用 selectedWRR.sel(totalWeight); 10.0.0.1:20884, weight=2 selectedWRR -> current = -3 返回10.0.0.1:20884這個實例
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 (weight < 0) { weight = 0; } 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); } 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); }
使慢的提供者收到更少請求,由於越慢的提供者的調用先後計數差會越大。dom
最小活躍數算法實現: 假定有3臺dubbo provider: 10.0.0.1:20884, weight=2,active=2 10.0.0.1:20886, weight=3,active=4 10.0.0.1:20888, weight=4,active=3 active=2最小,且只有一個2,因此選擇10.0.0.1:20884 假定有3臺dubbo provider: 10.0.0.1:20884, weight=2,active=2 10.0.0.1:20886, weight=3,active=2 10.0.0.1:20888, weight=4,active=3 active=2最小,且有2個,因此從[10.0.0.1:20884,10.0.0.1:20886 ]中選擇; 接下來的算法與隨機算法相似: 假設offset=1(即random.nextInt(5)=1) 1-2=-1<0?是,因此選中 10.0.0.1:20884, weight=2 假設offset=4(即random.nextInt(5)=4) 4-2=2<0?否,這時候offset=2, 2-3<0?是,因此選中 10.0.0.1:20886, weight=3
1: public class LeastActiveLoadBalance extends AbstractLoadBalance { 2: 3: public static final String NAME = "leastactive"; 4: 5: private final Random random = new Random(); 6: 7: @Override 8: protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) { 9: int length = invokers.size(); // 總個數 10: int leastActive = -1; // 最小的活躍數 11: int leastCount = 0; // 相同最小活躍數的個數 12: int[] leastIndexes = new int[length]; // 相同最小活躍數的下標 13: int totalWeight = 0; // 總權重 14: int firstWeight = 0; // 第一個權重,用於於計算是否相同 15: boolean sameWeight = true; // 是否全部權重相同 16: // 計算得到相同最小活躍數的數組和個數 17: for (int i = 0; i < length; i++) { 18: Invoker<T> invoker = invokers.get(i); 19: int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive(); // 活躍數 20: int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT); // 權重 21: if (leastActive == -1 || active < leastActive) { // 發現更小的活躍數,從新開始 22: leastActive = active; // 記錄最小活躍數 23: leastCount = 1; // 從新統計相同最小活躍數的個數 24: leastIndexes[0] = i; // 從新記錄最小活躍數下標 25: totalWeight = weight; // 從新累計總權重 26: firstWeight = weight; // 記錄第一個權重 27: sameWeight = true; // 還原權重相同標識 28: } else if (active == leastActive) { // 累計相同最小的活躍數 29: leastIndexes[leastCount++] = i; // 累計相同最小活躍數下標 30: totalWeight += weight; // 累計總權重 31: // 判斷全部權重是否同樣 32: if (sameWeight && weight != firstWeight) { 33: sameWeight = false; 34: } 35: } 36: } 37: // assert(leastCount > 0) 38: if (leastCount == 1) { 39: // 若是隻有一個最小則直接返回 40: return invokers.get(leastIndexes[0]); 41: } 42: if (!sameWeight && totalWeight > 0) { 43: // 若是權重不相同且權重大於0則按總權重數隨機 44: int offsetWeight = random.nextInt(totalWeight); 45: // 並肯定隨機值落在哪一個片段上 46: for (int i = 0; i < leastCount; i++) { 47: int leastIndex = leastIndexes[i]; 48: offsetWeight -= getWeight(invokers.get(leastIndex), invocation); 49: if (offsetWeight <= 0) { 50: return invokers.get(leastIndex); 51: } 52: } 53: } 54: // 若是權重相同或權重爲0則均等隨機 55: return invokers.get(leastIndexes[random.nextInt(leastCount)]); 56: } 57: 58: }
相同參數的請求老是發到同一提供者。當某一臺提供者掛時,本來發往該提供者的請求,基於虛擬節點,平攤到其它提供者,不會引發劇烈變更。ide
1: public class ConsistentHashLoadBalance extends AbstractLoadBalance { 2: 3: /** 4: * 服務方法與一致性哈希選擇器的映射 5: * 6: * KEY:serviceKey + "." + methodName 7: */ 8: private final ConcurrentMap<String, ConsistentHashSelector<?>> selectors = new ConcurrentHashMap<String, ConsistentHashSelector<?>>(); 9: 10: @SuppressWarnings("unchecked") 11: @Override 12: protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) { 13: String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName(); 14: // 基於 invokers 集合,根據對象內存地址來計算定義哈希值 15: int identityHashCode = System.identityHashCode(invokers); 16: // 得到 ConsistentHashSelector 對象。若爲空,或者定義哈希值變動(說明 invokers 集合發生變化),進行建立新的 ConsistentHashSelector 對象 17: ConsistentHashSelector<T> selector = (ConsistentHashSelector<T>) selectors.get(key); 18: if (selector == null || selector.identityHashCode != identityHashCode) { 19: selectors.put(key, new ConsistentHashSelector<T>(invokers, invocation.getMethodName(), identityHashCode)); 20: selector = (ConsistentHashSelector<T>) selectors.get(key); 21: } 22: return selector.select(invocation); 23: } 24: }
ConsistentHashSelector ,是 ConsistentHashLoadBalance 的內部類,一致性哈希選擇器,基於 Ketama 算法。ui
/** * 虛擬節點與 Invoker 的映射關係 */ private final TreeMap<Long, Invoker<T>> virtualInvokers; /** * 每一個Invoker 對應的虛擬節點數 */ private final int replicaNumber; /** * 定義哈希值 */ private final int identityHashCode; /** * 取值參數位置數組 */ private final int[] argumentIndex; 1: ConsistentHashSelector(List<Invoker<T>> invokers, String methodName, int identityHashCode) { 2: this.virtualInvokers = new TreeMap<Long, Invoker<T>>(); 3: // 設置 identityHashCode 4: this.identityHashCode = identityHashCode; 5: URL url = invokers.get(0).getUrl(); 6: // 初始化 replicaNumber 7: this.replicaNumber = url.getMethodParameter(methodName, "hash.nodes", 160); 8: // 初始化 argumentIndex 9: String[] index = Constants.COMMA_SPLIT_PATTERN.split(url.getMethodParameter(methodName, "hash.arguments", "0")); 10: argumentIndex = new int[index.length]; 11: for (int i = 0; i < index.length; i++) { 12: argumentIndex[i] = Integer.parseInt(index[i]); 13: } 14: // 初始化 virtualInvokers 15: for (Invoker<T> invoker : invokers) { 16: String address = invoker.getUrl().getAddress(); 17: // 每四個虛擬結點爲一組,爲何這樣?下面會說到 18: for (int i = 0; i < replicaNumber / 4; i++) { 19: // 這組虛擬結點獲得唯一名稱 20: byte[] digest = md5(address + i); 21: // Md5是一個16字節長度的數組,將16字節的數組每四個字節一組,分別對應一個虛擬結點,這就是爲何上面把虛擬結點四個劃分一組的緣由 22: for (int h = 0; h < 4; h++) { 23: // 對於每四個字節,組成一個long值數值,作爲這個虛擬節點的在環中的唯一key 24: long m = hash(digest, h); 25: virtualInvokers.put(m, invoker); 26: } 27: } 28: } 29: }
public Invoker<T> select(Invocation invocation) { // 基於方法參數,得到 KEY String key = toKey(invocation.getArguments()); // 計算 MD5 值 byte[] digest = md5(key); // 計算 KEY 值 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) { // 獲得大於當前 key 的那個子 Map ,而後從中取出第一個 key ,就是大於且離它最近的那個 key Map.Entry<Long, Invoker<T>> entry = virtualInvokers.tailMap(hash, true).firstEntry(); // 不存在,則取 virtualInvokers 第一個 if (entry == null) { entry = virtualInvokers.firstEntry(); } // 存在,則返回 return entry.getValue(); }