Ribbon提供的負載均衡算法IRule（四）

時間 2020-12-20

標籤 git github 算法緩存服務器併發負載均衡 dom ide 函數欄目負載均衡简体版

原文原文鏈接

1、Ribbon算法的介紹

Ribbon的源碼地址：https://github.com/Netflix/ribbongit

IRule：根據特定算法中從服務器列表中選取一個要訪問的服務，Ribbon默認的算法爲ZoneAvoidanceRule；github

Ribbon中的7中負載均衡算法：算法

（1）RoundRobinRule：輪詢；緩存

（2）RandomRule：隨機；服務器

（3）AvailabilityFilteringRule：會先過濾掉因爲屢次訪問故障而處於斷路器狀態的服務，還有併發的鏈接數量超過閾值的服務，而後對剩餘的服務列表按照輪詢策略進行訪問；併發

（4）WeightedResponseTimeRule：根據平均響應時間計算全部服務的權重，響應時間越快的服務權重越大被選中的機率越大。剛啓動時若是統計信息不足，則使用RoundRobinRule（輪詢）策略，等統計信息足夠，會切換到WeightedResponseTimeRule；負載均衡

（5）RetryRule：先按照RoundRobinRule（輪詢）策略獲取服務，若是獲取服務失敗則在指定時間內進行重試，獲取可用的服務；dom

（6）BestAvailableRule：會先過濾掉因爲屢次訪問故障而處於斷路器跳閘狀態的服務，而後選擇一個併發量最小的服務；ide

（7）ZoneAvoidanceRule：複合判斷Server所在區域的性能和Server的可用性選擇服務器，在沒有Zone的狀況下是相似輪詢的算法；函數

ribbion的負載均衡算法結構：

2、配置指定的負載均衡算法

一、打開消費者工程，增長以下的配置

@Configuration
public class ConfigBean
{
    @Bean
    @LoadBalanced //Ribbon 是客戶端負載均衡的工具；
    public RestTemplate getRestTemplate()
    {
        return new RestTemplate();
    }

    //配置負載均衡的策略爲隨機，默認算法爲輪詢算法
    @Bean
    public IRule myRule()
    {
        //return new RoundRobinRule();
        return new RandomRule();  
    }      
}

二、啓動類增長 @EnableEurekaClient 註解

@SpringBootApplication
@EnableEurekaClient     //本服務啓動後自動註冊到eureka中（若是用了註冊中心記得加）
public class DeptProvider8001_App
{
    public static void main(String[] args)
    {
        SpringApplication.run(DeptProvider8001_App.class, args);
    }
}

三、而後重啓這個消費者服務，訪問；能夠查看到隨機訪問生產者服務。

3、RetryRule(重試)

//具有重試機制的實例選擇功能
public class RetryRule extends AbstractLoadBalancerRule {
    //默認使用RoundRobinRule實例
    IRule subRule = new RoundRobinRule();
    //閾值爲500ms
    long maxRetryMillis = 500;
 
    public RetryRule() {
    }
 
    public RetryRule(IRule subRule) {
        this.subRule = (subRule != null) ? subRule : new RoundRobinRule();
    }
 
    public RetryRule(IRule subRule, long maxRetryMillis) {
        this.subRule = (subRule != null) ? subRule : new RoundRobinRule();
        this.maxRetryMillis = (maxRetryMillis > 0) ? maxRetryMillis : 500;
    }
 
    public void setRule(IRule subRule) {
        this.subRule = (subRule != null) ? subRule : new RoundRobinRule();
    }
 
    public IRule getRule() {
        return subRule;
    }
 
    public void setMaxRetryMillis(long maxRetryMillis) {
        if (maxRetryMillis > 0) {
            this.maxRetryMillis = maxRetryMillis;
        } else {
            this.maxRetryMillis = 500;
        }
    }
 
    public long getMaxRetryMillis() {
        return maxRetryMillis;
    }
 
    
    
    @Override
    public void setLoadBalancer(ILoadBalancer lb) {        
        super.setLoadBalancer(lb);
        subRule.setLoadBalancer(lb);
    }
 
 
    public Server choose(ILoadBalancer lb, Object key) {
        long requestTime = System.currentTimeMillis();
        long deadline = requestTime + maxRetryMillis;
 
        Server answer = null;
 
        answer = subRule.choose(key);
 
        if (((answer == null) || (!answer.isAlive()))
                && (System.currentTimeMillis() < deadline)) {
 
            InterruptTask task = new InterruptTask(deadline
                    - System.currentTimeMillis());
            //反覆重試
            while (!Thread.interrupted()) {
                //選擇實例
                answer = subRule.choose(key);
                //500ms內沒選擇到就返回null
                if (((answer == null) || (!answer.isAlive()))
                        && (System.currentTimeMillis() < deadline)) {
                    /* pause and retry hoping it's transient */
                    Thread.yield();
                } 
                else //若能選擇到實例，就返回
                {
                    break;
                }
            }
 
            task.cancel();
        }
 
        if ((answer == null) || (!answer.isAlive())) {
            return null;
        } else {
            return answer;
        }
    }
 
    @Override
    public Server choose(Object key) {
        return choose(getLoadBalancer(), key);
    }
 
    @Override
    public void initWithNiwsConfig(IClientConfig clientConfig) {
    }
}

4、WeightedResponseTimeRule（權重）

WeightedResponseTimeRule這個策略每30秒計算一次服務器響應時間，以響應時間做爲權重，響應時間越短的服務器被選中的機率越大。它有一個LoadBalancerStats類，這裏面有三個緩存，它的做用是記住發起請求服務提供者的一些參數，例如響應時間

有了緩存數據後權重是怎麼處理的呢，下面看WeightedResponseTimeRule，它有一個定時任務，定時去計算權重

//該策略是對RoundRobinRule的擴展，增長了根據實例的運行狀況來計算權重
//並根據權重來挑選實例，以達到更優的分配效果
public class WeightedResponseTimeRule extends RoundRobinRule {
 
    public static final IClientConfigKey<Integer> WEIGHT_TASK_TIMER_INTERVAL_CONFIG_KEY = new IClientConfigKey<Integer>() {
        @Override
        public String key() {
            return "ServerWeightTaskTimerInterval";
        }
        
        @Override
        public String toString() {
            return key();
        }
 
        @Override
        public Class<Integer> type() {
            return Integer.class;
        }
    };
    //默認30秒執行一次
    public static final int DEFAULT_TIMER_INTERVAL = 30 * 1000;
    
    private int serverWeightTaskTimerInterval = DEFAULT_TIMER_INTERVAL;
 
    private static final Logger logger = LoggerFactory.getLogger(WeightedResponseTimeRule.class);
    
    // 存儲權重的對象，該List中每一個權重所處的位置對應了負載均衡器維護實例清單中全部實例在
    //清單中的位置。
    private volatile List<Double> accumulatedWeights = new ArrayList<Double>();
    
 
    private final Random random = new Random();
 
    protected Timer serverWeightTimer = null;
 
    protected AtomicBoolean serverWeightAssignmentInProgress = new AtomicBoolean(false);
 
    String name = "unknown";
 
    public WeightedResponseTimeRule() {
        super();
    }
 
    public WeightedResponseTimeRule(ILoadBalancer lb) {
        super(lb);
    }
    
    @Override
    public void setLoadBalancer(ILoadBalancer lb) {
        super.setLoadBalancer(lb);
        if (lb instanceof BaseLoadBalancer) {
            name = ((BaseLoadBalancer) lb).getName();
        }
        initialize(lb);
    }
 
    void initialize(ILoadBalancer lb) {        
        if (serverWeightTimer != null) {
            serverWeightTimer.cancel();
        }
        serverWeightTimer = new Timer("NFLoadBalancer-serverWeightTimer-"
                + name, true);
        //啓動一個定時任務，用來爲每一個服務實例計算權重，默認30秒執行一次，調用DynamicServerWeighTask方法，向下找
        serverWeightTimer.schedule(new DynamicServerWeightTask(), 0,
                serverWeightTaskTimerInterval);
        // do a initial run
        ServerWeight sw = new ServerWeight(); 
        sw.maintainWeights();
 
        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
            public void run() {
                logger
                        .info("Stopping NFLoadBalancer-serverWeightTimer-"
                                + name);
                serverWeightTimer.cancel();
            }
        }));
    }
 
    public void shutdown() {
        if (serverWeightTimer != null) {
            logger.info("Stopping NFLoadBalancer-serverWeightTimer-" + name);
            serverWeightTimer.cancel();
        }
    }
 
    List<Double> getAccumulatedWeights() {
        return Collections.unmodifiableList(accumulatedWeights);
    }
 
    /*
    第一步：生成一個[0,maxTotalWeight]的隨機值
    第二步：遍歷權重列表，比較權重值與隨機數的大小，若是權重值大於隨機數，就拿當前權重列表
    的索引值去服務實例表獲取具體的實例。
    */
    @edu.umd.cs.findbugs.annotations.SuppressWarnings(value = "RCN_REDUNDANT_NULLCHECK_OF_NULL_VALUE")
    @Override
    public Server choose(ILoadBalancer lb, Object key) {
        if (lb == null) {
            return null;
        }
        Server server = null;
 
        while (server == null) {
            // get hold of the current reference in case it is changed from the other thread
            List<Double> currentWeights = accumulatedWeights;
            if (Thread.interrupted()) {
                return null;
            }
            List<Server> allList = lb.getAllServers();
 
            int serverCount = allList.size();
 
            if (serverCount == 0) {
                return null;
            }
 
            int serverIndex = 0;
 
            // 獲取最後一個實例的權重
            double maxTotalWeight = currentWeights.size() == 0 ? 0 : currentWeights.get(currentWeights.size() - 1);
            // 若是最後一個實例的權重小於0.001，則採用父類實現的現象輪詢的策略
            if (maxTotalWeight < 0.001d || serverCount != currentWeights.size()) {
                server =  super.choose(getLoadBalancer(), key);
                if(server == null) {
                    return server;
                }
            } else {
                // 產生一個[0,maxTotalWeight]的隨機值
                double randomWeight = random.nextDouble() * maxTotalWeight;
                int n = 0;
                for (Double d : currentWeights) {
                    //若是遍歷維護的權重清單，若權重值大於隨機獲得的數值，就選擇這個實例
                    if (d >= randomWeight) {
                        serverIndex = n;
                        break;
                    } else {
                        n++;
                    }
                }
 
                server = allList.get(serverIndex);
            }
 
            if (server == null) {
                /* Transient. */
                Thread.yield();
                continue;
            }
 
            if (server.isAlive()) {
                return (server);
            }
 
            // Next.
            server = null;
        }
        return server;
    }
 
    class DynamicServerWeightTask extends TimerTask {
        public void run() {
            ServerWeight serverWeight = new ServerWeight();
            try {
                //點擊maintainWeights能夠進入權重計算的方法
                serverWeight.maintainWeights(); 
            } catch (Exception e) {
                logger.error("Error running DynamicServerWeightTask for {}", name, e);
            }
        }
    }
 
    class ServerWeight {
/*該函數主要分爲兩個步驟
1 根據LoadBalancerStats中記錄的每一個實例的統計信息，累計全部實例的平均響應時間，
獲得總的平均響應時間totalResponseTime，該值用於後面的計算。
2 爲負載均衡器中維護的實例清單逐個計算權重（從第一個開始），計算規則爲：
weightSoFar+totalResponseTime-實例平均相應時間，其中weightSoFar初始化爲0，而且
每計算好一個權重須要累加到weightSoFar上供下一次計算使用。
示例：4個實例A、B、C、D，它們的平均響應時間爲10,40,80,100，因此總的響應時間爲
230，每一個實例的權重爲總響應時間與實例自身的平均響應時間的差的累積所得，因此實例A
B，C，D的權重分別爲：
A：230-10=220
B：220+230-40=410
C：410+230-80=560
D：560+230-100=690
須要注意的是，這裏的權重值只是表示各實例權重區間的上限，並不是某個實例的優先級，因此不
是數值越大被選中的機率就越大。而是由實例的權重區間來決定選中的機率和優先級。
A：[0,220]
B：(220,410]
C：(410,560]
D：(560,690)
實際上每一個區間的寬度就是：總的平均響應時間-實例的平均響應時間，因此實例的平均響應時間越短
，權重區間的寬度越大，而權重區間寬度越大被選中的機率就越大。
*/
        public void maintainWeights() {
            ILoadBalancer lb = getLoadBalancer();
            if (lb == null) {
                return;
            }
            
            if (!serverWeightAssignmentInProgress.compareAndSet(false,  true))  {
                return;
            }
            
            try {
                logger.info("Weight adjusting job started");
                AbstractLoadBalancer nlb = (AbstractLoadBalancer) lb;
                //根據負載均衡器的狀態信息進行計算
                LoadBalancerStats stats = nlb.getLoadBalancerStats();
                if (stats == null) {
                    // no statistics, nothing to do
                    return;
                }
                double totalResponseTime = 0;
                // 計算全部實例的平均響應時間的總和
                for (Server server : nlb.getAllServers()) {
                    // 若是服務實例的狀態快照不在緩存中，那麼這裏會進行自動加載
                    ServerStats ss = stats.getSingleServerStat(server);
                    totalResponseTime += ss.getResponseTimeAvg();
                }
                // 逐個計算每一個實例的權重
                Double weightSoFar = 0.0;
                
                // create new list and hot swap the reference
                List<Double> finalWeights = new ArrayList<Double>();
                //weightSoFar+totalResponseTime-實例平均相應時間
                for (Server server : nlb.getAllServers()) {
                    ServerStats ss = stats.getSingleServerStat(server);
                    double weight = totalResponseTime - ss.getResponseTimeAvg();
                    weightSoFar += weight;
                    finalWeights.add(weightSoFar);   
                }
                //將計算結果進行保存
                setWeights(finalWeights);
            } catch (Exception e) {
                logger.error("Error calculating server weights", e);
            } finally {
                serverWeightAssignmentInProgress.set(false);
            }
 
        }
    }
 
    void setWeights(List<Double> weights) {
        this.accumulatedWeights = weights;
    }
 
    @Override
    public void initWithNiwsConfig(IClientConfig clientConfig) {
        super.initWithNiwsConfig(clientConfig);
        serverWeightTaskTimerInterval = clientConfig.get(WEIGHT_TASK_TIMER_INTERVAL_CONFIG_KEY, DEFAULT_TIMER_INTERVAL);
    }
 
}