Ribbon之IRule

IRule是选择服务的一种策略。

• 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选择可用的服务。

• RandomRule

随机选择一个UP的服务。

Random rand; // 随机计数器

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

轮询获取服务。

public RoundRobinRule() {
    nextServerCyclicCounter = new AtomicInteger(0);// int线程安全计数器
}

public Server choose(ILoadBalancer lb, Object key) {
   
   ...
     
   int nextServerIndex = incrementAndGetModulo(serverCount); // nextServerCyclicCounter依次向后获取服务。 
   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

跳过熔断的服务,获取请求数最少的服务.通常与ServerListSubsetFilter一起使用.

public Server choose(Object key) {
    if (loadBalancerStats == null) {
        return super.choose(key); // 如果没有loadBalancerStats,则采用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

权重的方式挑选服务.服务实例响应时间越小的服务,则更容易被选中.如果服务实例响应的时间相差不大的,排在前面的服务实例更容易被选中.

// 继承了RoundRobinRule,也就是当WeightedResponseTimeRule不满足条件的时候,则采用RoundRobinRule的方式.
public class WeightedResponseTimeRule extends RoundRobinRule {
// 这个方式很重要,就是定时的计算每个服务实例的响应时间,并以此作为每个服务实例的权重.
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();
        }
    }));
}
// 定时任务内部类
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);
        }
    }
}
// 计算服务实例权重的核心方法.
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(); // 平均响应时间越短,则权重越大,就越容易被选中.
                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) {
     ...
     
    // 根据权重选择服务的核心代码 
    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的基础上,增加了重试的机制.

• ZoneAvoidanceRule

使用ZoneAvoidancePredicate和AvailabilityPredicate来判断是否选择某个server，前一个判断判定一个zone的运行性能是否可用，剔除不可用的zone（的所有server），AvailabilityPredicate用于过滤掉连接数过多的Server。