負載均衡算法WeightedRoundRobin(加權輪詢)簡介及算法實現

        Nginx的負載均衡默認算法是加權輪詢算法，本文簡單介紹算法的邏輯，並給出算法的Java實現版本。

        本文參考了Nginx的負載均衡 - 加權輪詢 (Weighted Round Robin) 。

        算法簡介

        有三個節點{a, b, c}，他們的權重分別是{a=5, b=1, c=1}。發送7次請求，a會被分配5次，b會被分配1次，c會被分配1次。

        一般的算法可能是：

        1、輪訓所有節點，找到一個最大權重節點；

        2、選中的節點權重-1；

        3、直到減到0，恢復該節點原始權重，繼續輪詢；

        這樣的算法看起來簡單，最終效果是：{a, a, a, a, a, b, c}，即前5次可能選中的都是a，這可能造成權重大的服務器造成過大壓力的同時，小權重服務器還很閑。

        Nginx的加權輪詢算法將保持選擇的平滑性，希望達到的效果可能是{a, b, a, a, c, a, a}，即盡可能均勻的分攤節點，節點分配不再是連續的。

 

        Nginx加權輪詢算法

        1、概念解釋，每個節點有三個權重變量，分別是：

        (1) weight: 約定權重，即在配置文件或初始化時約定好的每個節點的權重。

        (2) effectiveWeight: 有效權重，初始化為weight。

         在通訊過程中發現節點異常，則-1；

         之后再次選取本節點，調用成功一次則+1，直達恢復到weight；

         此變量的作用主要是節點異常，降低其權重。

        (3) currentWeight: 節點當前權重，初始化為0。

 

        2、算法邏輯

        (1) 輪詢所有節點，計算當前狀態下所有節點的effectiveWeight之和totalWeight；

        (2) currentWeight = currentWeight + effectiveWeight;  選出所有節點中currentWeight中最大的一個節點作為選中節點；

        (3) 選中節點的currentWeight = currentWeight - totalWeight；

 

        基於以上算法，我們看一個例子：

        這時有三個節點{a, b, c}，權重分別是{a=4, b=2, c=1}，共7次請求，初始currentWeight值為{0, 0, 0}，每次分配后的結果如下： 

請求序號	請求前currentWeight值	選中節點	請求后currentWeight值
1	{c=1,b=2,a=4}	a	{c=1,b=2,a=-3}
2	{c=2,b=4,a=1}	b	{c=2,b=-3,a=1}
3	{c=3,b=-1,a=5}	a	{c=3,b=-1,a=-2}
4	{c=4,b=1,a=2}	c	{c=-3,b=1,a=2}
5	{c=-2,b=3,a=6}	a	{c=-2,b=3,a=-1}
6	{c=-1,b=5,a=3}	b	{c=-1,b=-2,a=3}
7	{c=0,b=0,a=7}	a	{c=0,b=0,a=0}

        觀察到七次調用選中的節點順序為{a, b, a, c, a, b, a}，a節點選中4次，b節點選中2次，c節點選中1次，算法保持了currentWeight值從初始值{c=0,b=0,a=0}到7次調用后又回到{c=0,b=0,a=0}。

 

        算法實現

        下面附上筆者自己的Java版算法實現：

package com.example.demo.arithmetic;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created by caojun on 2018/2/20.
 *
 * 基本概念：
 * weight: 配置文件中指定的該后端的權重，這個值是固定不變的。
 * effective_weight: 后端的有效權重，初始值為weight。
 * 在釋放后端時，如果發現和后端的通信過程中發生了錯誤，就減小effective_weight。
 * 此后有新的請求過來時，在選取后端的過程中，再逐步增加effective_weight，最終又恢復到weight。
 * 之所以增加這個字段，是為了當后端發生錯誤時，降低其權重。
 * current_weight:
 * 后端目前的權重，一開始為0，之后會動態調整。那么是怎么個動態調整呢？
 * 每次選取后端時，會遍歷集群中所有后端，對於每個后端，讓它的current_weight增加它的effective_weight，
 * 同時累加所有后端的effective_weight，保存為total。
 * 如果該后端的current_weight是最大的，就選定這個后端，然后把它的current_weight減去total。
 * 如果該后端沒有被選定，那么current_weight不用減小。
 *
 * 算法邏輯：
 * 1. 對於每個請求，遍歷集群中的所有可用后端，對於每個后端peer執行：
 *     peer->current_weight += peer->effecitve_weight。
 *     同時累加所有peer的effective_weight，保存為total。
 * 2. 從集群中選出current_weight最大的peer，作為本次選定的后端。
 * 3. 對於本次選定的后端，執行：peer->current_weight -= total。
 *
 */
public class RoundRobinByWeightLoadBalance {

    //約定的invoker和權重的鍵值對
    final private List<Node> nodes;

    public RoundRobinByWeightLoadBalance(Map<Invoker, Integer> invokersWeight){
        if (invokersWeight != null && !invokersWeight.isEmpty()) {
            nodes = new ArrayList<>(invokersWeight.size());
            invokersWeight.forEach((invoker, weight)->nodes.add(new Node(invoker, weight)));
        }else
            nodes = null;
    }

    /**
     * 算法邏輯：
     * 1. 對於每個請求，遍歷集群中的所有可用后端，對於每個后端peer執行：
     *     peer->current_weight += peer->effecitve_weight。
     *     同時累加所有peer的effective_weight，保存為total。
     * 2. 從集群中選出current_weight最大的peer，作為本次選定的后端。
     * 3. 對於本次選定的后端，執行：peer->current_weight -= total。
     *
     * @Return ivoker
     */
    public Invoker select(){
        if (! checkNodes())
            return null;
        else if (nodes.size() == 1) {
            if (nodes.get(0).invoker.isAvalable())
                return nodes.get(0).invoker;
            else
                return null;
        }
        Integer total = 0;
        Node nodeOfMaxWeight = null;
        for (Node node : nodes) {
            total += node.effectiveWeight;
            node.currentWeight += node.effectiveWeight;

            if (nodeOfMaxWeight == null) {
                nodeOfMaxWeight = node;
            }else{
                nodeOfMaxWeight = nodeOfMaxWeight.compareTo(node) > 0 ? nodeOfMaxWeight : node;
            }
        }

        nodeOfMaxWeight.currentWeight -= total;
        return nodeOfMaxWeight.invoker;
    }

    public void onInvokeSuccess(Invoker invoker){
        if (checkNodes()){
            nodes.stream()
                    .filter((Node node)->invoker.id().equals(node.invoker.id()))
                    .findFirst()
                    .get()
                    .onInvokeSuccess();
        }
    }

    public void onInvokeFail(Invoker invoker){
        if (checkNodes()){
            nodes.stream()
                    .filter((Node node)->invoker.id().equals(node.invoker.id()))
                    .findFirst()
                    .get()
                    .onInvokeFail();
        }
    }

    private boolean checkNodes(){
        return (nodes != null && nodes.size() > 0);
    }

    public void printCurrenctWeightBeforeSelect(){
        if (checkNodes()) {
            final StringBuffer out = new StringBuffer("{");
            nodes.forEach(node->out.append(node.invoker.id())
                    .append("=")
                    .append(node.currentWeight+node.effectiveWeight)
                    .append(","));
            out.append("}");
            System.out.print(out);
        }
    }

    public void printCurrenctWeight(){
        if (checkNodes()) {
            final StringBuffer out = new StringBuffer("{");
            nodes.forEach(node->out.append(node.invoker.id())
                    .append("=")
                    .append(node.currentWeight)
                    .append(","));
            out.append("}");
            System.out.print(out);
        }
    }

    public interface Invoker{
        Boolean isAvalable();
        String id();
    }

    private static class Node implements Comparable<Node>{
        final Invoker invoker;
        final Integer weight;
        Integer effectiveWeight;
        Integer currentWeight;

        Node(Invoker invoker, Integer weight){
            this.invoker = invoker;
            this.weight = weight;
            this.effectiveWeight = weight;
            this.currentWeight = 0;
        }

        @Override
        public int compareTo(Node o) {
            return currentWeight > o.currentWeight ? 1 : (currentWeight.equals(o.currentWeight) ? 0 : -1);
        }

        public void onInvokeSuccess(){
            if (effectiveWeight < this.weight)
                effectiveWeight++;
        }

        public void onInvokeFail(){
            effectiveWeight--;
        }
    }

    public static void main(String[] args){
        Map<Invoker, Integer> invokersWeight = new HashMap<>(3);
        Integer aWeight = 4;
        Integer bWeight = 2;
        Integer cWeight = 1;

        invokersWeight.put(new Invoker() {
            @Override
            public Boolean isAvalable() {
                return true;
            }
            @Override
            public String id() {
                return "a";
            }
        }, aWeight);

        invokersWeight.put(new Invoker() {
            @Override
            public Boolean isAvalable() {
                return true;
            }
            @Override
            public String id() {
                return "b";
            }
        }, bWeight);

        invokersWeight.put(new Invoker() {
            @Override
            public Boolean isAvalable() {
                return true;
            }
            @Override
            public String id() {
                return "c";
            }
        }, cWeight);

        Integer times = 7;
        RoundRobinByWeightLoadBalance roundRobin = new RoundRobinByWeightLoadBalance(invokersWeight);
        for(int i=1; i<=times; i++){
            System.out.print(new StringBuffer(i+"").append("    "));
            roundRobin.printCurrenctWeightBeforeSelect();
            Invoker invoker = roundRobin.select();
            System.out.print(new StringBuffer("    ").append(invoker.id()).append("    "));
            roundRobin.printCurrenctWeight();
            System.out.println();
        }
    }
}