Dubbo原始碼解析之LoadBalance負載均衡

阿新 • • 發佈：2018-11-27

閱讀須知

dubbo版本：2.6.0
spring版本：4.3.8
文章中使用/* */註釋的方法會做深入分析

正文

dubbo一共支援四種負載均衡策略，RoundRobinLoadBalance（輪詢）、RandomLoadBalance（隨機）、LeastActiveLoadBalance（最少活躍）、ConsistentHashLoadBalance（一致性雜湊）。預設為隨機策略，我門在分析consumer呼叫過程中invoker的選擇時，看到了負載均衡策略的應用，下面我們分別來分析一下這四種負載均衡策略的實現細節：
AbstractLoadBalance：

public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) {
    if (invokers == null || invokers.size() == 0)
        return null;
    if (invokers.size() == 1)
        // 如果invoker集合中只有一個，直接返回第一個
        return invokers.get(0);
    /* 子類實現負載均衡選擇invoker */ 

    return doSelect(invokers, url, invocation);
}

首先來看預設的隨機策略：
RandomLoadBalance：

protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
    int length = invokers.size();
    int totalWeight = 0;
    // 標識每一個invoker是否有著相同的權重
    boolean 
 sameWeight = true;
    for (int i = 0; i < length; i++) {
        // 遍歷獲取每一個invoker的權重，權重可以進行配置
        int weight = getWeight(invokers.get(i), invocation);
        totalWeight += weight;
        // 對比當前invoker的權重和上一個invoker的權重
        // 如果發現有權重不相等的，說明不是每一個invoker的權重都一樣
        if (sameWeight && i > 0
                && weight != getWeight(invokers.get(i - 1), invocation)) {
            sameWeight = false;
        }
    }
    if (totalWeight > 0 && !sameWeight) {
        // 基於總權重獲取隨機數
        int offset = random.nextInt(totalWeight);
        for (int i = 0; i < length; i++) {
            // 遍歷invoker，用獲取的隨機數減去當前invoker的權重，如果小於0，則選擇當前invoker
            // 如果當前invoker的權重較大，這樣差值就更容易小於0，這樣選中的機率就越大
            offset -= getWeight(invokers.get(i), invocation);
            if (offset < 0) {
                return invokers.get(i);
            }
        }
    }
    // 如果所有的invoker具有相同的權重值或總權重等於0，則隨機返回一個invoker
    return invokers.get(random.nextInt(length));
}

下一個我們來看輪詢策略：
RoundRobinLoadBalance：

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 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) {
            // 新增invoker和權重的對映
            invokerToWeightMap.put(invokers.get(i), new IntegerWrapper(weight));
            weightSum += weight;
        }
    }
    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;
        for (int i = 0; i < maxWeight; i++) {
            for (Map.Entry<Invoker<T>, IntegerWrapper> each : invokerToWeightMap.entrySet()) {
                final Invoker<T> k = each.getKey();
                final IntegerWrapper v = each.getValue();
                // 當取模值為0或者遞減為0後，如果當前invoker的權重值還大於0，則選擇當前invoker
                // 這樣invoker的權重值越大，則遞減之後大於0的概率越大，選中的機率就越大
                if (mod == 0 && v.getValue() > 0) {
                    return k;
                }
                if (v.getValue() > 0) {
                    v.decrement(); // 權重值遞減
                    mod--; // 取模值遞減
                }
            }
        }
    }
    // 輪詢選擇
    return invokers.get(currentSequence % length);
}

接下來我們來看最少活躍策略：
LeastActiveLoadBalance：

protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
    int length = invokers.size();
    int leastActive = -1; // 所有invoker的最小活躍值
    int leastCount = 0; // 具有相同最小活躍值的invoker數量（leastActive）
    int[] leastIndexs = new int[length]; // 具有相同最小活躍值的invoker索引（leastActive）
    int totalWeight = 0; // 權重和，總權重
    int firstWeight = 0; // 初始化權重，用來做比較
    boolean sameWeight = true; // 每一個invoker是否有相同的權重值
    for (int i = 0; i < length; i++) {
        Invoker<T> invoker = invokers.get(i);
        // 活躍數量
        int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive();
        // 權重
        int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT);
        // 當發現具有較小活躍值的invoker時重新開始
        if (leastActive == -1 || active < leastActive) {
            leastActive = active; // 記錄當前最小活躍值
            // 重置具有相同最小活躍值的invoker數量，根據當前最小活躍值的invoker數量再次計數
            leastCount = 1;
            leastIndexs[0] = i; // 重置
            totalWeight = weight; // 重置
            firstWeight = weight; // 記錄第一個invoker的權重
            sameWeight = true; // 重置
        // 如果當前調invoker的活動值等於leaseActive，則累積
        } else if (active == leastActive) {
            leastIndexs[leastCount++] = i; // 記錄這個invoker的index
            totalWeight += weight; // 累加這個invoker的權重到總權重
            // 與初始化權重進行比較，判斷是否每一個invoker都具有相同的權重值
            if (sameWeight && i > 0
                    && weight != firstWeight) {
                sameWeight = false;
            }
        }
    }
    if (leastCount == 1) {
        // 如果我們只有一個具有最小活躍值的invoker，則直接返回此invoker
        return invokers.get(leastIndexs[0]);
    }
    if (!sameWeight && totalWeight > 0) {
        // 如果（並非每個invoker具有相同的權重並且至少一個invoker的權重 > 0），則根據totalWeight隨機選擇
        int offsetWeight = random.nextInt(totalWeight);
        // 根據隨機值獲取一個invoker
        for (int i = 0; i < leastCount; i++) {
            int leastIndex = leastIndexs[i];
            offsetWeight -= getWeight(invokers.get(leastIndex), invocation);
            if (offsetWeight <= 0)
                return invokers.get(leastIndex);
        }
    }
    // 如果所有呼叫者具有相同的權重值或totalWeight = 0，則從具有最小活躍值的集合中隨機返回一個
    return invokers.get(leastIndexs[random.nextInt(leastCount)]);
}

最後我們來看一致性雜湊策略，一致性雜湊，相同引數的請求總是發到同一提供者。當某一臺提供者掛時，原本發往該提供者的請求，基於虛擬節點，平攤到其它提供者。預設只對第一個引數Hash，如果要修改，可以配置<dubbo:parameter key="hash.arguments" value="0,1" />。預設使用160個虛擬節點，如果要修改，可以配置<dubbo:parameter key="hash.nodes" value="320" />。
ConsistentHashLoadBalance：

protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
    String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
    int identityHashCode = System.identityHashCode(invokers);
    // 嘗試從快取中獲取一致性雜湊選擇器
    ConsistentHashSelector<T> selector = (ConsistentHashSelector<T>) selectors.get(key);
    // 這裡的identityHashCode主要作用是確定invoker集合是否發生變化
    if (selector == null || selector.identityHashCode != identityHashCode) {
        /* 構建新的一致性雜湊選擇器並快取 */
        selectors.put(key, new ConsistentHashSelector<T>(invokers, invocation.getMethodName(), identityHashCode));
        selector = (ConsistentHashSelector<T>) selectors.get(key);
    }
    /* 選擇器選擇invoker */
    return selector.select(invocation);
}

ConsistentHashLoadBalance.ConsistentHashSelector：

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();
    // 預設160個虛擬節點
    this.replicaNumber = url.getMethodParameter(methodName, "hash.nodes", 160);
    // 預設進行雜湊匹配的引數index為0，也就是第一個
    String[] index = Constants.COMMA_SPLIT_PATTERN.split(url.getMethodParameter(methodName, "hash.arguments", "0"));
    // 記錄進行雜湊匹配的引數的index陣列
    argumentIndex = new int[index.length];
    for (int i = 0; i < index.length; i++) {
        argumentIndex[i] = Integer.parseInt(index[i]);
    }
    // 這裡的作用個人理解為儘量將每個invoker的虛擬節點均勻的打散在雜湊環上
    for (Invoker<T> invoker : invokers) {
        String address = invoker.getUrl().getAddress();
        for (int i = 0; i < replicaNumber / 4; i++) {
            byte[] digest = md5(address + i);
            for (int h = 0; h < 4; h++) {
                long m = hash(digest, h);
                // 放置虛擬節點
                virtualInvokers.put(m, invoker);
            }
        }
    }
}

ConsistentHashLoadBalance.ConsistentHashSelector：

public Invoker<T> select(Invocation invocation) {
    // 根據指定進行雜湊匹配的引數index取出引數
    String key = toKey(invocation.getArguments());
    byte[] digest = md5(key);
    /* 根據雜湊匹配invoker */
    return selectForKey(hash(digest, 0));
}

ConsistentHashLoadBalance.ConsistentHashSelector：

private Invoker<T> selectForKey(long hash) {
    Invoker<T> invoker;
    Long key = hash;
    // 判斷虛擬節點對映中是否有匹配的雜湊key
    if (!virtualInvokers.containsKey(key)) {
        // 取出對映中key大於或等於給定引數雜湊值的部分檢視
        SortedMap<Long, Invoker<T>> tailMap = virtualInvokers.tailMap(key);
        if (tailMap.isEmpty()) {
            // 如果取出的部分檢視是空的，則直接返回第一個key
            key = virtualInvokers.firstKey();
        } else {
            // 不為空則直接獲取部分檢視的第一個key
            key = tailMap.firstKey();
        }
    }
    // 根據key取出invoker返回
    invoker = virtualInvokers.get(key);
    return invoker;
}

到這裡，整個Dubbo的負載均衡原始碼分析就完成了。

Dubbo原始碼解析之LoadBalance負載均衡

閱讀須知

正文

Dubbo原始碼解析之LoadBalance負載均衡

dubbo原始碼解析之負載均衡

Dubbo原始碼解析之服務端接收訊息

Dubbo原始碼解析之服務釋出與註冊

Dubbo原始碼解析之consumer關聯provider

Dubbo原始碼解析之consumer呼叫篇

Dubbo原始碼解析之provider呼叫篇

Dubbo原始碼解析之consumer初始化

Dubbo原始碼解析之客戶端初始化及服務呼叫

dubbo原始碼解析之服務export 遠端服務②

Dubbo原始碼解析之SPI（一）：擴充套件類的載入過程

Dubbo 原始碼解析四 —— 負載均衡LoadBalance

深入Dubbo 原始碼解析 — 負載均衡LoadBalance

dubbo原始碼閱讀之負載均衡

Dubbo原理和原始碼解析之服務暴露

Dubbo原理和原始碼解析之標籤解析

Dubbo原理和原始碼解析之“微核心+外掛”機制

Dubbo原理和原始碼解析之服務引用

Nginx學習筆記06負載均衡之(一)負載均衡介紹

Android框架原始碼解析之（四）Picasso

Dubbo原始碼解析之LoadBalance負載均衡

閱讀須知

正文

相關推薦