dubbo負載均衡之RoundRobin

承接上文，本文繼續(xù)來看dubbo的加權(quán)RoundRobin負載均衡策略。先大概了解一下RoundRobin纷纫，加權(quán)RoundRobin就是按照節(jié)點(服務(wù)節(jié)點叁扫、存儲節(jié)點等)權(quán)重三妈，將請求分發(fā)到不同節(jié)點如下所示：

考慮有三個節(jié)點 A(5)、B(1)莫绣、C(1),括號內(nèi)是該節(jié)點的服務(wù)負載能力畴蒲，假設(shè)有10個請求(R)，下面是普通RoundRobin的做法

R1 -> A 对室， R2 -> v 模燥，R3 -> A ，R4 -> A 掩宜， R5 -> A 
R6 -> B
R7 -> C 
可以看到蔫骂，請求會按照權(quán)重分配到不同的節(jié)點。但是考慮一種情況锭亏，如果把請求量級放大到10^n,此時70%請求
會連續(xù)打到節(jié)點A纠吴，而其他節(jié)點則0請求；可能A節(jié)點被打掛慧瘤，流量也不會到BC節(jié)點戴已，不符合負載均衡的初衷固该。

那么，有沒有一種方式糖儡，可以讓請求平均但不連續(xù)的打到某一個節(jié)點伐坏？這就是下面要引出的平滑權(quán)重RoundRobin（比如nginx采用的負載均衡），下面來看下這種策略的實現(xiàn)方式：

單次請求握联，選擇策略：
1桦沉、選取所有節(jié)點中權(quán)重最大的節(jié)點（curMax）
2、更新當前權(quán)重最大節(jié)點的權(quán)重金闽，節(jié)點新權(quán)重（curMaxNew） = 節(jié)點當前權(quán)重（curMax） - 所有節(jié)點當前權(quán)重之和（total）
3纯露、所有節(jié)點權(quán)重更新，節(jié)點新權(quán)重（curNew） = 節(jié)點當前權(quán)重（cur） + 節(jié)點初始權(quán)重（init）
結(jié)束一次請求選擇過程代芜。

用代碼描述可能更清晰一點：

//初始化節(jié)點權(quán)重map
static{
        MAP.put("A",5);
        MAP.put("B",1);
        MAP.put("C",1);
}

/**
* 平滑加權(quán)輪詢
**/
static List<String> smoothRoundRobin(Map<String,Integer> candidateMap,int requestCount){
        Map<String,Integer> currentRoundMap = new HashMap<>(candidateMap);
        Integer sumWeight = candidateMap.entrySet().stream().mapToInt(value -> value.getValue()).sum();

        List<String> resultList = new ArrayList<>();
        for(int i=0;i<requestCount;i++){
            //當前權(quán)重最大的節(jié)點key,例如埠褪，初始是5
            String currentMaxWeight = currentMaxWeight(currentRoundMap);
            //選中，放入結(jié)果
            resultList.add(currentMaxWeight);
            //更新當前權(quán)重最大節(jié)點的權(quán)重挤庇；當前權(quán)重最大節(jié)點權(quán)重- 總權(quán)重
            currentRoundMap.put(currentMaxWeight,currentRoundMap.get(currentMaxWeight) - sumWeight);
            //權(quán)重重新分配钞速；當前每個節(jié)點權(quán)重 + 初始每個節(jié)點權(quán)重
            reRoundMapValue(currentRoundMap,MAP);
        }
        return resultList;
}

/**
* 輔助方法，尋找當前權(quán)重值最大節(jié)點
**/
static String currentMaxWeight(Map<String,Integer> candidateMap){
        List<Integer> weightList = new ArrayList<>(candidateMap.values());

        weightList.sort(Integer::compareTo);
        int maxWeight = weightList.get(weightList.size() - 1);
        for(Map.Entry<String,Integer> entry : candidateMap.entrySet()){
            if(entry.getValue() == maxWeight){
                return entry.getKey();
            }
        }
        return "";
}

/**
* 所有節(jié)點權(quán)重重置
**/
static void reRoundMapValue(Map<String,Integer> roundMap,Map<String,Integer> initialCandidateMap){
        roundMap.forEach((candidate,currentWeight) ->{
            currentWeight += initialCandidateMap.get(candidate);
            roundMap.put(candidate,currentWeight);
        });
}

//執(zhí)行結(jié)果：A->A->B->A->C->A->A
public static void main(String[] args) {
        List<String> roundRobinResult = smoothRoundRobin(MAP,7);
        System.out.println(StringUtils.join(roundRobinResult,"->"));
}

可以看到嫡秕，上面的策略能夠一定程度上保證請求不會持續(xù)打在一個節(jié)點上渴语，相對平均。dubbo的RoundRobin其實參考了nginx的負載均衡昆咽，邏輯類似驾凶，下面來看下dubbo的實現(xiàn)

// 權(quán)重實體，封裝了初始權(quán)重掷酗、當前權(quán)重以及上次更新時間狭郑；普通getter、setter方法省略汇在；
// 注意，這里除了權(quán)重值意外脏答，沒有invoker的相關(guān)信息
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.set(0);
    }
    public long increaseCurrent() {
        return current.addAndGet(weight);
    }
    public void sel(int total) {
        current.addAndGet(-1 * total);
    }
}

//權(quán)重實體緩存map糕殉，結(jié)構(gòu) <serviceKey.methodName,<URL.identity,權(quán)重實體>
private ConcurrentMap<String, ConcurrentMap<String, WeightedRoundRobin>> methodWeightMap = new ConcurrentHashMap<String, ConcurrentMap<String, WeightedRoundRobin>>();
//權(quán)重更新鎖
private AtomicBoolean updateLock = new AtomicBoolean();

/**
* 核心select方法
**/
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
            //方法級負載均衡
        String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
        ConcurrentMap<String, WeightedRoundRobin> map = methodWeightMap.get(key);
        if (map == null) {
            methodWeightMap.putIfAbsent(key, new ConcurrentHashMap<String, WeightedRoundRobin>());
            map = methodWeightMap.get(key);
        }
        int totalWeight = 0;
        //這里就很靈性了，保證后期maxCurrent值不會溢出殖告。
        long maxCurrent = Long.MIN_VALUE;
        long now = System.currentTimeMillis();
        Invoker<T> selectedInvoker = null;
        WeightedRoundRobin selectedWRR = null;
        // 遍歷所有invoker阿蝶，為每個invoker封裝一個WeightRoundRobin實體；
        for (Invoker<T> invoker : invokers) {
            // methodWeightMap的內(nèi)層map黄绩，key是url唯一標識
            String identifyString = invoker.getUrl().toIdentityString();
            WeightedRoundRobin weightedRoundRobin = map.get(identifyString);
            //調(diào)用基類權(quán)重計算方法初始化權(quán)重
            int weight = getWeight(invoker, invocation);

            if (weightedRoundRobin == null) {
                weightedRoundRobin = new WeightedRoundRobin();
                weightedRoundRobin.setWeight(weight);
                map.putIfAbsent(identifyString, weightedRoundRobin);
            }
            if (weight != weightedRoundRobin.getWeight()) {
                //weight changed,current置0
                weightedRoundRobin.setWeight(weight);
            }
            long cur = weightedRoundRobin.increaseCurrent();
            weightedRoundRobin.setLastUpdate(now);
            //初始maxCurrent為0羡洁，選組權(quán)重值最大的invoker
            if (cur > maxCurrent) {
                maxCurrent = cur
                selectedInvoker = invoker;
                selectedWRR = weightedRoundRobin;
            }
            totalWeight += weight;
        }
        // 這里做了優(yōu)化，invokers.size與map大小不一致爽丹，這時候意味著有的節(jié)點可能掛掉了需要剔除筑煮，重制RoundRobin實體
        // 新增一個updateTime用于限時更新methodWeightMap;
        // 更新策略：超過循環(huán)周期60s的節(jié)點
        if (!updateLock.get() && invokers.size() != map.size()) {
            if (updateLock.compareAndSet(false, true)) {
                try {
                    // copy -> modify -> update reference
                    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);
                }
            }
        }
            
        //一次選擇結(jié)束辛蚊，重置該WeightedRoundRobin權(quán)重（ 新權(quán)重 = 當前權(quán)重 - 總權(quán)重），然后返回真仲。
        if (selectedInvoker != null) {
            selectedWRR.sel(totalWeight);
            return selectedInvoker;
        }
        // should not happen here
        return invokers.get(0);
    }

dubbo的負載均衡是方法級的袋马，核心邏輯如下：

初次select時，針對每一個method的invoker秸应，初始化invoker權(quán)重
實例化WeightedRoundRobin并放入緩存虑凛；后續(xù)會根據(jù)invoker.gerUrl.identify獲取緩存的weightRoundRobin實體，在此基礎(chǔ)上做權(quán)重更新软啼；
每次select cur 自增(cur += weight),同時更新update時間戳桑谍；
選中cur最大的invoker作為返回結(jié)果，返回前會重置該WeightedRoundRobin的cur值祸挪；
dubbo在實際計算過程中加了一個超時時間锣披，如果當前時間戳 - 更新時間 > 超時時間，則認為該節(jié)點可能掛掉了匕积，直接從列表剔除盈罐，下一次select會重新初始化。

注：參考 dubbo源碼版本 2.7.1闪唆，歡迎指正盅粪。