Ribbon之IRule

IRule是選擇服務(wù)的一種策略。

• IRule

public interface IRule{
    /*
     * choose one alive server from lb.allServers or
     * lb.upServers according to key
     * 
     * @return choosen Server object. NULL is returned if none
     *  server is available 
     */
    public Server choose(Object key);
    
    public void setLoadBalancer(ILoadBalancer lb);
    
    public ILoadBalancer getLoadBalancer();    
}

choose選擇可用的服務(wù)。

• RandomRule
  

隨機選擇一個UP的服務(wù)。

Random rand; // 隨機計數(shù)器

public RandomRule() {
    rand = new Random();
}

public Server choose(ILoadBalancer lb, Object key) {
    ...
    
    List<Server> upList = lb.getReachableServers();
    List<Server> allList = lb.getAllServers();
    
    int index = rand.nextInt(serverCount); // 隨機選擇一個
    server = upList.get(index);
    
    ...

}

• RoundRobinRule
  

輪詢獲取服務(wù)。

public RoundRobinRule() {
    nextServerCyclicCounter = new AtomicInteger(0);// int線程安全計數(shù)器
}

public Server choose(ILoadBalancer lb, Object key) {
   
   ...
     
   int nextServerIndex = incrementAndGetModulo(serverCount); // nextServerCyclicCounter依次向后獲取服務(wù)。 
   server = allServers.get(nextServerIndex);
   
   ...

}

// 輪詢方法
private int incrementAndGetModulo(int modulo) {
    for (;;) {
        int current = nextServerCyclicCounter.get();
        int next = (current + 1) % modulo;
        if (nextServerCyclicCounter.compareAndSet(current, next))
            return next;
    }
}

• BestAvailableRule

跳過熔斷的服務(wù),獲取請求數(shù)最少的服務(wù).通常與ServerListSubsetFilter一起使用.

public Server choose(Object key) {
    if (loadBalancerStats == null) {
        return super.choose(key); // 如果沒有l(wèi)oadBalancerStats,則采用RoundRibonRule.
    }
    List<Server> serverList = getLoadBalancer().getAllServers();
    int minimalConcurrentConnections = Integer.MAX_VALUE;
    long currentTime = System.currentTimeMillis();
    Server chosen = null;
    for (Server server: serverList) {
        ServerStats serverStats = loadBalancerStats.getSingleServerStat(server);
        if (!serverStats.isCircuitBreakerTripped(currentTime)) {
            int concurrentConnections = serverStats.getActiveRequestsCount(currentTime);
            if (concurrentConnections < minimalConcurrentConnections) {
                minimalConcurrentConnections = concurrentConnections;
                chosen = server;
            }
        }
    }
    if (chosen == null) {
        return super.choose(key);
    } else {
        return chosen;
    }
}

• WeightedResponseTimeRule
  

權(quán)重的方式挑選服務(wù).服務(wù)實例響應(yīng)時間越小的服務(wù),則更容易被選中.如果服務(wù)實例響應(yīng)的時間相差不大的,排在前面的服務(wù)實例更容易被選中.

// 繼承了RoundRobinRule,也就是當(dāng)WeightedResponseTimeRule不滿足條件的時候,則采用RoundRobinRule的方式.
public class WeightedResponseTimeRule extends RoundRobinRule {

// 這個方式很重要,就是定時的計算每個服務(wù)實例的響應(yīng)時間,并以此作為每個服務(wù)實例的權(quán)重.
void initialize(ILoadBalancer lb) {        
    if (serverWeightTimer != null) {
        serverWeightTimer.cancel();
    }
    serverWeightTimer = new Timer("NFLoadBalancer-serverWeightTimer-" + name, true);
    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();
        }
    }));
}

// 定時任務(wù)內(nèi)部類
class DynamicServerWeightTask extends TimerTask {
    public void run() {
        ServerWeight serverWeight = new ServerWeight();
        try {
            serverWeight.maintainWeights();
        } catch (Exception e) {
            logger.error("Error running DynamicServerWeightTask for {}", name, e);
        }
    }
}

// 計算服務(wù)實例權(quán)重的核心方法.
class ServerWeight {

    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;
            // find maximal 95% response time
            for (Server server : nlb.getAllServers()) {
                // this will automatically load the stats if not in cache
                ServerStats ss = stats.getSingleServerStat(server);
                totalResponseTime += ss.getResponseTimeAvg();
            }
            // weight for each server is (sum of responseTime of all servers - responseTime)
            // so that the longer the response time, the less the weight and the less likely to be chosen
            Double weightSoFar = 0.0;
            
            // create new list and hot swap the reference
            List<Double> finalWeights = new ArrayList<Double>();
            for (Server server : nlb.getAllServers()) {
                ServerStats ss = stats.getSingleServerStat(server);
                double weight = totalResponseTime - ss.getResponseTimeAvg(); // 平均響應(yīng)時間越短,則權(quán)重越大,就越容易被選中.
                weightSoFar += weight;
                finalWeights.add(weightSoFar);   
            }
            setWeights(finalWeights);
        } catch (Exception e) {
            logger.error("Error calculating server weights", e);
        } finally {
            serverWeightAssignmentInProgress.set(false);
        }

    }
}

public Server choose(ILoadBalancer lb, Object key) {

     ...
     
    // 根據(jù)權(quán)重選擇服務(wù)的核心代碼 
    double randomWeight = random.nextDouble() * maxTotalWeight;
    // pick the server index based on the randomIndex
    int n = 0;
    for (Double d : currentWeights) {
        if (d >= randomWeight) {
            serverIndex = n;
            break;
        } else {
            n++;
        }
    }
    
    server = allList.get(serverIndex);
    
    ...
}


}

• RetryRule
  

在RoundRobinRule的基礎(chǔ)上,增加了重試的機制.

• ZoneAvoidanceRule