Dubbo 源碼解析四 —— 負載均衡LoadBalance

時間 2019-11-24

標籤 dubbo 源碼解析負載均衡 loadbalance 欄目 Dubbo 简体版

原文原文鏈接

歡迎來個人 Star Followers 後期後繼續更新Dubbo別的文章html

Dubbo 源碼分析系列之一環境搭建

Dubbo 入門之二 ——- 項目結構解析

Dubbo 源碼分析系列之三 —— 架構原理

技術點

面試中Dubbo負載均衡常問的幾點
常見負載均衡算法簡介
Dubbo 官方文檔介紹
Dubbo 負載均衡的策略
Dubbo 負載均衡源碼解析

面試中Dubbo負載均衡常問的幾點

談談dubbo中的負載均衡算法及特色
最小活躍數算法中是如何統計這個活躍數的
簡單談談你對一致性哈希算法的認識
Dubbo默認的負載均衡策略是什麼，爲何使用 RandomLoadBalance 隨機負載均衡算法
談談幾種負載均衡的優缺點
若是讓你設計負載均衡你將如何設計
源碼負載均衡你學到了什麼
有沒有將Dubbo的負載均衡的原理使用在實際的項目中

常見負載均衡算法簡介

首先引出一點負載均衡的目的是什麼？java

當一臺服務器的承受能力達到上限時，那麼就須要多臺服務器來組成集羣，提高應用總體的吞吐量，那麼這個時候就涉及到如何合理分配客戶端請求到集羣中不一樣的機器，這個過程就叫作負載均衡node

下面簡單介紹幾種負載均衡算法，有利於理解源碼中爲何這樣設計git

權重隨機算法

策略就是根據權重佔比隨機。算法很簡單，就是一根數軸。而後利用僞隨機數產生點，**看點落在了哪一個區域從而選擇對應的服務器github

權重輪詢算法

輪詢算法是指依次訪問可用服務器列表，其和隨機本質是同樣的處理，在無權重因素下，輪詢只是在選數軸上的點時採起自增對長度取餘方式。有權重因素下依然自增取餘，再看選取的點落在了哪一個區域。面試

一致性Hash負載均衡算法

利用Hash算法定位相同的服務器算法

普通的Hash：當客戶端請求到達是則使用 hash(client) % N,其中N是服務器數量，利用這個表達式計算出該客戶端對應的Server處理
一致性Hash：一致性Hash是把服務器分佈變成一個環形，每個hash(clinet)的結果會在該環上順時針尋找第一個與其鄰的Server節點

一致性Hash算法apache

—————————— 下面這部分是來源於dubbo 官方文檔 ------------------------------------api

Dubbo 官方文檔介紹

負載均衡

在集羣負載均衡時，Dubbo 提供了多種均衡策略，缺省爲 Random LoadBalance 隨機調用數組

負載均衡策略

Random LoadBalance

隨機，按權重設置隨機機率。
在一個截面上碰撞的機率高，但調用量越大分佈越均勻，並且按機率使用權重後也比較均勻，有利於動態調整提供者權重。

RoundRobin LoadBalance

輪詢，按公約後的權重設置輪詢比率。
存在慢的提供者累積請求的問題，好比：第二臺機器很慢，但沒掛，當請求調到第二臺時就卡在那，長此以往，全部請求都卡在調到第二臺上。

LeastActive LoadBalance

最少活躍調用數，相同活躍數的隨機，活躍數指調用先後計數差。
使慢的提供者收到更少請求，由於越慢的提供者的調用先後計數差會越大。

ConsistentHash LoadBalance

一致性 Hash，相同參數的請求老是發到同一提供者。
當某一臺提供者掛時，本來發往該提供者的請求，基於虛擬節點，平攤到其它提供者，不會引發劇烈變更。
算法參見：http://en.wikipedia.org/wiki/Consistent_hashing
缺省只對第一個參數 Hash，若是要修改，請配置 <dubbo:parameter key="hash.arguments" value="0,1" />
缺省用 160 份虛擬節點，若是要修改，請配置 <dubbo:parameter key="hash.nodes" value="320" />

配置

服務端服務級別

<dubbo:service interface="..." loadbalance="roundrobin" />

客戶端服務級別

<dubbo:reference interface="..." loadbalance="roundrobin" />

服務端方法級別

<dubbo:service interface="...">
    <dubbo:method name="..." loadbalance="roundrobin"/>
</dubbo:service>

客戶端方法級別

<dubbo:reference interface="...">
    <dubbo:method name="..." loadbalance="roundrobin"/>
</dubbo:reference>

———————————————— Dubbo 官方文檔已結束 ------------------------------------------

Dubbo 負載均衡的策略

上面官網文檔已經說明 Dubbo 的負載均衡算法總共有4種

隨機算法 RandomLoadBalance（默認）

輪訓算法 RoundRobinLoadBalance

最小活躍數算法 LeastActiveLoadBalance

一致性hash算法 ConsistentHashLoadBalance

咱們先看下接口的繼承圖

LoadBalance

首先查看 LoadBalance 接口

Invoker select(List invokers, URL url, Invocation invocation) throws RpcException;

LoadBalance 定義了一個方法就是從 invokers 列表中選取一個

AbstractLoadBalance

AbstractLoadBalance 抽象類是全部負載均衡策略實現類的父類，實現了LoadBalance接口的方法，同時提供抽象方法交由子類實現，

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)
            return invokers.get(0);
        return doSelect(invokers, url, invocation);
    }

    protected abstract <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation);

下面對四種均衡策略依次解析

RandomLoadBalance(隨機)

隨機，按權重設置隨機機率。

在一個截面上碰撞的機率高，但調用量越大分佈越均勻，並且按機率使用權重後也比較均勻，有利於動態調整提供者權重。

RandomLoadBalance#doSelect()

@Override
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
    //先得到invoker 集合大小
    int length = invokers.size(); // Number of invokers
    //總權重
    int totalWeight = 0; // The sum of weights
    //每一個invoker是否有相同的權重
    boolean sameWeight = true; // Every invoker has the same weight?
    // 計算總權重
    for (int i = 0; i < length; i++) {
        //得到單個invoker 的權重
        int weight = getWeight(invokers.get(i), invocation);
        //累加
        totalWeight += weight; // Sum
        if (sameWeight && i > 0 && weight != getWeight(invokers.get(i - 1), invocation)) {
            sameWeight = false;
        }
    }
    // 權重不相等，隨機後，判斷在哪一個 Invoker 的權重區間中
    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));
}

算法分析

假定有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;
假設offset=1（即random.nextInt(9)=1）
1-2=-1<0？是，因此選中 10.0.0.1:20884, weight=2

假設offset=4（即random.nextInt(9)=4）
4-2=2<0？否，這時候offset=2， 2-3<0？是，因此選中 10.0.0.1:20886, weight=3

假設offset=7（即random.nextInt(9)=7）
7-2=5<0？否，這時候offset=5， 5-3=2<0？否，這時候offset=2， 2-4<0？是，因此選中 10.0.0.1:20888, weight=4

流程圖

RoundRobinLoadBalance#doSelect()(輪詢)

輪詢，按公約後的權重設置輪詢比率。

存在慢的提供者累積請求的問題，好比：第二臺機器很慢，但沒掛，當請求調到第二臺時就卡在那，長此以往，全部請求都卡在調到第二臺上。

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(); // Number of invokers
    int maxWeight = 0; // The maximum weight
    int minWeight = Integer.MAX_VALUE; // The minimum weight
    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); // Choose the maximum weight
        minWeight = Math.min(minWeight, weight); // Choose the minimum weight
        if (weight > 0) {
            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);
    }
    // 得到當前順序號，並遞增 + 1
    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 ，當前 Invoker 還有剩餘權重，返回該 Invoker 對象
                if (mod == 0 && v.getValue() > 0) {
                    return k;
                }
                // 若 Invoker 還有權重值，扣除它( value )和剩餘權重( mod )。
                if (v.getValue() > 0) {
                    v.decrement();
                    mod--;
                }
            }
        }
    }
    // 權重相等，平均順序得到
    // Round robin
    return invokers.get(currentSequence % length);
}

算法說明

假定有3臺權重都同樣的dubbo provider:
10.0.0.1:20884, weight=100
10.0.0.1:20886, weight=100
10.0.0.1:20888, weight=100
輪詢算法的實現：
其調用方法某個方法(key)的sequence從0開始：
sequence=0時，選擇invokers.get(0%3)=10.0.0.1:20884
sequence=1時，選擇invokers.get(1%3)=10.0.0.1:20886
sequence=2時，選擇invokers.get(2%3)=10.0.0.1:20888
sequence=3時，選擇invokers.get(3%3)=10.0.0.1:20884
sequence=4時，選擇invokers.get(4%3)=10.0.0.1:20886
sequence=5時，選擇invokers.get(5%3)=10.0.0.1:20888

LeastActiveLoadBalance(最少活躍數)

最少活躍調用數，相同活躍數的隨機，活躍數指調用先後計數差。

使慢的提供者收到更少請求，由於越慢的提供者的調用先後計數差會越大。

protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
    // 總個數
    int length = invokers.size(); // Number of invokers
    // 最少的活躍數
    int leastActive = -1; // The least active value of all invokers
    // 相同最小活躍數的個數
    int leastCount = 0; // The number of invokers having the same least active value (leastActive)
    // 相同最小活躍數的下標
    int[] leastIndexs = new int[length]; // The index of invokers having the same least active value (leastActive)
    //總權重
    int totalWeight = 0; // The sum of weights
    // 第一個權重，用於於計算是否相同
    int firstWeight = 0; // Initial value, used for comparision
    // 是否全部權重相同
    boolean sameWeight = true; // Every invoker has the same weight value?
    // 計算得到相同最小活躍數的數組和個數
    for (int i = 0; i < length; i++) {
        Invoker<T> invoker = invokers.get(i);
        // 活躍數
        int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive(); // Active number
        // 權重
        int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT); // Weight
        // 發現更小的活躍數，從新開始
        if (leastActive == -1 || active < leastActive) { // Restart, when find a invoker having smaller least active value.
            // 記錄最小活躍數
            leastActive = active; // Record the current least active value
            // 從新統計相同最小活躍數的個數
            leastCount = 1; // Reset leastCount, count again based on current leastCount
            // 從新記錄最小活躍數下標
            leastIndexs[0] = i; // Reset
            // 從新統計總權重
            totalWeight = weight; // Reset
            // 記錄第一個權重
            firstWeight = weight; // Record the weight the first invoker
            // 還原權重標識
            sameWeight = true; // Reset, every invoker has the same weight value?
        // 累計相同最小的活躍數
        } else if (active == leastActive) { // If current invoker's active value equals with leaseActive, then accumulating.
            // 累計相同最小活躍數下標
            leastIndexs[leastCount++] = i; // Record index number of this invoker
            // 累計總權重
            totalWeight += weight; // Add this invoker's weight to totalWeight.
            // 判斷全部權重是否同樣
            // If every invoker has the same weight?
            if (sameWeight && i > 0
                    && weight != firstWeight) {
                sameWeight = false;
            }
        }
    }
    // assert(leastCount > 0)
    if (leastCount == 1) {
        // 若是隻有一個最小則直接返回
        // If we got exactly one invoker having the least active value, return this invoker directly.
        return invokers.get(leastIndexs[0]);
    }
    if (!sameWeight && totalWeight > 0) {
        // 若是權重不相同且權重大於0則按總權重數隨機
        // If (not every invoker has the same weight & at least one invoker's weight>0), select randomly based on totalWeight.
        int offsetWeight = ThreadLocalRandom.current().nextInt(totalWeight);
        // 並肯定隨機值落在哪一個片段上
        // Return a invoker based on the random value.
        for (int i = 0; i < leastCount; i++) {
            int leastIndex = leastIndexs[i];
            offsetWeight -= getWeight(invokers.get(leastIndex), invocation);
            if (offsetWeight <= 0)
                return invokers.get(leastIndex);
        }
    }
    // 若是權重相同或權重爲0則均等隨機
    // If all invokers have the same weight value or totalWeight=0, return evenly.
    return invokers.get(leastIndexs[ThreadLocalRandom.current().nextInt(leastCount)]);
}

簡單思路介紹

歸納起來就兩部分,一部分是活躍數和權重的統計,另外一部分是選擇invoker.也就是他把最小活躍數的invoker統計到leastIndexs數組中,若是權重一致(這個一致的規則參考上面的隨機算法)或者總權重爲0,則均等隨機調用,若是不一樣,則從leastIndexs數組中按照權重比例調用

算法說明

最小活躍數算法實現：
假定有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

流程圖

ConsistentHashLoadBalance(一致性哈希)

一致性 Hash，相同參數的請求老是發到同一提供者。

當某一臺提供者掛時，本來發往該提供者的請求，基於虛擬節點，平攤到其它提供者，不會引發劇烈變更。

源碼其實分爲四個步驟

定義全局一致性hash選擇器的ConcurrentMap<String, ConsistentHashSelector<?>> selectors，key爲方法名稱，例如com.alibaba.dubbo.demo.TestService.getRandomNumber
若是一致性hash選擇器不存在或者與之前保存的一致性hash選擇器不同（即dubbo服務provider有變化，經過System.identityHashCode(invokers)計算一個identityHashCode值）則須要從新構造一個一致性hash選擇器
構造一個一致性hash選擇器ConsistentHashSelector的源碼以下，經過參數i和h打散Invoker在TreeMap上的位置，replicaNumber默認值爲160，因此最終virtualInvokers這個TreeMap的size爲invokers.size()*replicaNumber
選擇Invoker的步驟
1. 根據Invocation中的參數invocation.getArguments()轉成key
2. 算出這個key的md5值
3. 根據md5值的hash值從TreeMap中選擇一個Invoker

下面源碼解析+註釋

public class ConsistentHashLoadBalance extends AbstractLoadBalance {
    public static final String NAME = "consistenthash";

    /**
     * 服務方法與一致性哈希選擇器的映射
     *
     * KEY：serviceKey + "." + methodName
     */
    private final ConcurrentMap<String, ConsistentHashSelector<?>> selectors = new ConcurrentHashMap<String, ConsistentHashSelector<?>>();

    @SuppressWarnings("unchecked")
    @Override
    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        String methodName = RpcUtils.getMethodName(invocation);
        // 基於 invokers 集合，根據對象內存地址來計算定義哈希值
        String key = invokers.get(0).getUrl().getServiceKey() + "." + methodName;
        int identityHashCode = System.identityHashCode(invokers);
        // 得到 ConsistentHashSelector 對象。若爲空，或者定義哈希值變動（說明 invokers 集合發生變化），
        // 進行建立新的 ConsistentHashSelector 對象
        ConsistentHashSelector<T> selector = (ConsistentHashSelector<T>) selectors.get(key);
        if (selector == null || selector.identityHashCode != identityHashCode) {
            selectors.put(key, new ConsistentHashSelector<T>(invokers, methodName, identityHashCode));
            selector = (ConsistentHashSelector<T>) selectors.get(key);
        }
        return selector.select(invocation);
    }

    private static final class ConsistentHashSelector<T> {

        /**
         * 虛擬節點與 Invoker 的映射關係
         */
        private final TreeMap<Long, Invoker<T>> virtualInvokers;

        /**
         * 每一個Invoker 對應的虛擬節點數
         */
        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>>();
            // 設置 identityHashCode
            this.identityHashCode = identityHashCode;
            URL url = invokers.get(0).getUrl();
            // 初始化 replicaNumber
            this.replicaNumber = url.getMethodParameter(methodName, "hash.nodes", 160);
            // 初始化 argumentIndex
            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]);
            }
            // 初始化 virtualInvokers
            for (Invoker<T> invoker : invokers) {
                String address = invoker.getUrl().getAddress();
                // 每四個虛擬結點爲一組，爲何這樣？下面會說到
                for (int i = 0; i < replicaNumber / 4; i++) {
                    // 這組虛擬結點獲得唯一名稱
                    byte[] digest = md5(address + i);
                    // Md5是一個16字節長度的數組，將16字節的數組每四個字節一組，
                    // 分別對應一個虛擬結點，這就是爲何上面把虛擬結點四個劃分一組的緣由
                    for (int h = 0; h < 4; h++) {
                        // 對於每四個字節，組成一個long值數值，作爲這個虛擬節點的在環中的唯一key
                        long m = hash(digest, h);
                        virtualInvokers.put(m, invoker);
                    }
                }
            }
        }

        public Invoker<T> select(Invocation invocation) {
            // 基於方法參數，得到 KEY
            String key = toKey(invocation.getArguments());
            // 計算 MD5 值
            byte[] digest = md5(key);
            // 計算 KEY 值
            return selectForKey(hash(digest, 0));
        }

        /**
         * 基於方法參數，得到 KEY
         * @param args
         * @return
         */
        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();
        }

        /**
         * 選一個 Invoker 對象
         * @param hash
         * @return
         */
        private Invoker<T> selectForKey(long hash) {
            // 獲得大於當前 key 的那個子 Map ，而後從中取出第一個 key ，就是大於且離它最近的那個 key
            Map.Entry<Long, Invoker<T>> entry = virtualInvokers.ceilingEntry(hash);
            // 不存在，則取 virtualInvokers 第一個
            if (entry == null) {
                entry = virtualInvokers.firstEntry();
            }
            // 存在，則返回
            return entry.getValue();
        }

        /**
         * 對於每四個字節，組成一個 Long 值數值，作爲這個虛擬節點的在環中的唯一 KEY
         * @param digest
         * @param number
         * @return
         */
        private long hash(byte[] digest, int number) {
            return (((long) (digest[3 + number * 4] & 0xFF) << 24)
                    | ((long) (digest[2 + number * 4] & 0xFF) << 16)
                    | ((long) (digest[1 + number * 4] & 0xFF) << 8)
                    | (digest[number * 4] & 0xFF))
                    & 0xFFFFFFFFL;
        }

        /**
         * MD5 是一個 16 字節長度的數組，將 16 字節的數組每四個字節一組，
         * 分別對應一個虛擬結點，這就是爲何上面把虛擬結點四個劃分一組的緣由
         * @param value
         * @return
         */
        private byte[] md5(String value) {
            MessageDigest md5;
            try {
                md5 = MessageDigest.getInstance("MD5");
            } catch (NoSuchAlgorithmException e) {
                throw new IllegalStateException(e.getMessage(), e);
            }
            md5.reset();
            byte[] bytes;
            try {
                bytes = value.getBytes("UTF-8");
            } catch (UnsupportedEncodingException e) {
                throw new IllegalStateException(e.getMessage(), e);
            }
            md5.update(bytes);
            return md5.digest();
        }

    }

}

一致性哈希算法的三個關鍵點原理， down機影響面， 虛擬節點

原理

一致性Hash(Consistent Hashing)原理剖析

down 機影響

在某個節點掛機的時候，會根據虛擬節點選擇下一個節點。隻影響到一個節點，其餘的節點不受到影響

虛擬節點

根據一致性Hash算法將生成不少的虛擬節點，這些節點落在圓環中。當某個節點down掉，則壓力會給到指定的虛擬節點