Dubbo源碼解析之LoadBalance負載均衡

閱讀須知

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負載均衡

閱讀須知

正文

繼續閱讀

centos上利用nginx配置tomcat負載均衡

NGINX_TOMCAT 負載均衡配置

我對SOA的認識以及心得

企業級負載均衡解決方案之六：UCloud四層負載均衡解決方案Vortex

手把手實作RPC架構--簡易版Dubbo構造（三）用戶端動态代理

配置ipvsadm服務

裸眼3D技術原理及單晶片解決方案

PHP工程師面臨成長瓶頸

cisco交換機指令彙總

Oracle實作高可用性的工具（負載均衡/故障切換）

2.6核心netfilter包截獲

阿裡巴巴分布式服務架構 Dubbo 團隊成員梁飛專訪

阿裡巴巴分布式服務架構Dubbo介紹引言SOA化服務架構DubboDubbo的主要特點

Dubbo：來自于阿裡巴巴的分布式服務架構 Dubbo：來自于阿裡巴巴的分布式服務架構

多線程擷取随機數性能對比（C vs C++）

nginx location中斜線的位置的重要性