Dubbo源碼分析6之負(fù)載均衡
1.簡(jiǎn)介
LoadBalance 中文意思為負(fù)載均衡，它的職責(zé)是將網(wǎng)絡(luò)請(qǐng)求，或者其他形式的負(fù)載“均攤”到不同的機(jī)器上。避免集群中部分服務(wù)器壓力過(guò)大，而另一些服務(wù)器比較空閑的情況。通過(guò)負(fù)載均衡，可以讓每臺(tái)服務(wù)器獲取到適合自己處理能力的負(fù)載。在為高負(fù)載服務(wù)器分流的同時(shí)，還可以避免資源浪費(fèi)，一舉兩得。負(fù)載均衡可分為軟件負(fù)載均衡和硬件負(fù)載均衡。在我們?nèi)粘ｉ_(kāi)發(fā)中，一般很難接觸到硬件負(fù)載均衡。但軟件負(fù)載均衡還是可以接觸到的，比如 Nginx。在 Dubbo 中，也有負(fù)載均衡的概念和相應(yīng)的實(shí)現(xiàn)。Dubbo 需要對(duì)服務(wù)消費(fèi)者的調(diào)用請(qǐng)求進(jìn)行分配，避免少數(shù)服務(wù)提供者負(fù)載過(guò)大。服務(wù)提供者負(fù)載過(guò)大，會(huì)導(dǎo)致部分請(qǐng)求超時(shí)。因此將負(fù)載均衡到每個(gè)服務(wù)提供者上，是非常必要的。Dubbo 提供了4種負(fù)載均衡實(shí)現(xiàn)，分別是基于權(quán)重隨機(jī)算法的 RandomLoadBalance、基于最少活躍調(diào)用數(shù)算法的 LeastActiveLoadBalance、基于 hash 一致性的 ConsistentHashLoadBalance，以及基于加權(quán)輪詢算法的 RoundRobinLoadBalance。這幾個(gè)負(fù)載均衡算法代碼不是很長(zhǎng)，但是想看懂也不是很容易，需要大家對(duì)這幾個(gè)算法的原理有一定了解才行。如果不是很了解，也沒(méi)不用太擔(dān)心。我們會(huì)在分析每個(gè)算法的源碼之前，對(duì)算法原理進(jìn)行簡(jiǎn)單的講解，幫助大家建立初步的印象。
本系列文章在編寫(xiě)之初是基于 Dubbo 2.6.4 的，近期，Dubbo 2.6.5 發(fā)布了，其中就有針對(duì)對(duì)負(fù)載均衡部分的優(yōu)化。因此我們?cè)诜治鐾?2.6.4 版本后的源碼后，會(huì)另外分析 2.6.5 更新的部分。其他的就不多說(shuō)了，進(jìn)入正題吧。
2.源碼分析
在 Dubbo 中，所有負(fù)載均衡實(shí)現(xiàn)類均繼承自 AbstractLoadBalance，該類實(shí)現(xiàn)了 LoadBalance 接口，并封裝了一些公共的邏輯。所以在分析負(fù)載均衡實(shí)現(xiàn)之前，先來(lái)看一下 AbstractLoadBalance 的邏輯。首先來(lái)看一下負(fù)載均衡的入口方法 select，如下：
@Override
public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) {
if (invokers == null || invokers.isEmpty())
return null;
// 如果 invokers 列表中僅有一個(gè) Invoker，直接返回即可，無(wú)需進(jìn)行負(fù)載均衡
if (invokers.size() == 1)
return invokers.get(0);
// 調(diào)用 doSelect 方法進(jìn)行負(fù)載均衡，該方法為抽象方法，由子類實(shí)現(xiàn)
return doSelect(invokers, url, invocation);
}
protected abstract <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation);
select 方法的邏輯比較簡(jiǎn)單，首先會(huì)檢測(cè) invokers 集合的合法性，然后再檢測(cè) invokers 集合元素?cái)?shù)量。如果只包含一個(gè) Invoker，直接返回該 Inovker 即可。如果包含多個(gè) Invoker，此時(shí)需要通過(guò)負(fù)載均衡算法選擇一個(gè) Invoker。具體的負(fù)載均衡算法由子類實(shí)現(xiàn)，接下來(lái)章節(jié)會(huì)對(duì)這些子類一一進(jìn)行詳細(xì)分析。
AbstractLoadBalance 除了實(shí)現(xiàn)了 LoadBalance 接口方法，還封裝了一些公共邏輯，比如服務(wù)提供者權(quán)重計(jì)算邏輯。具體實(shí)現(xiàn)如下：
protected int getWeight(Invoker<?> invoker, Invocation invocation) {
// 從 url 中獲取權(quán)重 weight 配置值
int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT);
if (weight > 0) {
// 獲取服務(wù)提供者啟動(dòng)時(shí)間戳
long timestamp = invoker.getUrl().getParameter(Constants.REMOTE_TIMESTAMP_KEY, 0L);
if (timestamp > 0L) {
// 計(jì)算服務(wù)提供者運(yùn)行時(shí)長(zhǎng)
int uptime = (int) (System.currentTimeMillis() - timestamp);
// 獲取服務(wù)預(yù)熱時(shí)間，默認(rèn)為10分鐘
int warmup = invoker.getUrl().getParameter(Constants.WARMUP_KEY, Constants.DEFAULT_WARMUP);
// 如果服務(wù)運(yùn)行時(shí)間小于預(yù)熱時(shí)間，則重新計(jì)算服務(wù)權(quán)重，即降權(quán)
if (uptime > 0 && uptime < warmup) {
// 重新計(jì)算服務(wù)權(quán)重
weight = calculateWarmupWeight(uptime, warmup, weight);
}
}
}
return weight;
}
static int calculateWarmupWeight(int uptime, int warmup, int weight) {
// 計(jì)算權(quán)重，下面代碼邏輯上形似于 (uptime / warmup) * weight。
// 隨著服務(wù)運(yùn)行時(shí)間 uptime 增大，權(quán)重計(jì)算值 ww 會(huì)慢慢接近配置值 weight
int ww = (int) ((float) uptime / ((float) warmup / (float) weight));
return ww < 1 ? 1 : (ww > weight ? weight : ww);
}
上面是權(quán)重的計(jì)算過(guò)程，該過(guò)程主要用于保證當(dāng)服務(wù)運(yùn)行時(shí)長(zhǎng)小于服務(wù)預(yù)熱時(shí)間時(shí)，對(duì)服務(wù)進(jìn)行降權(quán)，避免讓服務(wù)在啟動(dòng)之初就處于高負(fù)載狀態(tài)。服務(wù)預(yù)熱是一個(gè)優(yōu)化手段，與此類似的還有 JVM 預(yù)熱。主要目的是讓服務(wù)啟動(dòng)后“低功率”運(yùn)行一段時(shí)間，使其效率慢慢提升至最佳狀態(tài)。
關(guān)于 AbstractLoadBalance 就先分析到這，接下來(lái)分析各個(gè)實(shí)現(xiàn)類的代碼。首先，我們從 Dubbo 缺省的實(shí)現(xiàn)類 RandomLoadBalance 看起。
2.1 RandomLoadBalance
RandomLoadBalance 是加權(quán)隨機(jī)算法的具體實(shí)現(xiàn)，它的算法思想很簡(jiǎn)單。假設(shè)我們有一組服務(wù)器 servers = [A, B, C]，他們對(duì)應(yīng)的權(quán)重為 weights = [5, 3, 2]，權(quán)重總和為10?，F(xiàn)在把這些權(quán)重值平鋪在一維坐標(biāo)值上，[0, 5) 區(qū)間屬于服務(wù)器 A，[5, 8) 區(qū)間屬于服務(wù)器 B，[8, 10) 區(qū)間屬于服務(wù)器 C。接下來(lái)通過(guò)隨機(jī)數(shù)生成器生成一個(gè)范圍在 [0, 10) 之間的隨機(jī)數(shù)，然后計(jì)算這個(gè)隨機(jī)數(shù)會(huì)落到哪個(gè)區(qū)間上。比如數(shù)字3會(huì)落到服務(wù)器 A 對(duì)應(yīng)的區(qū)間上，此時(shí)返回服務(wù)器 A 即可。權(quán)重越大的機(jī)器，在坐標(biāo)軸上對(duì)應(yīng)的區(qū)間范圍就越大，因此隨機(jī)數(shù)生成器生成的數(shù)字就會(huì)有更大的概率落到此區(qū)間內(nèi)。只要隨機(jī)數(shù)生成器產(chǎn)生的隨機(jī)數(shù)分布性很好，在經(jīng)過(guò)多次選擇后，每個(gè)服務(wù)器被選中的次數(shù)比例接近其權(quán)重比例。比如，經(jīng)過(guò)一萬(wàn)次選擇后，服務(wù)器 A 被選中的次數(shù)大約為5000次，服務(wù)器 B 被選中的次數(shù)約為3000次，服務(wù)器 C 被選中的次數(shù)約為2000次。
以上就是 RandomLoadBalance 背后的算法思想，比較簡(jiǎn)單。下面開(kāi)始分析源碼。
public class RandomLoadBalance extends AbstractLoadBalance {
public static final String NAME = "random";
private final Random random = new Random();
@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
int length = invokers.size();
int totalWeight = 0;
boolean sameWeight = true;
// 下面這個(gè)循環(huán)有兩個(gè)作用，第一是計(jì)算總權(quán)重 totalWeight，
// 第二是檢測(cè)每個(gè)服務(wù)提供者的權(quán)重是否相同
for (int i = 0; i < length; i++) {
int weight = getWeight(invokers.get(i), invocation);
// 累加權(quán)重
totalWeight += weight;
// 檢測(cè)當(dāng)前服務(wù)提供者的權(quán)重與上一個(gè)服務(wù)提供者的權(quán)重是否相同，
// 不相同的話，則將 sameWeight 置為 false。
if (sameWeight && i > 0
&& weight != getWeight(invokers.get(i - 1), invocation)) {
sameWeight = false;
}
}
// 下面的 if 分支主要用于獲取隨機(jī)數(shù)，并計(jì)算隨機(jī)數(shù)落在哪個(gè)區(qū)間上
if (totalWeight > 0 && !sameWeight) {
// 隨機(jī)獲取一個(gè) [0, totalWeight) 區(qū)間內(nèi)的數(shù)字
int offset = random.nextInt(totalWeight);
// 循環(huán)讓 offset 數(shù)減去服務(wù)提供者權(quán)重值，當(dāng) offset 小于0時(shí)，返回相應(yīng)的 Invoker。
// 舉例說(shuō)明一下，我們有 servers = [A, B, C]，weights = [5, 3, 2]，offset = 7。
// 第一次循環(huán)，offset - 5 = 2 > 0，即 offset > 5，
// 表明其不會(huì)落在服務(wù)器 A 對(duì)應(yīng)的區(qū)間上。
// 第二次循環(huán)，offset - 3 = -1 < 0，即 5 < offset < 8，
// 表明其會(huì)落在服務(wù)器 B 對(duì)應(yīng)的區(qū)間上
for (int i = 0; i < length; i++) {
// 讓隨機(jī)值 offset 減去權(quán)重值
offset -= getWeight(invokers.get(i), invocation);
if (offset < 0) {
// 返回相應(yīng)的 Invoker
return invokers.get(i);
}
}
}
// 如果所有服務(wù)提供者權(quán)重值相同，此時(shí)直接隨機(jī)返回一個(gè)即可
return invokers.get(random.nextInt(length));
}
}
RandomLoadBalance 的算法思想比較簡(jiǎn)單，在經(jīng)過(guò)多次請(qǐng)求后，能夠?qū)⒄{(diào)用請(qǐng)求按照權(quán)重值進(jìn)行“均勻”分配。當(dāng)然 RandomLoadBalance 也存在一定的缺點(diǎn)，當(dāng)調(diào)用次數(shù)比較少時(shí)，Random 產(chǎn)生的隨機(jī)數(shù)可能會(huì)比較集中，此時(shí)多數(shù)請(qǐng)求會(huì)落到同一臺(tái)服務(wù)器上。這個(gè)缺點(diǎn)并不是很嚴(yán)重，多數(shù)情況下可以忽略。RandomLoadBalance 是一個(gè)簡(jiǎn)單，高效的負(fù)載均衡實(shí)現(xiàn)，因此 Dubbo 選擇它作為缺省實(shí)現(xiàn)。
關(guān)于 RandomLoadBalance 就先到這了，接下來(lái)分析 LeastActiveLoadBalance。
2.2 LeastActiveLoadBalance
LeastActiveLoadBalance 翻譯過(guò)來(lái)是最小活躍數(shù)負(fù)載均衡?；钴S調(diào)用數(shù)越小，表明該服務(wù)提供者效率越高，單位時(shí)間內(nèi)可處理更多的請(qǐng)求。此時(shí)應(yīng)優(yōu)先將請(qǐng)求分配給該服務(wù)提供者。在具體實(shí)現(xiàn)中，每個(gè)服務(wù)提供者對(duì)應(yīng)一個(gè)活躍數(shù) active。初始情況下，所有服務(wù)提供者活躍數(shù)均為0。每收到一個(gè)請(qǐng)求，活躍數(shù)加1，完成請(qǐng)求后則將活躍數(shù)減1。在服務(wù)運(yùn)行一段時(shí)間后，性能好的服務(wù)提供者處理請(qǐng)求的速度更快，因此活躍數(shù)下降的也越快，此時(shí)這樣的服務(wù)提供者能夠優(yōu)先獲取到新的服務(wù)請(qǐng)求、這就是最小活躍數(shù)負(fù)載均衡算法的基本思想。除了最小活躍數(shù)，LeastActiveLoadBalance 在實(shí)現(xiàn)上還引入了權(quán)重值。所以準(zhǔn)確的來(lái)說(shuō)，LeastActiveLoadBalance 是基于加權(quán)最小活躍數(shù)算法實(shí)現(xiàn)的。舉個(gè)例子說(shuō)明一下，在一個(gè)服務(wù)提供者集群中，有兩個(gè)性能優(yōu)異的服務(wù)提供者。某一時(shí)刻它們的活躍數(shù)相同，此時(shí) Dubbo 會(huì)根據(jù)它們的權(quán)重去分配請(qǐng)求，權(quán)重越大，獲取到新請(qǐng)求的概率就越大。如果兩個(gè)服務(wù)提供者權(quán)重相同，此時(shí)隨機(jī)選擇一個(gè)即可。關(guān)于 LeastActiveLoadBalance 的背景知識(shí)就先介紹到這里，下面開(kāi)始分析源碼。
public class LeastActiveLoadBalance extends AbstractLoadBalance {
public static final String NAME = "leastactive";
private final Random random = new Random();
@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
int length = invokers.size();
// 最小的活躍數(shù)
int leastActive = -1;
// 具有相同“最小活躍數(shù)”的服務(wù)者提供者（以下用 Invoker 代稱）數(shù)量
int leastCount = 0;
// leastIndexs 用于記錄具有相同“最小活躍數(shù)”的 Invoker 在 invokers 列表中的下標(biāo)信息
int[] leastIndexs = new int[length];
int totalWeight = 0;
// 第一個(gè)最小活躍數(shù)的 Invoker 權(quán)重值，用于與其他具有相同最小活躍數(shù)的 Invoker 的權(quán)重進(jìn)行對(duì)比，
// 以檢測(cè)是否“所有具有相同最小活躍數(shù)的 Invoker 的權(quán)重”均相等
int firstWeight = 0;
boolean sameWeight = true;
// 遍歷 invokers 列表
for (int i = 0; i < length; i++) {
Invoker<T> invoker = invokers.get(i);
// 獲取 Invoker 對(duì)應(yīng)的活躍數(shù)
int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive();
// 獲取權(quán)重 - ??
int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT);
// 發(fā)現(xiàn)更小的活躍數(shù)，重新開(kāi)始
if (leastActive == -1 || active < leastActive) {
// 使用當(dāng)前活躍數(shù) active 更新最小活躍數(shù) leastActive
leastActive = active;
// 更新 leastCount 為 1
leastCount = 1;
// 記錄當(dāng)前下標(biāo)值到 leastIndexs 中
leastIndexs[0] = i;
totalWeight = weight;
firstWeight = weight;
sameWeight = true;
// 當(dāng)前 Invoker 的活躍數(shù) active 與最小活躍數(shù) leastActive 相同
} else if (active == leastActive) {
// 在 leastIndexs 中記錄下當(dāng)前 Invoker 在 invokers 集合中的下標(biāo)
leastIndexs[leastCount++] = i;
// 累加權(quán)重
totalWeight += weight;
// 檢測(cè)當(dāng)前 Invoker 的權(quán)重與 firstWeight 是否相等，
// 不相等則將 sameWeight 置為 false
if (sameWeight && i > 0
&& weight != firstWeight) {
sameWeight = false;
}
}
}
// 當(dāng)只有一個(gè) Invoker 具有最小活躍數(shù)，此時(shí)直接返回該 Invoker 即可
if (leastCount == 1) {
return invokers.get(leastIndexs[0]);
}
// 有多個(gè) Invoker 具有相同的最小活躍數(shù)，但它們之間的權(quán)重不同
if (!sameWeight && totalWeight > 0) {
// 隨機(jī)生成一個(gè) [0, totalWeight) 之間的數(shù)字
int offsetWeight = random.nextInt(totalWeight);
// 循環(huán)讓隨機(jī)數(shù)減去具有最小活躍數(shù)的 Invoker 的權(quán)重值，
// 當(dāng) offset 小于等于0時(shí)，返回相應(yīng)的 Invoker
for (int i = 0; i < leastCount; i++) {
int leastIndex = leastIndexs[i];
// 獲取權(quán)重值，并讓隨機(jī)數(shù)減去權(quán)重值 - ??
offsetWeight -= getWeight(invokers.get(leastIndex), invocation);
if (offsetWeight <= 0)
return invokers.get(leastIndex);
}
}
// 如果權(quán)重相同或權(quán)重為0時(shí)，隨機(jī)返回一個(gè) Invoker
return invokers.get(leastIndexs[random.nextInt(leastCount)]);
}
}
上面代碼的邏輯比較多，我們?cè)诖a中寫(xiě)了大量的注釋，有幫助大家理解代碼邏輯。下面簡(jiǎn)單總結(jié)一下以上代碼所做的事情，如下：
- 遍歷 invokers 列表，尋找活躍數(shù)最小的 Invoker
- 如果有多個(gè) Invoker 具有相同的最小活躍數(shù)，此時(shí)記錄下這些 Invoker 在 invokers 集合中的下標(biāo)，并累加它們的權(quán)重，比較它們的權(quán)重值是否相等
- 如果只有一個(gè) Invoker 具有最小的活躍數(shù)，此時(shí)直接返回該 Invoker 即可
- 如果有多個(gè) Invoker 具有最小活躍數(shù)，且它們的權(quán)重不相等，此時(shí)處理方式和 RandomLoadBalance 一致
- 如果有多個(gè) Invoker 具有最小活躍數(shù)，但它們的權(quán)重相等，此時(shí)隨機(jī)返回一個(gè)即可
以上就是 LeastActiveLoadBalance 大致的實(shí)現(xiàn)邏輯，大家在閱讀的源碼的過(guò)程中要注意區(qū)分活躍數(shù)與權(quán)重這兩個(gè)概念，不要混為一談。
以上分析是基于 Dubbo 2.6.4 版本進(jìn)行的，由于近期 Dubbo 2.6.5 發(fā)布了，并對(duì) LeastActiveLoadBalance 進(jìn)行了一些修改，下面簡(jiǎn)單來(lái)介紹一下修改內(nèi)容。回到上面的源碼中，我們?cè)谏厦娴拇a中標(biāo)注了兩個(gè)黃色的五角星??。兩處標(biāo)記對(duì)應(yīng)的代碼分別如下：
int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT);
offsetWeight -= getWeight(invokers.get(leastIndex), invocation);
問(wèn)題出在服務(wù)預(yù)熱階段，第一行代碼直接從 url 中取權(quán)重值，未被降權(quán)過(guò)。第二行代碼獲取到的是經(jīng)過(guò)降權(quán)后的權(quán)重。第一行代碼獲取到的權(quán)重值最終會(huì)被累加到權(quán)重總和 totalWeight 中，這個(gè)時(shí)候會(huì)導(dǎo)致一個(gè)問(wèn)題。offsetWeight 是一個(gè)在 [0, totalWeight) 范圍內(nèi)的隨機(jī)數(shù)，而它所減去的是經(jīng)過(guò)降權(quán)的權(quán)重。很有可能在經(jīng)過(guò) leastCount 次運(yùn)算后，offsetWeight 仍然是大于0的，導(dǎo)致無(wú)法選中 Invoker。這個(gè)問(wèn)題對(duì)應(yīng)的 issue 為 #904，并在 pull request #2172 中被修復(fù)。具體的修復(fù)邏輯是將標(biāo)注一處的代碼修改為：
// afterWarmup 等價(jià)于上面的 weight 變量，這樣命名是為了強(qiáng)調(diào)該變量經(jīng)過(guò)了 warmup 降權(quán)處理
int afterWarmup = getWeight(invoker, invocation);
另外，2.6.4 版本中的 LeastActiveLoadBalance 還有一個(gè)缺陷，即當(dāng)一組 Invoker 具有相同的最小活躍數(shù)，且其中一個(gè) Invoker 的權(quán)重值為1，此時(shí)這個(gè) Invoker 無(wú)法被選中。缺陷代碼如下：
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);
}
問(wèn)題出在了offsetWeight <= 0上，舉例說(shuō)明，假設(shè)有一組 Invoker 的權(quán)重為 5、2、1，offsetWeight 最大值為 7。假設(shè) offsetWeight = 7，你會(huì)發(fā)現(xiàn)，當(dāng) for 循環(huán)進(jìn)行第二次遍歷后 offsetWeight = 7 - 5 - 2 = 0，提前返回了。此時(shí)，此時(shí)權(quán)重為1的 Invoker 就沒(méi)有機(jī)會(huì)被選中了。該問(wèn)題在 Dubbo 2.6.5 中被修復(fù)了，修改后的代碼如下：
int offsetWeight = random.nextInt(totalWeight) + 1;
以上就是 Dubbo 2.6.5 對(duì) LeastActiveLoadBalance 的更新，內(nèi)容不是很多，先分析到這。接下來(lái)分析基于一致性 hash 思想的 ConsistentHashLoadBalance。
2.3 ConsistentHashLoadBalance
一致性 hash 算法由麻省理工學(xué)院的 Karger 及其合作者于1997年提出的，算法提出之初是用于大規(guī)模緩存系統(tǒng)的負(fù)載均衡。它的工作過(guò)程是這樣的，首先根據(jù) ip 或者其他的信息為緩存節(jié)點(diǎn)生成一個(gè) hash，并將這個(gè) hash 投射到 [0, 232 - 1] 的圓環(huán)上。當(dāng)有查詢或?qū)懭胝?qǐng)求時(shí)，則為緩存項(xiàng)的 key 生成一個(gè) hash 值。然后查找第一個(gè)大于或等于該 hash 值的緩存節(jié)點(diǎn)，并到這個(gè)節(jié)點(diǎn)中查詢或?qū)懭刖彺骓?xiàng)。如果當(dāng)前節(jié)點(diǎn)掛了，則在下一次查詢或?qū)懭刖彺鏁r(shí)，為緩存項(xiàng)查找另一個(gè)大于其 hash 值的緩存節(jié)點(diǎn)即可。大致效果如下圖所示，每個(gè)緩存節(jié)點(diǎn)在圓環(huán)上占據(jù)一個(gè)位置。如果緩存項(xiàng)的 key 的 hash 值小于緩存節(jié)點(diǎn) hash 值，則到該緩存節(jié)點(diǎn)中存儲(chǔ)或讀取緩存項(xiàng)。比如下面綠色點(diǎn)對(duì)應(yīng)的緩存項(xiàng)將會(huì)被存儲(chǔ)到 cache-2 節(jié)點(diǎn)中。由于 cache-3 掛了，原本應(yīng)該存到該節(jié)點(diǎn)中的緩存項(xiàng)最終會(huì)存儲(chǔ)到 cache-4 節(jié)點(diǎn)中。
下面來(lái)看看一致性 hash 在 Dubbo 中的應(yīng)用。我們把上圖的緩存節(jié)點(diǎn)替換成 Dubbo 的服務(wù)提供者，于是得到了下圖：
這里相同顏色的節(jié)點(diǎn)均屬于同一個(gè)服務(wù)提供者，比如 Invoker1-1，Invoker1-2，……, Invoker1-160。這樣做的目的是通過(guò)引入虛擬節(jié)點(diǎn)，讓 Invoker 在圓環(huán)上分散開(kāi)來(lái)，避免數(shù)據(jù)傾斜問(wèn)題。所謂數(shù)據(jù)傾斜是指，由于節(jié)點(diǎn)不夠分散，導(dǎo)致大量請(qǐng)求落到了同一個(gè)節(jié)點(diǎn)上，而其他節(jié)點(diǎn)只會(huì)接收到了少量請(qǐng)求的情況。比如：
如上，由于 Invoker-1 和 Invoker-2 在圓環(huán)上分布不均，導(dǎo)致系統(tǒng)中75%的請(qǐng)求都會(huì)落到 Invoker-1 上，只有 25% 的請(qǐng)求會(huì)落到 Invoker-2 上。解決這個(gè)問(wèn)題辦法是引入虛擬節(jié)點(diǎn)，通過(guò)虛擬節(jié)點(diǎn)均衡各個(gè)節(jié)點(diǎn)的請(qǐng)求量。
到這里背景知識(shí)就普及完了，接下來(lái)開(kāi)始分析源碼。我們先從 ConsistentHashLoadBalance 的 doSelect 方法開(kāi)始看起，如下：
public class ConsistentHashLoadBalance extends AbstractLoadBalance {
private final ConcurrentMap<String, ConsistentHashSelector<?>> selectors =
new ConcurrentHashMap<String, ConsistentHashSelector<?>>();
@Override
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 identityHashCode = System.identityHashCode(invokers);
ConsistentHashSelector<T> selector = (ConsistentHashSelector<T>) selectors.get(key);
// 如果 invokers 是一個(gè)新的 List 對(duì)象，意味著服務(wù)提供者數(shù)量發(fā)生了變化，可能新增也可能減少了。
// 此時(shí) selector.identityHashCode != identityHashCode 條件成立
if (selector == null || selector.identityHashCode != identityHashCode) {
// 創(chuàng)建新的 ConsistentHashSelector
selectors.put(key, new ConsistentHashSelector<T>(invokers, methodName, identityHashCode));
selector = (ConsistentHashSelector<T>) selectors.get(key);
}
// 調(diào)用 ConsistentHashSelector 的 select 方法選擇 Invoker
return selector.select(invocation);
}
private static final class ConsistentHashSelector<T> {...}
}
如上，doSelect 方法主要做了一些前置工作，比如檢測(cè) invokers 列表是不是變動(dòng)過(guò)，以及創(chuàng)建 ConsistentHashSelector。這些工作做完后，接下來(lái)開(kāi)始調(diào)用 ConsistentHashSelector 的 select 方法執(zhí)行負(fù)載均衡邏輯。在分析 select 方法之前，我們先來(lái)看一下一致性 hash 選擇器 ConsistentHashSelector 的初始化過(guò)程，如下：
private static final class ConsistentHashSelector<T> {
// 使用 TreeMap 存儲(chǔ) Invoker 虛擬節(jié)點(diǎn)
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();
// 獲取虛擬節(jié)點(diǎn)數(shù)，默認(rèn)為160
this.replicaNumber = url.getMethodParameter(methodName, "hash.nodes", 160);
// 獲取參與 hash 計(jì)算的參數(shù)下標(biāo)值，默認(rèn)對(duì)第一個(gè)參數(shù)進(jìn)行 hash 運(yùn)算
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++) {
// 對(duì) address + i 進(jìn)行 md5 運(yùn)算，得到一個(gè)長(zhǎng)度為16的字節(jié)數(shù)組
byte[] digest = md5(address + i);
// 對(duì) digest 部分字節(jié)進(jìn)行4次 hash 運(yùn)算，得到四個(gè)不同的 long 型正整數(shù)
for (int h = 0; h < 4; h++) {
// h = 0 時(shí)，取 digest 中下標(biāo)為 0 ~ 3 的4個(gè)字節(jié)進(jìn)行位運(yùn)算
// h = 1 時(shí)，取 digest 中下標(biāo)為 4 ~ 7 的4個(gè)字節(jié)進(jìn)行位運(yùn)算
// h = 2, h = 3 時(shí)過(guò)程同上
long m = hash(digest, h);
// 將 hash 到 invoker 的映射關(guān)系存儲(chǔ)到 virtualInvokers 中，
// virtualInvokers 需要提供高效的查詢操作，因此選用 TreeMap 作為存儲(chǔ)結(jié)構(gòu)
virtualInvokers.put(m, invoker);
}
}
}
}
}
ConsistentHashSelector 的構(gòu)造方法執(zhí)行了一系列的初始化邏輯，比如從配置中獲取虛擬節(jié)點(diǎn)數(shù)以及參與 hash 計(jì)算的參數(shù)下標(biāo)，默認(rèn)情況下只使用第一個(gè)參數(shù)進(jìn)行 hash。需要特別說(shuō)明的是，ConsistentHashLoadBalance 的負(fù)載均衡邏輯只受參數(shù)值影響，具有相同參數(shù)值的請(qǐng)求將會(huì)被分配給同一個(gè)服務(wù)提供者。ConsistentHashLoadBalance 不 關(guān)系權(quán)重，因此使用時(shí)需要注意一下。
在獲取虛擬節(jié)點(diǎn)數(shù)和參數(shù)下標(biāo)配置后，接下來(lái)要做的事情是計(jì)算虛擬節(jié)點(diǎn) hash 值，并將虛擬節(jié)點(diǎn)存儲(chǔ)到 TreeMap 中。到此，ConsistentHashSelector 初始化工作就完成了。接下來(lái)，我們來(lái)看看 select 方法的邏輯。
public Invoker<T> select(Invocation invocation) {
// 將參數(shù)轉(zhuǎn)為 key
String key = toKey(invocation.getArguments());
// 對(duì)參數(shù) key 進(jìn)行 md5 運(yùn)算
byte[] digest = md5(key);
// 取 digest 數(shù)組的前四個(gè)字節(jié)進(jìn)行 hash 運(yùn)算，再將 hash 值傳給 selectForKey 方法，
// 尋找合適的 Invoker
return selectForKey(hash(digest, 0));
}
private Invoker<T> selectForKey(long hash) {
// 到 TreeMap 中查找第一個(gè)節(jié)點(diǎn)值大于或等于當(dāng)前 hash 的 Invoker
Map.Entry<Long, Invoker<T>> entry = virtualInvokers.tailMap(hash, true).firstEntry();
// 如果 hash 大于 Invoker 在圓環(huán)上最大的位置，此時(shí) entry = null，
// 需要將 TreeMap 的頭節(jié)點(diǎn)賦值給 entry
if (entry == null) {
entry = virtualInvokers.firstEntry();
}
// 返回 Invoker
return entry.getValue();
}
如上，選擇的過(guò)程相對(duì)比較簡(jiǎn)單了。首先是對(duì)參數(shù)進(jìn)行 md5 以及 hash 運(yùn)算，得到一個(gè) hash 值。然后再拿這個(gè)值到 TreeMap 中查找目標(biāo) Invoker 即可。
到此關(guān)于 ConsistentHashLoadBalance 就分析完了。在閱讀 ConsistentHashLoadBalance 源碼之前，大家一定要先補(bǔ)充背景知識(shí)，不然很難看懂代碼邏輯。
2.4 RondRobinLoadBalance
本節(jié)，我們來(lái)看一下 Dubbo 中加權(quán)輪詢負(fù)載均衡的實(shí)現(xiàn) RoundRobinLoadBalance。在詳細(xì)分析源碼前，我們先來(lái)了解一下什么是加權(quán)輪詢。這里從最簡(jiǎn)單的輪詢開(kāi)始講起，所謂輪詢是指將請(qǐng)求輪流分配給每臺(tái)服務(wù)器。舉個(gè)例子，我們有三臺(tái)服務(wù)器 A、B、C。我們將第一個(gè)請(qǐng)求分配給服務(wù)器 A，第二個(gè)請(qǐng)求分配給服務(wù)器 B，第三個(gè)請(qǐng)求分配給服務(wù)器 C，第四個(gè)請(qǐng)求再次分配給服務(wù)器 A。這個(gè)過(guò)程就叫做輪詢。輪詢是一種無(wú)狀態(tài)負(fù)載均衡算法，實(shí)現(xiàn)簡(jiǎn)單，適用于每臺(tái)服務(wù)器性能相近的場(chǎng)景下。但現(xiàn)實(shí)情況下，我們并不能保證每臺(tái)服務(wù)器性能均相近。如果我們將等量的請(qǐng)求分配給性能較差的服務(wù)器，這顯然是不合理的。因此，這個(gè)時(shí)候我們需要對(duì)輪詢過(guò)程進(jìn)行加權(quán)，以調(diào)控每臺(tái)服務(wù)器的負(fù)載。經(jīng)過(guò)加權(quán)后，每臺(tái)服務(wù)器能夠得到的請(qǐng)求數(shù)比例，接近或等于他們的權(quán)重比。比如服務(wù)器 A、B、C 權(quán)重比為 5:2:1。那么在8次請(qǐng)求中，服務(wù)器 A 將收到其中的5次請(qǐng)求，服務(wù)器 B 會(huì)收到其中的2次請(qǐng)求，服務(wù)器 C 則收到其中的1次請(qǐng)求。
以上就是加權(quán)輪詢的算法思想，搞懂了這個(gè)思想，接下來(lái)我們就可以分析源碼了。我們先來(lái)看一下 2.6.4 版本的 RoundRobinLoadBalance。
public class RoundRobinLoadBalance extends AbstractLoadBalance {
public static final String NAME = "roundrobin";
private final ConcurrentMap<String, AtomicPositiveInteger> sequences =
new ConcurrentHashMap<String, AtomicPositiveInteger>();
@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
// key = 全限定類名 + "." + 方法名，比如 com.xxx.DemoService.sayHello
String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
int length = invokers.size();
// 最大權(quán)重
int maxWeight = 0;
// 最小權(quán)重
int minWeight = Integer.MAX_VALUE;
final LinkedHashMap<Invoker<T>, IntegerWrapper> invokerToWeightMap = new LinkedHashMap<Invoker<T>, IntegerWrapper>();
// 權(quán)重總和
int weightSum = 0;
// 下面這個(gè)循環(huán)主要用于查找最大和最小權(quán)重，計(jì)算權(quán)重總和等
for (int i = 0; i < length; i++) {
int weight = getWeight(invokers.get(i), invocation);
// 獲取最大和最小權(quán)重
maxWeight = Math.max(maxWeight, weight);
minWeight = Math.min(minWeight, weight);
if (weight > 0) {
// 將 weight 封裝到 IntegerWrapper 中
invokerToWeightMap.put(invokers.get(i), new IntegerWrapper(weight));
// 累加權(quán)重
weightSum += weight;
}
}
// 查找 key 對(duì)應(yīng)的對(duì)應(yīng) AtomicPositiveInteger 實(shí)例，為空則創(chuàng)建。
// 這里可以把 AtomicPositiveInteger 看成一個(gè)黑盒，大家只要知道
// AtomicPositiveInteger 用于記錄服務(wù)的調(diào)用編號(hào)即可。至于細(xì)節(jié)，
// 大家如果感興趣，可以自行分析
AtomicPositiveInteger sequence = sequences.get(key);
if (sequence == null) {
sequences.putIfAbsent(key, new AtomicPositiveInteger());
sequence = sequences.get(key);
}
// 獲取當(dāng)前的調(diào)用編號(hào)
int currentSequence = sequence.getAndIncrement();
// 如果最小權(quán)重小于最大權(quán)重，表明服務(wù)提供者之間的權(quán)重是不相等的
if (maxWeight > 0 && minWeight < maxWeight) {
// 使用調(diào)用編號(hào)對(duì)權(quán)重總和進(jìn)行取余操作
int mod = currentSequence % weightSum;
// 進(jìn)行 maxWeight 次遍歷
for (int i = 0; i < maxWeight; i++) {
// 遍歷 invokerToWeightMap
for (Map.Entry<Invoker<T>, IntegerWrapper> each : invokerToWeightMap.entrySet()) {
// 獲取 Invoker
final Invoker<T> k = each.getKey();
// 獲取權(quán)重包裝類 IntegerWrapper
final IntegerWrapper v = each.getValue();
// 如果 mod = 0，且權(quán)重大于0，此時(shí)返回相應(yīng)的 Invoker
if (mod == 0 && v.getValue() > 0) {
return k;
}
// mod != 0，且權(quán)重大于0，此時(shí)對(duì)權(quán)重和 mod 分別進(jìn)行自減操作
if (v.getValue() > 0) {
v.decrement();
mod--;
}
}
}
}
// 服務(wù)提供者之間的權(quán)重相等，此時(shí)通過(guò)輪詢選擇 Invoker
return invokers.get(currentSequence % length);
}
// IntegerWrapper 是一個(gè) int 包裝類，主要包含了一個(gè)自減方法。
private static final class IntegerWrapper {
private int value;
public void decrement() {
this.value--;
}
// 省略部分代碼
}
}
如上，RoundRobinLoadBalance 的每行代碼都不是很難理解，但是將它們組合在一起之后，就不是很好理解了。所以下面我們舉例進(jìn)行說(shuō)明，假設(shè)我們有三臺(tái)服務(wù)器 servers = [A, B, C]，對(duì)應(yīng)的權(quán)重為 weights = [2, 5, 1]。接下來(lái)對(duì)上面的邏輯進(jìn)行簡(jiǎn)單的模擬。
mod = 0：滿足條件，此時(shí)直接返回服務(wù)器 A
mod = 1：需要進(jìn)行一次遞減操作才能滿足條件，此時(shí)返回服務(wù)器 B
mod = 2：需要進(jìn)行兩次遞減操作才能滿足條件，此時(shí)返回服務(wù)器 C
mod = 3：需要進(jìn)行三次遞減操作才能滿足條件，經(jīng)過(guò)遞減后，服務(wù)器權(quán)重為 [1, 4, 0]，此時(shí)返回服務(wù)器 A
mod = 4：需要進(jìn)行四次遞減操作才能滿足條件，經(jīng)過(guò)遞減后，服務(wù)器權(quán)重為 [0, 4, 0]，此時(shí)返回服務(wù)器 B
mod = 5：需要進(jìn)行五次遞減操作才能滿足條件，經(jīng)過(guò)遞減后，服務(wù)器權(quán)重為 [0, 3, 0]，此時(shí)返回服務(wù)器 B
mod = 6：需要進(jìn)行六次遞減操作才能滿足條件，經(jīng)過(guò)遞減后，服務(wù)器權(quán)重為 [0, 2, 0]，此時(shí)返回服務(wù)器 B
mod = 7：需要進(jìn)行七次遞減操作才能滿足條件，經(jīng)過(guò)遞減后，服務(wù)器權(quán)重為 [0, 1, 0]，此時(shí)返回服務(wù)器 B
經(jīng)過(guò)8次調(diào)用后，我們得到的負(fù)載均衡結(jié)果為 [A, B, C, A, B, B, B, B]，次數(shù)比 A:B:C = 2:5:1，等于權(quán)重比。當(dāng) sequence = 8 時(shí)，mod = 0，此時(shí)重頭再來(lái)。從上面的模擬過(guò)程可以看出，當(dāng) mod >= 3 后，服務(wù)器 C 就不會(huì)被選中了，因?yàn)樗臋?quán)重被減為0了。當(dāng) mod >= 4 后，服務(wù)器 A 的權(quán)重被減為0，此后 A 就不會(huì)再被選中。
以上是 2.6.4 版本的 RoundRobinLoadBalance 分析過(guò)程，2.6.4 版本的 RoundRobinLoadBalance 在某些情況下存在著比較嚴(yán)重的性能問(wèn)題，該問(wèn)題最初是在 issue #2578 中被反饋出來(lái)。問(wèn)題出在了 Invoker 的返回時(shí)機(jī)上，RoundRobinLoadBalance 需要在mod == 0 && v.getValue() > 0 條件成立的情況下才會(huì)被返回相應(yīng)的 Invoker。假如 mod 很大，比如 10000，50000，甚至更大時(shí)，doSelect 方法需要進(jìn)行很多次計(jì)算才能將 mod 減為0。由此可知，doSelect 的效率與 mod 有關(guān)，時(shí)間復(fù)雜度為 O(mod)。mod 又受最大權(quán)重 maxWeight 的影響，因此當(dāng)某個(gè)服務(wù)提供者配置了非常大的權(quán)重，此時(shí) RoundRobinLoadBalance 會(huì)產(chǎn)生比較嚴(yán)重的性能問(wèn)題。這個(gè)問(wèn)題被反饋后，社區(qū)很快做了回應(yīng)。并對(duì) RoundRobinLoadBalance 的代碼進(jìn)行了重構(gòu)，將時(shí)間復(fù)雜度優(yōu)化至了常量級(jí)別。這個(gè)優(yōu)化可以說(shuō)很好了，下面我們來(lái)學(xué)習(xí)一下優(yōu)化后的代碼。
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>();
@Override
protected <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<>();
// 查找最大和最小權(quán)重
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));
}
}
// 獲取當(dāng)前服務(wù)對(duì)應(yīng)的調(diào)用序列對(duì)象 AtomicPositiveInteger
AtomicPositiveInteger sequence = sequences.get(key);
if (sequence == null) {
// 創(chuàng)建 AtomicPositiveInteger，默認(rèn)值為0
sequences.putIfAbsent(key, new AtomicPositiveInteger());
sequence = sequences.get(key);
}
// 獲取下標(biāo)序列對(duì)象 AtomicPositiveInteger
AtomicPositiveInteger indexSeq = indexSeqs.get(key);
if (indexSeq == null) {
// 創(chuàng)建 AtomicPositiveInteger，默認(rèn)值為 -1
indexSeqs.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;
// 每循環(huán)一輪（index = 0），重新計(jì)算 currentWeight
if (index == 0) {
currentWeight = sequence.incrementAndGet() % maxWeight;
}
// 檢測(cè) Invoker 的權(quán)重是否大于 currentWeight，大于則返回
if (getWeight(invokerToWeightList.get(index), invocation) > currentWeight) {
return invokerToWeightList.get(index);
}
}
}
// 所有 Invoker 權(quán)重相等，此時(shí)進(jìn)行普通的輪詢即可
return invokers.get(sequence.incrementAndGet() % length);
}
}
上面代碼的邏輯是這樣的，每進(jìn)行一輪循環(huán)，重新計(jì)算 currentWeight。如果當(dāng)前 Invoker 權(quán)重大于 currentWeight，則返回該 Invoker。下面舉例說(shuō)明，假設(shè)服務(wù)器 [A, B, C] 對(duì)應(yīng)權(quán)重 [5, 2, 1]。
第一輪循環(huán)，currentWeight = 1，可返回 A 和 B
第二輪循環(huán)，currentWeight = 2，返回 A
第三輪循環(huán)，currentWeight = 3，返回 A
第四輪循環(huán)，currentWeight = 4，返回 A
第五輪循環(huán)，currentWeight = 0，返回 A, B, C
如上，這里的一輪循環(huán)是指 index 再次變?yōu)?所經(jīng)歷過(guò)的循環(huán)，這里可以把 index = 0 看做是一輪循環(huán)的開(kāi)始。每一輪循環(huán)的次數(shù)與 Invoker 的數(shù)量有關(guān)，Invoker 數(shù)量通常不會(huì)太多，所以我們可以認(rèn)為上面代碼的時(shí)間復(fù)雜度為常數(shù)級(jí)。
重構(gòu)后的 RoundRobinLoadBalance 看起來(lái)已經(jīng)很不錯(cuò)了，但是在代碼更新不久后，很快又被重構(gòu)了。這次重構(gòu)原因是新的 RoundRobinLoadBalance 在某些情況下選出的服務(wù)器序列不夠均勻。比如，服務(wù)器 [A, B, C] 對(duì)應(yīng)權(quán)重 [5, 1, 1]。進(jìn)行7次負(fù)載均衡后，選擇出來(lái)的序列為 [A, A, A, A, A, B, C]。前5個(gè)請(qǐng)求全部都落在了服務(wù)器 A上，這將會(huì)使服務(wù)器 A 短時(shí)間內(nèi)接收大量的請(qǐng)求，壓力陡增。而 B 和 C 此時(shí)無(wú)請(qǐng)求，處于空閑狀態(tài)。而我們期望的結(jié)果是這樣的 [A, A, B, A, C, A, A]，不同服務(wù)器可以穿插獲取請(qǐng)求。為了增加負(fù)載均衡結(jié)果的平滑性，社區(qū)再次對(duì) RoundRobinLoadBalance 的實(shí)現(xiàn)進(jìn)行了重構(gòu)，這次重構(gòu)參考自 Nginx 的平滑加權(quán)輪詢負(fù)載均衡。每個(gè)服務(wù)器對(duì)應(yīng)兩個(gè)權(quán)重，分別為 weight 和 currentWeight。其中 weight 是固定的，currentWeight 會(huì)動(dòng)態(tài)調(diào)整，初始值為0。當(dāng)有新的請(qǐng)求進(jìn)來(lái)時(shí)，遍歷服務(wù)器列表，讓它的 currentWeight 加上自身權(quán)重。遍歷完成后，找到最大的 currentWeight，并將其減去權(quán)重總和，然后返回相應(yīng)的服務(wù)器即可。
上面描述不是很好理解，下面還是舉例進(jìn)行說(shuō)明。這里仍然使用服務(wù)器 [A, B, C] 對(duì)應(yīng)權(quán)重 [5, 1, 1] 的例子說(shuō)明，現(xiàn)在有7個(gè)請(qǐng)求依次進(jìn)入負(fù)載均衡邏輯，選擇過(guò)程如下：
| 請(qǐng)求編號(hào) | currentWeight 數(shù)組 | 選擇結(jié)果 | 減去權(quán)重總和后的 currentWeight 數(shù)組 |
|---|---|---|---|
| 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] |
如上，經(jīng)過(guò)平滑性處理后，得到的服務(wù)器序列為 [A, A, B, A, C, A, A]，相比之前的序列 [A, A, A, A, A, B, C]，分布性要好一些。初始情況下 currentWeight = [0, 0, 0]，第7個(gè)請(qǐng)求處理完后，currentWeight 再次變?yōu)?[0, 0, 0]。
以上就是平滑加權(quán)輪詢的計(jì)算過(guò)程，接下來(lái)，我們來(lái)看看 Dubbo-2.6.5 是如何實(shí)現(xiàn)上面的計(jì)算過(guò)程的。
public class RoundRobinLoadBalance extends AbstractLoadBalance {
public static final String NAME = "roundrobin";
private static int RECYCLE_PERIOD = 60000;
protected static class WeightedRoundRobin {
// 服務(wù)提供者權(quán)重
private int weight;
// 當(dāng)前權(quán)重
private AtomicLong current = new AtomicLong(0);
// 最后一次更新時(shí)間
private long lastUpdate;
public void setWeight(int weight) {
this.weight = weight;
// 初始情況下，current = 0
current.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 結(jié)構(gòu)，存儲(chǔ)的數(shù)據(jù)結(jié)構(gòu)示例如下：
// {
// "UserService.query": {
// "url1": WeightedRoundRobin@123,
// "url2": WeightedRoundRobin@456,
// },
// "UserService.update": {
// "url1": WeightedRoundRobin@123,
// "url2": WeightedRoundRobin@456,
// }
// }
// 最外層為服務(wù)類名 + 方法名，第二層為 url 到 WeightedRoundRobin 的映射關(guān)系。
// 這里我們可以將 url 看成是服務(wù)提供者的 id
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 映射表，如果為空，則創(chuàng)建一個(gè)新的
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;
// 獲取當(dāng)前時(shí)間
long now = System.currentTimeMillis();
Invoker<T> selectedInvoker = null;
WeightedRoundRobin selectedWRR = null;
// 下面這個(gè)循環(huán)主要做了這樣幾件事情：
// 1. 遍歷 Invoker 列表，檢測(cè)當(dāng)前 Invoker 是否有
// 相應(yīng)的 WeightedRoundRobin，沒(méi)有則創(chuàng)建
// 2. 檢測(cè) Invoker 權(quán)重是否發(fā)生了變化，若變化了，
// 則更新 WeightedRoundRobin 的 weight 字段
// 3. 讓 current 字段加上自身權(quán)重，等價(jià)于 current += weight
// 4. 設(shè)置 lastUpdate 字段，即 lastUpdate = now
// 5. 尋找具有最大 current 的 Invoker，以及 Invoker 對(duì)應(yīng)的 WeightedRoundRobin，
// 暫存起來(lái)，留作后用
// 6. 計(jì)算權(quán)重總和
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;
}
// 檢測(cè)當(dāng)前 Invoker 是否有對(duì)應(yīng)的 WeightedRoundRobin，沒(méi)有則創(chuàng)建
if (weightedRoundRobin == null) {
weightedRoundRobin = new WeightedRoundRobin();
// 設(shè)置 Invoker 權(quán)重
weightedRoundRobin.setWeight(weight);
// 存儲(chǔ) url 唯一標(biāo)識(shí) identifyString 到 weightedRoundRobin 的映射關(guān)系
map.putIfAbsent(identifyString, weightedRoundRobin);
weightedRoundRobin = map.get(identifyString);
}
// Invoker 權(quán)重不等于 WeightedRoundRobin 中保存的權(quán)重，說(shuō)明權(quán)重變化了，此時(shí)進(jìn)行更新
if (weight != weightedRoundRobin.getWeight()) {
weightedRoundRobin.setWeight(weight);
}
// 讓 current 加上自身權(quán)重，等價(jià)于 current += weight
long cur = weightedRoundRobin.increaseCurrent();
// 設(shè)置 lastUpdate，表示近期更新過(guò)
weightedRoundRobin.setLastUpdate(now);
// 找出最大的 current
if (cur > maxCurrent) {
maxCurrent = cur;
// 將具有最大 current 權(quán)重的 Invoker 賦值給 selectedInvoker
selectedInvoker = invoker;
// 將 Invoker 對(duì)應(yīng)的 weightedRoundRobin 賦值給 selectedWRR，留作后用
selectedWRR = weightedRoundRobin;
}
// 計(jì)算權(quán)重總和
totalWeight += weight;
}
// 對(duì) <identifyString, WeightedRoundRobin> 進(jìn)行檢查，過(guò)濾掉長(zhǎng)時(shí)間未被更新的節(jié)點(diǎn)。
// 該節(jié)點(diǎn)可能掛了，invokers 中不包含該節(jié)點(diǎn)，所以該節(jié)點(diǎn)的 lastUpdate 長(zhǎng)時(shí)間無(wú)法被更新。
// 若未更新時(shí)長(zhǎng)超過(guò)閾值后，就會(huì)被移除掉，默認(rèn)閾值為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);
// 遍歷修改，即移除過(guò)期記錄
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 減去權(quán)重總和，等價(jià)于 current -= totalWeight
selectedWRR.sel(totalWeight);
// 返回具有最大 current 的 Invoker
return selectedInvoker;
}
// should not happen here
return invokers.get(0);
}
}
以上就是 Dubbo-2.6.5 版本的 RoundRobinLoadBalance，大家如果能夠理解平滑加權(quán)輪詢算法的計(jì)算過(guò)程，再配合代碼中注釋，理解上面的代碼應(yīng)該不難。
3.總結(jié)
本篇文章對(duì) Dubbo 中的幾種負(fù)載均衡實(shí)現(xiàn)進(jìn)行了詳細(xì)的分析，內(nèi)容比較多，大家慢慢消化。理解負(fù)載均衡代碼邏輯的關(guān)鍵之處在于對(duì)背景知識(shí)的理解，因此大家在閱讀源碼前，務(wù)必先了解每種負(fù)載均衡對(duì)應(yīng)的背景知識(shí)。
