Dubbo 源码分析 - 集群容错之 LoadBalance
1.簡介
LoadBalance 中文意思為負載均衡，它的職責是將網絡請求，或者其他形式的負載“均攤”到不同的機器上。避免集群中部分服務器壓力過大，而另一些服務器比較空閑的情況。通過負載均衡，可以讓每臺服務器獲取到適合自己處理能力的負載。在為高負載的服務器分流的同時，還可以避免資源浪費，一舉兩得。負載均衡可分為軟件負載均衡和硬件負載均衡。在我們日常開發中，一般很難接觸到硬件負載均衡。但軟件負載均衡還是能夠接觸到一些的，比如 Nginx。在 Dubbo 中，也有負載均衡的概念和相應的實現。Dubbo 需要對服務消費者的調用請求進行分配，避免少數服務提供者負載過大。服務提供者負載過大，會導致部分服務調用超時。因此將負載均衡到每個服務提供者上，是非常必要的。Dubbo 提供了4種負載均衡實現，分別是基于權重隨機算法的 RandomLoadBalance、基于最少活躍調用數算法的 LeastActiveLoadBalance、基于 hash 一致性的 ConsistentHashLoadBalance，以及基于加權輪詢算法的 RoundRobinLoadBalance。這幾個負載均衡算法代碼不是很長，但是想看懂也不是很容易，需要大家對這幾個算法的原理有一定了解才行。如果不是很了解，也沒不用太擔心。我會在分析每個算法的源碼之前，對算法原理進行簡單的講解，幫助大家建立初步的印象。
我在寫 Dubbo 源碼分析系列文章之初，當時 Dubbo 最新的版本為 2.6.4。近期，Dubbo 2.6.5 發布了，其中就有對負載均衡部分代碼修改。因此我在分析完 2.6.4 版本后的源碼后，會另外分析 2.6.5 更新的部分。本篇文章內容非常之豐富，需要大家耐心閱讀。好了，其他的就不多說了，進入正題吧。
?2.源碼分析
在 Dubbo 中，所有負載均衡實現類均繼承自 AbstractLoadBalance，該類實現了 LoadBalance 接口方法，并封裝了一些公共的邏輯。所以在分析負載均衡實現之前，先來看一下 AbstractLoadBalance 的邏輯。首先來看一下負載均衡的入口方法 select，如下：
| 1 2 3 4 5 6 7 8 9 10 11 12 13 | @Override public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) {if (invokers == null || invokers.isEmpty())return null;// 如果 invokers 列表中僅有一個 Invoker，直接返回即可，無需進行負載均衡if (invokers.size() == 1)return invokers.get(0);// 調用 doSelect 方法進行負載均衡，該方法為抽象方法，由子類實現return doSelect(invokers, url, invocation); }protected abstract <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation); |
select 方法的邏輯比較簡單，首先會檢測 invokers 集合的合法性，然后再檢測 invokers 集合元素數量。如果只包含一個 Invoker，直接返回該 Inovker 即可。如果包含多個 Invoker，此時需要通過負載均衡算法選擇一個 Invoker。具體的負載均衡算法由子類實現，接下來章節會對這些子類進行詳細分析。
AbstractLoadBalance 除了實現了 LoadBalance 接口方法，還封裝了一些公共邏輯 —— 服務提供者權重計算邏輯。具體實現如下：
| 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 | protected int getWeight(Invoker<?> invoker, Invocation invocation) {// 從 url 中獲取 weight 配置值int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT);if (weight > 0) {// 獲取服務提供者啟動時間戳long timestamp = invoker.getUrl().getParameter(Constants.REMOTE_TIMESTAMP_KEY, 0L);if (timestamp > 0L) {// 計算服務提供者運行時長int uptime = (int) (System.currentTimeMillis() - timestamp);// 獲取服務預熱時間，默認為10分鐘int warmup = invoker.getUrl().getParameter(Constants.WARMUP_KEY, Constants.DEFAULT_WARMUP);// 如果服務運行時間小于預熱時間，則重新計算服務權重，即降權if (uptime > 0 && uptime < warmup) {// 重新計算服務權重weight = calculateWarmupWeight(uptime, warmup, weight);}}}return weight; }static int calculateWarmupWeight(int uptime, int warmup, int weight) {// 計算權重，下面代碼邏輯上形似于 (uptime / warmup) * weight。// 隨著服務運行時間 uptime 增大，權重計算值 ww 會慢慢接近配置值 weightint ww = (int) ((float) uptime / ((float) warmup / (float) weight));return ww < 1 ? 1 : (ww > weight ? weight : ww); } |
上面是權重的計算過程，該過程主要用于保證當服務運行時長小于服務預熱時間時，對服務進行降權，避免讓服務在啟動之初就處于高負載狀態。服務預熱是一個優化手段，與此類似的還有 JVM 預熱。主要目的是讓服務啟動后“低功率”運行一段時間，使其效率慢慢提升至最佳狀態。關于預熱方面的更多知識，大家感興趣可以自己搜索一下。
關于 AbstractLoadBalance 就先分析到這，接下來分析各個實現類的代碼。首先，我們從 Dubbo 缺省的實現類 RandomLoadBalance 看起。
?2.1 RandomLoadBalance
RandomLoadBalance 是加權隨機算法的具體實現，它的算法思想很簡單。假設我們有一組服務器 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 背后的算法思想，比較簡單，不多說了，下面開始分析源碼。
| 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 | public class RandomLoadBalance extends AbstractLoadBalance {public static final String NAME = "random";private final Random random = new Random();@Overrideprotected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {int length = invokers.size();int totalWeight = 0;boolean sameWeight = true;// 下面這個循環有兩個作用，第一是計算總權重 totalWeight，// 第二是檢測每個服務提供者的權重是否相同，若不相同，則將 sameWeight 置為 falsefor (int i = 0; i < length; i++) {int weight = getWeight(invokers.get(i), invocation);// 累加權重totalWeight += weight;// 檢測當前服務提供者的權重與上一個服務提供者的權重是否相同，// 不相同的話，則將 sameWeight 置為 false。if (sameWeight && i > 0&& weight != getWeight(invokers.get(i - 1), invocation)) {sameWeight = false;}}// 下面的 if 分支主要用于獲取隨機數，并計算隨機數落在哪個區間上if (totalWeight > 0 && !sameWeight) {// 隨機獲取一個 [0, totalWeight) 之間的數字int offset = random.nextInt(totalWeight);// 循環讓 offset 數減去服務提供者權重值，當 offset 小于0時，返回相應的 Invoker。// 還是用上面的例子進行說明，servers = [A, B, C]，weights = [5, 3, 2]，offset = 7。// 第一次循環，offset - 5 = 2 > 0，說明 offset 肯定不會落在服務器 A 對應的區間上。// 第二次循環，offset - 3 = -1 < 0，表明 offset 落在服務器 B 對應的區間上for (int i = 0; i < length; i++) {// 讓隨機值 offset 減去權重值offset -= getWeight(invokers.get(i), invocation);if (offset < 0) {// 返回相應的 Invokerreturn invokers.get(i);}}}// 如果所有服務提供者權重值相同，此時直接隨機返回一個即可return invokers.get(random.nextInt(length));} } |
RandomLoadBalance 的算法思想比較簡單，在經過多次請求后，能夠將調用請求按照權重值進行“均勻”分配。當然 RandomLoadBalance 也存在一定的缺點，當調用次數比較少時，Random 產生的隨機數可能會比較集中，此時多數請求會落到同一臺服務器上。這個缺點并不是很嚴重，多數情況下可以忽略。RandomLoadBalance 是一個簡單，高效的負載均衡實現，因此 Dubbo 選擇它作為缺省實現。
關于 RandomLoadBalance 就先到這了，接下來分析 LeastActiveLoadBalance。
?2.2 LeastActiveLoadBalance
LeastActiveLoadBalance 翻譯過來是最小活躍數負載均衡，所謂的最小活躍數可理解為最少連接數。即服務提供者目前正在處理的請求數（一個請求對應一條連接）最少，表明該服務提供者效率高，單位時間內可處理更多的請求。此時應優先將請求分配給該服務提供者。在具體實現中，每個服務提供者對應一個活躍數 active。初始情況下，所有服務提供者活躍數均為0。每收到一個請求，活躍數加1，完成請求后則將活躍數減1。在服務運行一段時間后，性能好的服務提供者處理請求的速度更快，因此活躍數下降的也越快。此時這樣的服務提供者能夠優先獲取到新的服務請求，這就是最小活躍數負載均衡算法的基本思想。除了最小活躍數，LeastActiveLoadBalance 在實現上還引入了權重值。所以準確的來說，LeastActiveLoadBalance 是基于加權最小活躍數算法實現的。舉個例子說明一下，在一個服務提供者集群中，有兩個性能優異的服務提供者。某一時刻它們的活躍數相同，此時 Dubbo 會根據它們的權重去分配請求，權重越大，獲取到新請求的可能性就越大。如果兩個服務提供者權重相同，此時隨機選擇一個即可。關于 LeastActiveLoadBalance 的背景知識就先介紹到這里，下面開始分析源碼。
| 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 | public class LeastActiveLoadBalance extends AbstractLoadBalance {public static final String NAME = "leastactive";private final Random random = new Random();@Overrideprotected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {int length = invokers.size();// 最小的活躍數int leastActive = -1;// 具有相同“最小活躍數”的服務者提供者（以下用 Invoker 代稱）數量int leastCount = 0; // leastIndexs 用于記錄具有相同“最小活躍數”的 Invoker 在 invokers 列表中的下標信息int[] leastIndexs = new int[length];int totalWeight = 0;// 第一個最小活躍數的 Invoker 權重值，用于與其他具有相同最小活躍數的 Invoker 的權重進行對比，// 以檢測是否所有具有相同最小活躍數的 Invoker 的權重均相等int firstWeight = 0;boolean sameWeight = true;// 遍歷 invokers 列表for (int i = 0; i < length; i++) {Invoker<T> invoker = 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) {// 使用當前活躍數 active 更新最小活躍數 leastActiveleastActive = active;// 更新 leastCount 為 1leastCount = 1;// 記錄當前下標值到 leastIndexs 中leastIndexs[0] = i;totalWeight = weight;firstWeight = weight;sameWeight = true;// 當前 Invoker 的活躍數 active 與最小活躍數 leastActive 相同 } else if (active == leastActive) {// 在 leastIndexs 中記錄下當前 Invoker 在 invokers 集合中的下標leastIndexs[leastCount++] = i;// 累加權重totalWeight += weight;// 檢測當前 Invoker 的權重與 firstWeight 是否相等，// 不相等則將 sameWeight 置為 falseif (sameWeight && i > 0&& weight != firstWeight) {sameWeight = false;}}}// 當只有一個 Invoker 具有最小活躍數，此時直接返回該 Invoker 即可if (leastCount == 1) {return invokers.get(leastIndexs[0]);}// 有多個 Invoker 具有相同的最小活躍數，但他們的權重不同if (!sameWeight && totalWeight > 0) {// 隨機獲取一個 [0, totalWeight) 之間的數字int offsetWeight = random.nextInt(totalWeight);// 循環讓隨機數減去具有最小活躍數的 Invoker 的權重值，// 當 offset 小于等于0時，返回相應的 Invokerfor (int i = 0; i < leastCount; i++) {int leastIndex = leastIndexs[i];// 獲取權重值，并讓隨機數減去權重值 - ??offsetWeight -= getWeight(invokers.get(leastIndex), invocation);if (offsetWeight <= 0)return invokers.get(leastIndex);}}// 如果權重相同或權重為0時，隨機返回一個 Invokerreturn invokers.get(leastIndexs[random.nextInt(leastCount)]);} } |
如上，為了幫助大家理解代碼，我在上面的代碼中寫了大量的注釋。下面簡單總結一下以上代碼所做的事情，如下：
以上就是 LeastActiveLoadBalance 大致的實現邏輯，大家在閱讀的源碼的過程中要注意區分活躍數與權重這兩個概念，不要混為一談。
以上分析是基于 Dubbo 2.6.4 版本進行了，由于近期 Dubbo 2.6.5 發布了，對負載均衡部分的代碼進行了一些更新。這其中就包含了本節分析的 LeastActiveLoadBalance，所以下面簡單說明一下 Dubbo 2.6.5 對 LeastActiveLoadBalance 進行了怎樣的修改。回到上面的源碼中，我在上面的代碼中標注了兩個黃色的五角星??。兩處標記對應的代碼分別如下：
| 1 | int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT); |
| 1 | offsetWeight -= getWeight(invokers.get(leastIndex), invocation); |
問題出在服務預熱階段，第一行代碼直接從 url 中去權重值，未被降權過。第二行代碼獲取到的是經過降權后的權重。第一行代碼獲取到的權重值最終會被累加到權重總和 totalWeight 中，這個時候會導致一個問題。offsetWeight 是一個在 [0, totalWeight) 范圍內的隨機數，而它所減去的是經過降權的權重。很有可能在經過 leastCount 次運算后，offsetWeight 仍然是大于0的，導致無法選中 Invoker。這個問題對應的 issue 為?#904，在 pull request?#2172?中被修復。具體的修復邏輯是將標注一處的代碼修改為：
| 1 2 | // afterWarmup 等價于上面的 weight 變量，這樣命名是為了強調該變量經過 warmup 降權處理了 int afterWarmup = getWeight(invoker, invocation); |
另外，2.6.4 版本中的 LeastActiveLoadBalance 還要一個缺陷，即當一組 Invoker 具有相同的最小活躍數，且其中一個 Invoker 的權重值為1，此時這個 Invoker 無法被選中。缺陷代碼如下：
| 1 2 3 4 5 6 7 | 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); } |
問題就出在了offsetWeight <= 0上，舉例說明，假設有一組 Invoker 的權重為 5、2、1，offsetWeight 最大值為 7。假設 offsetWeight = 7，你會發現，當 for 循環進行第二次遍歷后 offsetWeight = 7 - 5 - 2 = 0，提前返回了。此時，權重為1的 Invoker 就沒有機會被選中。這個修改起來也不難，可以將?offsetWeight < 0，不過 Dubbo 的是將offsetWeight + 1，也就是：
| 1 | int offsetWeight = random.nextInt(totalWeight) + 1; |
兩種改動都行，不過我認為覺得第一種方式更好一點，可與 RandomLoadBalance 邏輯保持一致。這里+1有點突兀，大家讀到這里要特地思考一下為什么要+1。
以上就是 Dubob 2.6.5 對 LeastActiveLoadBalance 的更新，不是很難理解，就不多說了。接下來分析基于一致性 hash 思想的 ConsistentHashLoadBalance。
?2.3 ConsistentHashLoadBalance
一致性 hash 算法由麻省理工學院的 Karger 及其合作者于1997年提供出的，算法提出之初是用于大規模緩存系統的負載均衡。它的工作過程是這樣的，首先根據 ip 獲取其他的信息為緩存節點生成一個 hash，并將這個 hash 投射到 [0, 232?- 1] 的圓環上。當有查詢或寫入請求時，則為緩存項的 key 生成一個 hash 值。然后查找第一個大于或等于該 hash 值的緩存節點，并到這個節點中查詢或寫入緩存項。如果當前節點掛了，則在下一次查詢或寫入緩存時，為緩存項查找另一個大于其 hash 值的緩存節點即可。大致效果如下，每個緩存節點在圓環上占據一個位置。如果緩存項的 key 的 hash 值小于緩存節點 hash 值，則到該緩存節點中存儲或讀取緩存項。比如下面綠色點對應的緩存項存儲到 cache-2 節點中。由于 cache-3 掛了，原本應該存到該節點中的緩存想最終會存儲到 cache-4 節點中。
關于一致性 hash 算法，我這里只做掃盲。具體的細節不討論，大家請自行補充相關的背景知識。下面來看看一致性 hash 在 Dubbo 中的應用。我們把上圖的緩存節點替換成 Dubbo 的服務提供者，于是得到了下圖：
這里相同顏色的節點均屬于同一個服務提供者，比如 Invoker1-1，Invoker1-2，……, Invoker1-160。這樣做的目的是通過引入虛擬節點，讓 Invoker 在圓環上分散開來，避免數據傾斜問題。所謂數據傾斜是指，由于節點不夠分散，導致大量請求落到了同一個節點上，而其他節點只會接收到了少量的請求。比如：
如上，由于 Invoker-1 和 Invoker-2 在圓環上分布不均，導致系統中75%的請求都會落到 Invoker-1 上，只有 25% 的請求會落到 Invoker-2 上。解決這個問題辦法是引入虛擬節點，通過虛擬節點均衡各個節點的請求量。
到這里背景知識就普及完了，接下來開始分析源碼。我們先從 ConsistentHashLoadBalance 的 doSelect 方法開始看起，如下：
| 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 | public class ConsistentHashLoadBalance extends AbstractLoadBalance {private final ConcurrentMap<String, ConsistentHashSelector<?>> selectors = new ConcurrentHashMap<String, ConsistentHashSelector<?>>();@Overrideprotected <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 原始的 hashcodeint identityHashCode = System.identityHashCode(invokers);ConsistentHashSelector<T> selector = (ConsistentHashSelector<T>) selectors.get(key);// 如果 invokers 是一個新的 List 對象，意味著服務提供者數量發生了變化，可能新增也可能減少了。// 此時 selector.identityHashCode != identityHashCode 條件成立if (selector == null || selector.identityHashCode != identityHashCode) {// 創建新的 ConsistentHashSelectorselectors.put(key, new ConsistentHashSelector<T>(invokers, methodName, identityHashCode));selector = (ConsistentHashSelector<T>) selectors.get(key);}// 調用 ConsistentHashSelector 的 select 方法選擇 Invokerreturn selector.select(invocation);}private static final class ConsistentHashSelector<T> {...} } |
如上，doSelect 方法主要做了一些前置工作，比如檢測 invokers 列表是不是變動過，以及創建 ConsistentHashSelector。這些工作做完后，接下來開始調用 select 方法執行負載均衡邏輯。在分析 select 方法之前，我們先來看一下一致性 hash 選擇器 ConsistentHashSelector 的初始化過程，如下：
| 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 | 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();// 獲取虛擬節點數，默認為160this.replicaNumber = url.getMethodParameter(methodName, "hash.nodes", 160);// 獲取參與 hash 計算的參數下標值，默認對第一個參數進行 hash 運算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 (Invoker<T> invoker : invokers) {String address = invoker.getUrl().getAddress();for (int i = 0; i < replicaNumber / 4; i++) {// 對 address + i 進行 md5 運算，得到一個長度為16的字節數組byte[] digest = md5(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);}}}} } |
ConsistentHashSelector 進行了一些列的初始化方法，比如從配置中獲取虛擬節點數以及參與 hash 計算的參數下標，默認情況下只使用第一個參數進行 hash。需要特別說明的是，ConsistentHashLoadBalance 的負載均衡邏輯只受參數值影響，具有相同參數值的請求將會被分配給同一個服務提供者。ConsistentHashLoadBalance 不 care 權重，因此使用時需要注意一下。
在獲取虛擬節點數和參數下標配置后，接下來要做到事情是計算虛擬節點 hash 值，并將虛擬節點存儲到 TreeMap 中。到此，ConsistentHashSelector 初始化工作就完成了。接下來，我們再來看看 select 方法的邏輯。
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 | public Invoker<T> select(Invocation invocation) {// 將參數轉為 keyString key = toKey(invocation.getArguments());// 對參數 key 進行 md5 運算byte[] digest = md5(key);// 取 digest 數組的前四個字節進行 hash 運算，再將 hash 值傳給 selectForKey 方法，// 尋找合適的 Invokerreturn selectForKey(hash(digest, 0)); }private Invoker<T> selectForKey(long hash) {// 到 TreeMap 中查找第一個節點值大于或等于當前 hash 的 InvokerMap.Entry<Long, Invoker<T>> entry = virtualInvokers.tailMap(hash, true).firstEntry();// 如果 hash 大于 Invoker 在圓環上最大的位置，此時 entry = null，// 需要將 TreeMap 的頭結點賦值給 entryif (entry == null) {entry = virtualInvokers.firstEntry();}// 返回 Invokerreturn entry.getValue(); } |
如上，選擇的過程比較簡單了。首先是對參數進行 md5 以及 hash 運算，得到一個 hash 值。然后再拿這個值到 TreeMap 中查找目標 Invoker 即可。
到此關于 ConsistentHashLoadBalance 就分析完了。在閱讀 ConsistentHashLoadBalance 之前，大家一定要先補充背景知識。否者即使這里只有一百多行代碼，也很難看懂。好了，本節先分析到這。
?2.4 RoundRobinLoadBalance
本節，我們來看一下 Dubbo 中的加權輪詢負載均衡的實現 RoundRobinLoadBalance。在詳細分析源碼前，我們還是先來了解一下什么是加權輪詢。這里從最簡單的輪詢開始講起，所謂輪詢就是將請求輪流分配給一組服務器。舉個例子，我們有三臺服務器 A、B、C。我們將第一個請求分配給服務器 A，第二個請求分配給服務器 B，第三個請求分配給服務器 C，第四個請求再次分配給服務器 A。這個過程就叫做輪詢。輪詢是一種無狀態負載均衡算法，實現簡單，適用于每臺服務器性能相近的場景下。顯然，現實情況下，我們并不能保證每臺服務器性能均相近。如果我們將等量的請求分配給性能較差的服務器，這顯然是不合理的。因此，這個時候我們需要加權輪詢算法，對輪詢過程進行干預，使得性能好的服務器可以得到更多的請求，性能差的得到的少一些。每臺服務器能夠得到的請求數比例，接近或等于他們的權重比。比如服務器 A、B、C 權重比為 5:2:1。那么在8次請求中，服務器 A 將獲取到其中的5次請求，服務器 B 獲取到其中的2次請求，服務器 C 則獲取到其中的1次請求。
以上就是加權輪詢的算法思想，搞懂了這個思想，接下來我們就可以分析源碼了。我們先來看一下 2.6.4 版本的 RoundRobinLoadBalance。
| 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 | public class RoundRobinLoadBalance extends AbstractLoadBalance {public static final String NAME = "roundrobin";private final ConcurrentMap<String, AtomicPositiveInteger> sequences = new ConcurrentHashMap<String, AtomicPositiveInteger>();@Overrideprotected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {// key = 全限定類名 + "." + 方法名，比如 com.xxx.DemoService.sayHelloString key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();int length = invokers.size();// 最大權重int maxWeight = 0;// 最小權重int minWeight = Integer.MAX_VALUE;final LinkedHashMap<Invoker<T>, IntegerWrapper> invokerToWeightMap = new LinkedHashMap<Invoker<T>, IntegerWrapper>();// 權重總和int weightSum = 0;// 下面這個循環主要用于查找最大和最小權重，計算權重總和等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 (weight > 0) {// 將 weight 封裝到 IntegerWrapper 中invokerToWeightMap.put(invokers.get(i), new IntegerWrapper(weight));// 累加權重weightSum += weight;}}// 查找 key 對應的對應 AtomicPositiveInteger 實例，為空則創建。// 這里可以把 AtomicPositiveInteger 看成一個黑盒，大家只要知道// AtomicPositiveInteger 用于記錄服務的調用編號即可。至于細節，// 大家如果感興趣，可以自行分析AtomicPositiveInteger sequence = sequences.get(key);if (sequence == null) {sequences.putIfAbsent(key, new AtomicPositiveInteger());sequence = sequences.get(key);}// 獲取當前的調用編號int currentSequence = sequence.getAndIncrement();// 如果 最小權重 < 最大權重，表明服務提供者之間的權重是不相等的if (maxWeight > 0 && minWeight < maxWeight) {// 使用調用編號對權重總和進行取余操作int mod = currentSequence % weightSum;// 進行 maxWeight 次遍歷for (int i = 0; i < maxWeight; i++) {// 遍歷 invokerToWeightMapfor (Map.Entry<Invoker<T>, IntegerWrapper> each : invokerToWeightMap.entrySet()) {// 獲取 Invokerfinal Invoker<T> k = each.getKey();// 獲取權重包裝類 IntegerWrapperfinal IntegerWrapper v = each.getValue();// 如果 mod = 0，且權重大于0，此時返回相應的 Invokerif (mod == 0 && v.getValue() > 0) {return k;}// mod != 0，且權重大于0，此時對權重和 mod 分別進行自減操作if (v.getValue() > 0) {v.decrement();mod--;}}}}// 服務提供者之間的權重相等，此時通過輪詢選擇 Invokerreturn invokers.get(currentSequence % length);}// IntegerWrapper 是一個 int 包裝類，主要包含了一個自減方法。// 與 Integer 不同，Integer 是不可變類，而 IntegerWrapper 是可變類private static final class IntegerWrapper {private int value;public void decrement() {this.value--;}// 省略部分代碼} } |
如上，RoundRobinLoadBalance 的每行代碼都不是很難理解，但是將它們組合到一起之后，好像就看不懂了。這里對上面代碼的主要邏輯進行總結，如下：
以上過程對應的原理不太好解釋，所以下面直接舉例說明把。假設我們有三臺服務器 servers = [A, B, C]，對應的權重為 weights = [2, 5, 1]。接下來對上面的邏輯進行簡單的模擬。
mod = 0：滿足條件，此時直接返回服務器 A
mod = 1：需要進行一次遞減操作才能滿足條件，此時返回服務器 B
mod = 2：需要進行兩次遞減操作才能滿足條件，此時返回服務器 C
mod = 3：需要進行三次遞減操作才能滿足條件，經過遞減后，服務器權重為 [1, 4, 0]，此時返回服務器 A
mod = 4：需要進行四次遞減操作才能滿足條件，經過遞減后，服務器權重為 [0, 4, 0]，此時返回服務器 B
mod = 5：需要進行五次遞減操作才能滿足條件，經過遞減后，服務器權重為 [0, 3, 0]，此時返回服務器 B
mod = 6：需要進行六次遞減操作才能滿足條件，經過遞減后，服務器權重為 [0, 2, 0]，此時返回服務器 B
mod = 7：需要進行七次遞減操作才能滿足條件，經過遞減后，服務器權重為 [0, 1, 0]，此時返回服務器 B
經過8次調用后，我們得到的負載均衡結果為 [A, B, C, A, B, B, B, B]，次數比 A:B:C = 2:5:1，等于權重比。當 sequence = 8 時，mod = 0，此時重頭再來。從上面的模擬過程可以看出，當 mod >= 3 后，服務器 C 就不會被選中了，因為它的權重被減為0了。當 mod >= 4 后，服務器 A 的權重被減為0，此后 A 就不會再被選中。
以上是 2.6.4 版本的 RoundRobinLoadBalance 分析過程，大家如果看不懂，自己可以定義一些權重組合進行模擬。也可以寫點測試用例，進行調試分析，總之不要死看。
2.6.4 版本的 RoundRobinLoadBalance 存在著比較嚴重的性能問題，該問題最初是在?issue #2578?中被反饋出來。問題出在了 Invoker 的返回時機上，RoundRobinLoadBalance 需要在mod == 0 && v.getValue() > 0?條件成立的情況下才會被返回相應的 Invoker。假如 mod 很大，比如 10000，50000，甚至更大時，doSelect 方法需要進行很多次計算才能將 mod 減為0。由此可知，doSelect 的效率與 mod 有關，時間復雜度為 O(mod)。mod 又受最大權重 maxWeight 的影響，因此當某個服務提供者配置了非常大的權重，此時 RoundRobinLoadBalance 會產生比較嚴重的性能問題。這個問題被反饋后，社區很快做了回應。并對 RoundRobinLoadBalance 的代碼進行了重構，將時間復雜度優化至了常量級別。這個優化可以說很好了，下面我們來學習一下優化后的代碼。
| 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 | public class RoundRobinLoadBalance extends AbstractLoadBalance {public static final String NAME = "roundrobin";private final ConcurrentMap<String, AtomicPositiveInteger> sequences = new ConcurrentHashMap<String, AtomicPositiveInteger>();private final ConcurrentMap<String, AtomicPositiveInteger> indexSeqs = new ConcurrentHashMap<String, AtomicPositiveInteger>();@Overrideprotected <T> Invoker<T> doSelect(List<Invoker<T>> 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;final List<Invoker<T>> invokerToWeightList = new ArrayList<>();// 查找最大和最小權重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 (weight > 0) {invokerToWeightList.add(invokers.get(i));}}// 獲取當前服務對應的調用序列對象 AtomicPositiveIntegerAtomicPositiveInteger sequence = sequences.get(key);if (sequence == null) {// 創建 AtomicPositiveInteger，默認值為0sequences.putIfAbsent(key, new AtomicPositiveInteger());sequence = sequences.get(key);}// 獲取下標序列對象 AtomicPositiveIntegerAtomicPositiveInteger indexSeq = indexSeqs.get(key);if (indexSeq == null) {// 創建 AtomicPositiveInteger，默認值為 -1indexSeqs.putIfAbsent(key, new AtomicPositiveInteger(-1));indexSeq = indexSeqs.get(key);}if (maxWeight > 0 && minWeight < maxWeight) {length = invokerToWeightList.size();while (true) {int index = indexSeq.incrementAndGet() % length;int currentWeight = sequence.get() % maxWeight;// 每循環一輪（index = 0），重新計算 currentWeightif (index == 0) {currentWeight = sequence.incrementAndGet() % maxWeight;}// 檢測 Invoker 的權重是否大于 currentWeight，大于則返回if (getWeight(invokerToWeightList.get(index), invocation) > currentWeight) {return invokerToWeightList.get(index);}}}// 所有 Invoker 權重相等，此時進行普通的輪詢即可return invokers.get(sequence.incrementAndGet() % length);} } |
上面代碼的邏輯是這樣的，每進行一輪循環，重新計算 currentWeight。如果當前 Invoker 權重大于 currentWeight，則返回該 Invoker。還是舉例說明吧，假設服務器 [A, B, C] 對應權重 [5, 2, 1]。
第一輪循環，currentWeight = 1，可返回 A 和 B
第二輪循環，currentWeight = 2，返回 A
第三輪循環，currentWeight = 3，返回 A
第四輪循環，currentWeight = 4，返回 A
第五輪循環，currentWeight = 0，返回 A, B, C
如上，這里的一輪循環是指 index 再次變為0所經歷過的循環，這里可以把 index = 0 看做是一輪循環的開始。每一輪循環的次數與 Invoker 的數量有關，Invoker 數量通常不會太多，所以我們可以認為上面代碼的時間復雜度為常數級。
重構后的 RoundRobinLoadBalance 看起來已經很不錯了，但是在代碼更新不久后，很有又被重構了。這次重構原因是新的 RoundRobinLoadBalance 在某些情況下選出的服務器序列不夠均勻。比如，服務器 [A, B, C] 對應權重 [5, 1, 1]。現在進行7次負載均衡，選擇出來的序列為 [A, A, A, A, A, B, C]。前5個請求全部都落在了服務器 A上，分布不夠均勻。這將會使服務器 A 短時間內接收大量的請求，壓力陡增。而 B 和 C 無請求，處于空閑狀態。我們期望的結果是這樣的 [A, A, B, A, C, A, A]，不同服務器可以穿插獲取請求。為了增加負載均衡結果的平滑性，社區再次對 RoundRobinLoadBalance 的實現進行了重構。這次重構參考自 Nginx 的平滑加權輪詢負載均衡，實現原理是這樣的。每個服務器對應兩個權重，分別為 weight 和 currentWeight。其中 weight 是固定的，currentWeight 是會動態調整，初始值為0。當有新的請求進來時，遍歷服務器列表，讓它的 currentWeight 加上自身權重。遍歷完成后，找到最大的 currentWeight，并將其減去權重總和，然后返回相應的服務器即可。
上面描述不是很好理解，下面還是舉例說明吧。仍然使用服務器 [A, B, C] 對應權重 [5, 1, 1] 的例子進行說明，現在有7個請求依次進入負載均衡邏輯，選擇過程如下：
| 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] |
如上，經過平滑性處理后，得到的服務器序列為 [A, A, B, A, C, A, A]，相比之前的序列 [A, A, A, A, A, B, C]，分布性要好一些。初始情況下 currentWeight = [0, 0, 0]，第7個請求處理完后，currentWeight 再次變為 [0, 0, 0]，是不是很神奇。這個結果也不難理解，在7次計算過程中，每個服務器的 currentWeight 都增加了自身權重 weight * 7，得到 currentWeight = [35, 7, 7]，A 被選中5次，要被減去 5 * 7。B 和 C 被選中1次，要被減去 1 * 7。于是 currentWeight = [35, 7, 7] - [35, 7, 7] = [0, 0, 0]。
以上就是平滑加權輪詢的計算過程，現在大家應該對平滑加權輪詢算法了有了一些了解。接下來，我們來看看 Dubbo-2.6.5 是如何實現上面的計算過程的。
| 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 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 | public class RoundRobinLoadBalance extends AbstractLoadBalance {public static final String NAME = "roundrobin";private static int RECYCLE_PERIOD = 60000;protected static class WeightedRoundRobin {// 服務提供者權重private int weight;// 當前權重private AtomicLong current = new AtomicLong(0);// 最后一次更新時間private long lastUpdate;public void setWeight(int weight) {this.weight = weight;// 初始情況下，current = 0current.set(0);}public long increaseCurrent() {// current = current + weight；return current.addAndGet(weight);}public void sel(int total) {// current = current - total;current.addAndGet(-1 * total);}}// 嵌套 Map 結構，存儲的數據結構示例如下：// {// "UserService.query": {// "url1": WeightedRoundRobin@123, // "url2": WeightedRoundRobin@456, // },// "UserService.update": {// "url1": WeightedRoundRobin@123, // "url2": WeightedRoundRobin@456,// }// }// 最外層為服務類名 + 方法名，第二層為 url 到 WeightedRoundRobin 的映射關系。// 這里我們可以將 url 看成是服務提供者的 idprivate ConcurrentMap<String, ConcurrentMap<String, WeightedRoundRobin>> methodWeightMap = new ConcurrentHashMap<String, ConcurrentMap<String, WeightedRoundRobin>>();// 原子更新鎖private AtomicBoolean updateLock = new AtomicBoolean();@Overrideprotected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();// 獲取 url 到 WeightedRoundRobin 映射表，如果為空，則創建一個新的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;// 下面這個循環主要做了這樣幾件事情：// 1. 遍歷 Invoker 列表，檢測當前 Invoker 是否有// 對應的 WeightedRoundRobin，沒有則創建// 2. 檢測 Invoker 權重是否發生了變化，若變化了，// 則更新 WeightedRoundRobin 的 weight 字段// 3. 讓 current 字段加上自身權重，等價于 current += weight// 4. 設置 lastUpdate 字段，即 lastUpdate = now// 5. 尋找具有最大 current 的 Invoker 以及 WeightedRoundRobin，// 暫存起來，留作后用// 6. 計算權重總和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;}// 檢測當前 Invoker 是否有對應的 WeightedRoundRobin，沒有則創建if (weightedRoundRobin == null) {weightedRoundRobin = new WeightedRoundRobin();// 設置 Invoker 權重weightedRoundRobin.setWeight(weight);// 存儲 url 唯一標識 identifyString 到 weightedRoundRobin 的映射關系map.putIfAbsent(identifyString, weightedRoundRobin);weightedRoundRobin = map.get(identifyString);}// Invoker 權重不等于 WeightedRoundRobin 中保存的權重，說明權重變化了，此時進行更新if (weight != weightedRoundRobin.getWeight()) {weightedRoundRobin.setWeight(weight);}// 讓 current 加上自身權重，等價于 current += weightlong cur = weightedRoundRobin.increaseCurrent();// 設置 lastUpdate，表示近期更新過weightedRoundRobin.setLastUpdate(now);// 找出最大的 current if (cur > maxCurrent) {maxCurrent = cur;// 將具有最大 current 權重的 Invoker 賦值給 selectedInvokerselectedInvoker = invoker;// 將 Invoker 對應的 weightedRoundRobin 賦值給 selectedWRR，留作后用selectedWRR = weightedRoundRobin;}// 計算權重總和totalWeight += weight;}// 對 <identifyString, WeightedRoundRobin> 進行檢查，過濾掉長時間未被更新的節點。// 該節點可能掛了，invokers 中不包含該節點，所以該節點的 lastUpdate 長時間無法被更新。// 若未更新時長超過閾值后，就會被移除掉，默認閾值為60秒。if (!updateLock.get() && invokers.size() != map.size()) {if (updateLock.compareAndSet(false, true)) {try {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) {// 讓 current 減去權重總和，等價于 current -= totalWeightselectedWRR.sel(totalWeight);// 返回具有最大 current 的 Invokerreturn selectedInvoker;}// should not happen herereturn invokers.get(0);} } |
以上就是 Dubbo-2.6.5 版本的 RoundRobinLoadBalance，大家如果能夠理解平滑加權輪詢算法的計算過程，再配合我寫的注釋，理解上面的代碼應該不難。
以上就是關于 RoundRobinLoadBalance 全部的分析，內容有點多，大家慢慢消化吧。好了，本節先到這。
?3.總結
本篇文章對 Dubbo 中的幾種負載均衡實現進行了詳細的分析，總的來說，這篇文章寫的還是有點累的。主要是每介紹一種負載均衡實現，就要介紹一下相關背景。另一方面，這里很多東西對于我來說，也完全是新的。在此之前，我對負載均衡算法并沒太多了解。這篇文章基本上是邊學邊寫的，總共耗時5天。本來想簡單寫寫算了，但最后還是決定寫詳細點。好在，現在寫完了，我也可以放松一下了。
本篇文章是我的 Dubbo 源碼分析系列文章關于集群容錯部分的最后一篇文章，寫完感覺學到了很多東西。通過堅持不懈的閱讀代碼，寫技術文章，使得我對 Dubbo 有了更深入的了解。當然，這還遠遠不夠。后續還有很多東西要了解，比如 Nacos、Sentinel 等。長路漫漫，步履不停。
好了，本篇文章到這里就結束了。感謝大家的閱讀。
?參考
- 負載均衡之加權輪詢算法 - CSDN
- dubbo源碼-預熱warmup過程 - 簡書
- 一致性哈希算法原理 - 博客園
?附錄：Dubbo 源碼分析系列文章
| 2018-10-01 | Dubbo 源碼分析 - SPI 機制 |
| 2018-10-13 | Dubbo 源碼分析 - 自適應拓展原理 |
| 2018-10-31 | Dubbo 源碼分析 - 服務導出 |
| 2018-11-12 | Dubbo 源碼分析 - 服務引用 |
| 2018-11-17 | Dubbo 源碼分析 - 集群容錯之 Directory |
| 2018-11-20 | Dubbo 源碼分析 - 集群容錯之 Router |
| 2018-11-24 | Dubbo 源碼分析 - 集群容錯之 Cluster |
| 2018-11-29 | Dubbo 源碼分析 - 集群容錯之 LoadBalance |
- 本文鏈接：?https://www.tianxiaobo.com/2018/11/29/Dubbo-源碼分析-集群容錯之-LoadBalance/
http://www.tianxiaobo.com/2018/11/29/Dubbo-%E6%BA%90%E7%A0%81%E5%88%86%E6%9E%90-%E9%9B%86%E7%BE%A4%E5%AE%B9%E9%94%99%E4%B9%8B-LoadBalance/?
總結
以上是生活随笔為你收集整理的Dubbo 源码分析 - 集群容错之 LoadBalance的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： Dubbo 源码分析 - 集群容错之 C
- 下一篇： 消息中间件和JMS介绍