微服务架构之均衡组件 Ribbon解析：负载均衡策略实现

负载均衡策略实现

IRule是定义Ribbon负载均衡策略的接口，你可以通过实现该接口来自定义自己的负载均衡策略。该接口的默认Bean实例在RibbonClientConfiguration中给出。IRule接口的choose函数就是从一堆服务器中根据一定规则选出一个服务器。IRule有很多默认的实现类，这些实现类根据不同的算法和逻辑来处理负载均衡。Ribbon实现的IRule有一下。在大多数情况下，这些默认的实现类是可以满足需求的，如果有特性的需求，可以自己实现。

如图所示，Ribbon内置的IRule的子类有：

• BestAvailableRule 选择最小请求数
• ClientConfigEnabledRoundRobinRule 轮询RandomRule 随机选择一个server
• RoundRobinRule 轮询选择server
• RetryRule 根据轮询的方式重试
• WeightedResponseTimeRule 根据响应时间去分配一个weight ，weight越低，被选择的可能性就越低
• ZoneAvoidanceRule 根据server的zone区域和可用性来轮询选择

RoundRobinRule、RandomRule这样的不依赖于Server运行状况的策略，也有AvailabilityFilteringRule、WeightedResponseTimeRule等多种基于收集到的Server运行状况决策的策略。判断运行状况时有，判断单个server的，也有判断整个zone的，适用于各种不同场景需求。

@Bean
@ConditionalOnMissingBean
public IRule ribbonRule(IClientConfig config) {
    //如果在配置中设置了Rule就返回，否则使用默认的zoneAvoidanceRle
    if (this.propertiesFactory.isSet(IRule.class, name)) {
        return this.propertiesFactory.get(IRule.class, config, name);
    }
    ZoneAvoidanceRule rule = new ZoneAvoidanceRule();
    rule.initWithNiwsConfig(config);
    return rule;
}

ZoneAvoidanceRule是使用CompositePredicate来基于zone和可用性来过滤服务器。ZoneAvoidanceRule相关的类图如图所示。

ZoneAvoidanceRule是Ribbon默认的IRule实例，较为复杂，在本小节的后续部分再进行讲解。而ClientConfigEnabledRoundRobinRule是比较常用的IRule的子类之一，它使用的负载均衡策略是最为常见的Round Robin策略，也就是简单轮询策略。

public class ClientConfigEnabledRoundRobinRule extends AbstractLoadBalancerRule {

    RoundRobinRule roundRobinRule = new RoundRobinRule();
    @Override
    public Server choose(Object key) {
        if (roundRobinRule != null) {
            return roundRobinRule.choose(key);
        } else {
            throw new IllegalArgumentException(
                    "This class has not been initialized with the RoundRobinRule class");
        }
    }
}

RoundRobinRule会以轮询的方式依次将请求调度不同的服务器，即每次调度执行i = (i + 1) mod n，并选出第i台服务器。

public class RoundRobinRule extends AbstractLoadBalancerRule {

    private AtomicInteger nextServerCyclicCounter;
    private static final boolean AVAILABLE_ONLY_SERVERS = true;
    private static final boolean ALL_SERVERS = false;

    public RoundRobinRule() {
        nextServerCyclicCounter = new AtomicInteger(0);
    }

    public RoundRobinRule(ILoadBalancer lb) {
        this();
        setLoadBalancer(lb);
    }

    public Server choose(ILoadBalancer lb, Object key) {
        Server server = null;
        int count = 0;
        //当server没有被选出，或者循环了10次以下时进入执行语句块
        while (server == null && count++ < 10) {
            //获得当前可以到达的所有Server列表，会使用到IPing
            List<Server> reachableServers = lb.getReachableServers();
            //获得所有Server列表
            List<Server> allServers = lb.getAllServers();
            int upCount = reachableServers.size();
            int serverCount = allServers.size();
            //如果当前可用的server数量为0或者所有server数量为0，则直接返回null
            if ((upCount == 0) || (serverCount == 0)) {
                log.warn("No up servers available from load balancer: " + lb);
                return null;
            }
            //根据server总数来计算下一个执行请求的server的index
            int nextServerIndex = incrementAndGetModulo(serverCount);
            //从server列表中找出该server
            server = allServers.get(nextServerIndex);
            //如果为null，继续执行
            if (server == null) {
                Thread.yield();
                continue;
            }
            //判断server状态。
            if (server.isAlive() && (server.isReadyToServe())) {
                return (server);
            }

            server = null;
        }
        return server;
    }

    private int incrementAndGetModulo(int modulo) {
        for (;;) {
            //这就是i = (i + 1) mod n .因为会涉及多线程，所以这里使用CAS方法，进行for循环无限重试加上AtomicInteger的原子自增
            int current = nextServerCyclicCounter.get();
            int next = (current + 1) % modulo;
            if (nextServerCyclicCounter.compareAndSet(current, next))
                return next;
        }
    }

    @Override
    public Server choose(Object key) {
        return choose(getLoadBalancer(), key);
    }
}

ZoneAvoidanceRule则是比较复杂的负载均衡策略。ZoneAvoidanceRule是根据server的zone区域和可用性来进行负载均衡。PredicateBasedRule是ZoneAvoidanceRule的基类，它选择服务器的策略是先使用ILoadBalancer获取服务器列表，再使用AbstractServerPredicate来进行服务器的过滤,并且对剩下的服务器使用round robin策略选出最终的服务器。

public abstract class PredicateBasedRule extends ClientConfigEnabledRoundRobinRule {

    public abstract AbstractServerPredicate getPredicate();

    @Override
    public Server choose(Object key) {
        ILoadBalancer lb = getLoadBalancer();
        Optional<Server> server = getPredicate().chooseRoundRobinAfterFiltering(lb.getServerList(false), key);
        if (server.isPresent()) {
            return server.get();
        } else {
            return null;
        }       
    }
}

PredicateBasedRule提供了getPredicate函数供子类实现，来让不同的子类提供不同的AbstractServerPredicate实例来进行不同的服务器过滤策略。而ZoneAvoidanceRule使用的是由ZoneAvoidancePredicate和AvailabilityPredicate组成的复合策略CompositePredicate，前一个判断判定一个zone的运行性能是否可用，剔除不可用的zone（的所有server），AvailabilityPredicate用于过滤掉连接数过多的Server。

private CompositePredicate compositePredicate;

public ZoneAvoidanceRule() {
    super();
    ZoneAvoidancePredicate zonePredicate = new ZoneAvoidancePredicate(this);
    AvailabilityPredicate availabilityPredicate = new AvailabilityPredicate(this);
    compositePredicate = createCompositePredicate(zonePredicate, availabilityPredicate);
}

//将两个Predicate组合成一个CompositePredicate
private CompositePredicate createCompositePredicate(ZoneAvoidancePredicate p1, AvailabilityPredicate p2) {
    return CompositePredicate.withPredicates(p1, p2)
                        .addFallbackPredicate(p2)
                        .addFallbackPredicate(AbstractServerPredicate.alwaysTrue())
                        .build();

}
//实现`getPredicate`接口，返回构造函数中生成的compositePredicate
@Override
public AbstractServerPredicate getPredicate() {
    return compositePredicate;
}    

CompositePredicate的chooseRoundRobinAfterFiltering函数是继承基类AbstractServerPredicate的实现。它会调用首先调用getEligibleServers函数来通过Predicate过滤服务器列表，然后使用round robin策略选择出一个服务器进行返回。

//AbstractServerPredicate.java
//先用redicate来获取一个可用server的集合，然后用round robin算法来选择一个。
public Optional<Server> chooseRoundRobinAfterFiltering(List<Server> servers) {
    List<Server> eligible = getEligibleServers(servers);
    if (eligible.size() == 0) {
        return Optional.absent();
    }
    // (i + 1) % n
    return Optional.of(eligible.get(nextIndex.getAndIncrement() % eligible.size()));
}

当loadBalancerKey为null时，getEligibleServers函数会使用serverOnlyPredicate来依次过滤服务器列表。

public List<Server> getEligibleServers(List<Server> servers, Object loadBalancerKey) {
    if (loadBalancerKey == null) {
        return ImmutableList.copyOf(Iterables.filter(servers, this.getServerOnlyPredicate()));            
    } else {
        //遍历servers，都调用对应predicate的apply函数来判断该serve是否可用。
        List<Server> results = Lists.newArrayList();
        for (Server server: servers) {
            if (this.apply(new PredicateKey(loadBalancerKey, server))) {
                results.add(server);
            }
        }
        return results;            
    }
}

serverOnlyPredicate则会调用其apply函数，并将Server对象封装成PredicateKey当作参数传入。AbstractServerPredicate并没有实现apply函数，由它的子类来实现从而达到不同子类实现不同过滤策略的目的。

private final Predicate<Server> serverOnlyPredicate =  new Predicate<Server>() {
        @Override
        public boolean apply(@Nullable Server input) {                    
            return AbstractServerPredicate.this.apply(new PredicateKey(input));
        }
    };

ZoneAvoidanceRule类中CompositePredicate对象的apply函数就会依次调用ZoneAvoidancePredicate和AvailabilityPredicate的apply函数。

ZoneAvoidancePredicate以一个区域(也就是一个Zone)为单位考察可用性，对于不可用的区域整个丢弃，从剩下区域中选可用的server。判断出最差的区域，排除掉最差区域。在剩下的区域中，将按照服务器实例数的概率抽样法选择，从而判断判定一个zone的运行性能是否可用，剔除不可用的zone中的所有server。

//ZoneAvoidancePredicate.java
@Override
public boolean apply(@Nullable PredicateKey input) {
    if (!ENABLED.get()) {
        return true;
    }
    String serverZone = input.getServer().getZone();
    if (serverZone == null) {
        //如果sever没有zone相关的信息，则直接返回
        return true;
    }
    //LoadBalancerStats存储着每个server或者node的执行特征和运行记录。这些信息可用供动态负载均衡策略使用。
    LoadBalancerStats lbStats = getLBStats();
    if (lbStats == null) {
        //如果没有则直接返回
        return true;
    }
    if (lbStats.getAvailableZones().size() <= 1) {
        // 如果只有一个zone，那么也是直接返回。
        return true;
    }
    //为了效率，先看一下lbStats中记录的zone列表是否包含当前这个zone
    Map<String, ZoneSnapshot> zoneSnapshot = ZoneAvoidanceRule.createSnapshot(lbStats);
    if (!zoneSnapshot.keySet().contains(serverZone)) {
        // 如果该serverZone不存在，那么也直接返回
        return true;
    }
    //调用ZoneAvoidanceRule的getAvailableZone方法来获取可用的zone列表
    Set<String> availableZones = ZoneAvoidanceRule.getAvailableZones(zoneSnapshot, triggeringLoad.get(), triggeringBlackoutPercentage.get());
    //判断当前zone是否在可用zone列表中
    if (availableZones != null) {
        return availableZones.contains(input.getServer().getZone());
    } else {
        return false;
    }
}    

在继续学习之前，希望读者要先了解一下Zone相关的基本概念。Zone是多个服务实例的集合，Zone之间的服务实例访问会有更大的延迟，Zone之内的服务实例访问延迟较小。ZoneSnapshot存储了关于Zone的一些性能参数，比如说实例数量、断路器断开数、活动请求数、实例平均负载。

public class ZoneSnapshot {
    final int instanceCount; //实例数
    final double loadPerServer; //平均负载
    final int circuitTrippedCount; //断路器断开数量
    final int activeRequestsCount; //活动请求数量
}

ZoneAvoidanceRule的createSnapshot其实就是将所有的Zone列表转化为以其名称为键值的Map表，供ZoneAvoidancePredicate的apply函数使用。

//ZoneAvoidanceRule中的静态方法
//将LoadbalancerStats中的availableZones列表转换为Map再返回。
static Map<String, ZoneSnapshot> createSnapshot(LoadBalancerStats lbStats) {
        Map<String, ZoneSnapshot> map = new HashMap<String, ZoneSnapshot>();
    for (String zone : lbStats.getAvailableZones()) {
        ZoneSnapshot snapshot = lbStats.getZoneSnapshot(zone);
        map.put(zone, snapshot);
    }
    return map;
}

getAvailableZones函数是用来筛选Zone列表的，首先，它会遍历一遍Zone列表，在遍历的过程中，它会做两件事情，它会依据Zone内的实例数，实例的平均负载时间，实例故障率等指标将Zone从列表中删除；它会维护一个最坏Zone列表，当某个Zone的平均负载时间小于但是接近全局最坏负载时间时，就会将该Zone加入到最坏Zone列表，如果某个Zone的平均负载时间大于最坏负载时间时，它将会清楚掉之前的最坏Zone列表，以该Zone的平均负载时间为全局最坏负载时间，继续最坏Zone列表的构建。在函数最后，如果全局最坏负载数据大于系统设定的负载时间，则在最坏Zone列表中随机选择出一个Zone，将其从Zone列表中删除。

public static Set<String> getAvailableZones(
        Map<String, ZoneSnapshot> snapshot, double triggeringLoad,
        double triggeringBlackoutPercentage) {
    if (snapshot.isEmpty()) {
        return null;
    }
    Set<String> availableZones = new HashSet<String>(snapshot.keySet());
    if (availableZones.size() == 1) {
        return availableZones;
    }

    Set<String> worstZones = new HashSet<String>();
    double maxLoadPerServer = 0;
    boolean limitedZoneAvailability = false;
    //遍历所有的zone来判定
    for (Map.Entry<String, ZoneSnapshot> zoneEntry : snapshot.entrySet()) {
        String zone = zoneEntry.getKey();
        ZoneSnapshot zoneSnapshot = zoneEntry.getValue();
        //判定该zone中的服务实例数
        int instanceCount = zoneSnapshot.getInstanceCount();
        if (instanceCount == 0) {
            //如果zone中没有服务实例，那么去除掉该zone
            availableZones.remove(zone);
            limitedZoneAvailability = true;
        } else {
            double loadPerServer = zoneSnapshot.getLoadPerServer();
            //Zone区域内实例平均负载小于零，或者实例故障率（断路器断开次数/实例数）大于等于阈值（默认为0.99999，则去除掉该zone
            if (((double) zoneSnapshot.getCircuitTrippedCount())
                    / instanceCount >= triggeringBlackoutPercentage
                    || loadPerServer < 0) {
                availableZones.remove(zone);
                limitedZoneAvailability = true;
            } else {
                //如果该zone的平均负载和最大负载的差距小于一定量，则将该zone加入到最坏zone集合中
                if (Math.abs(loadPerServer - maxLoadPerServer) < 0.000001d) {
                    worstZones.add(zone);
                } else if (loadPerServer > maxLoadPerServer) {
                    //否则，如果该zone平均负载还大于最大负载。
                    maxLoadPerServer = loadPerServer;
                    //清楚掉最坏zone集合，将该zone加入
                    worstZones.clear();
                    worstZones.add(zone);
                }
            }

        }
    }
    //如果最大平均负载小于设定的阈值直接返回
    if (maxLoadPerServer < triggeringLoad && !limitedZoneAvailability) {
        // zone override is not needed here
        return availableZones;
    }
    //如果大于，则从最坏zone集合中随机剔除一个。
    String zoneToAvoid = randomChooseZone(snapshot, worstZones);
    if (zoneToAvoid != null) {
        availableZones.remove(zoneToAvoid);
    }
    return availableZones;
}

ZoneAvoidancePredicate的apply调用结束之后，AvailabilityPredicate的apply也会被调用。它是依据掉断路器是否断开或者服务器链接数是否超出阈值来进行服务过滤的预测规则。

//AvailabilityPredicate.java
@Override
public boolean apply(@Nullable PredicateKey input) {
    LoadBalancerStats stats = getLBStats();
    if (stats == null) {
        return true;
    }
    //获得关于该server的信息记录
    return !shouldSkipServer(stats.getSingleServerStat(input.getServer()));
}


private boolean shouldSkipServer(ServerStats stats) {   
    //如果该server的断路器已经打开，或者它的链接数大于预设的阈值，那么就需要将该server过滤掉
    if ((CIRCUIT_BREAKER_FILTERING.get() && stats.isCircuitBreakerTripped())
            || stats.getActiveRequestsCount() >= activeConnectionsLimit.get()) {
        return true;
    }
    return false;
}

本小节主要讲解了ClientConfigEnabledRoundRobinRule和ZoneAvoidanceRule这两个负载均衡策略，其他策略的具体实现，有兴趣的读者可以自己去了解，就不在这里进行一一讲解了。