Netflix Ribbon 可用过滤策略 AvailabilityFilteringRule

可用过滤策略的主要作用是过滤掉那些因为连接失败、断路器跳闸等原因而处于不可用状态的服务实例，然后从剩余的可用服务实例中选择一个进行请求分发。

它会结合服务实例的连接状态和断路器状态来进行判断。如果一个服务实例的连接失败次数超过一定阈值或者断路器处于跳闸状态，那么这个实例就会被认为是不可用的，在进行负载均衡选择时会被排除在外。

可用过滤策略特点

动态过滤：能够根据服务实例的实时状态进行动态过滤，确保只选择可用的服务实例进行请求分发。
结合断路器：与断路器机制紧密结合，当断路器跳闸时，对应的服务实例会被快速排除，提高系统的稳定性和可靠性。

配置方式

在 application.properties 中，使用如下配置：

user.ribbon.NFLoadBalancerRuleClassName=com.netflix.loadbalancer.AvailabilityFilteringRule

将在名为 user 的客户端上开启可用过滤策略。

源码分析

316b61205e9e279b5d723082b1b84a7a_1729136949218-3cabc7bd-b0f7-45ba-8bfd-a9f253d67d90.png

Ribbon 可用过滤策略通过 com.netflix.loadbalancer.AvailabilityFilteringRule 类实现，代码如下：

public class AvailabilityFilteringRule extends PredicateBasedRule {
    // 一个复合谓词/断言，使用了 Guava 中的 Predicate
    // Guava 中的 Predicate 是一个用于判断给定对象是否满足特定条件的接口。
    // 下面 predicate 成员由 AvailabilityPredicate 和一个总是返回 true 的
    // 备用谓词（AbstractServerPredicate.alwaysTrue()）组成
    private AbstractServerPredicate predicate = CompositePredicate.withPredicate(
        new AvailabilityPredicate(this, (IClientConfig)null)).
            addFallbackPredicate(AbstractServerPredicate.alwaysTrue()).build();

    public AvailabilityFilteringRule() {
    }

    // 用于在运行时根据新的客户端配置重新初始化谓词。它使用新的客户端配置创建一个
    // 新的AvailabilityPredicate，并构建一个新的复合谓词来更新predicate成员变量。
    // 这使得负载均衡规则可以根据不同的客户端配置动态调整选择服务实例的策略。
    public void initWithNiwsConfig(IClientConfig clientConfig) {
        this.predicate = CompositePredicate.withPredicate(
            new AvailabilityPredicate(this, clientConfig))
                .addFallbackPredicate(AbstractServerPredicate.alwaysTrue()).build();
    }

    // 用于获取可用服务实例的数量
    @Monitor(
        name = "AvailableServersCount",
        type = DataSourceType.GAUGE
    )
    public int getAvailableServersCount() {
        ILoadBalancer lb = this.getLoadBalancer();
        List<Server> servers = lb.getAllServers();
        return servers == null ? 0 : Collections2.filter(servers, this.predicate.getServerOnlyPredicate()).size();
    }

    // 负载均衡规则的核心方法，用于选择一个服务实例
    public Server choose(Object key) {
        // 计数器
        int count = 0;
        // 先使用轮询策略选择一个服务器
        for(Server server = this.roundRobinRule.choose(key); count++ <= 10; 
            server = this.roundRobinRule.choose(key)) {
            // 检查当前选择的服务实例是否满足谓词
            // 如果满足谓词，则返回该服务实例；否则，继续使用轮询规则选择下一个服务实例。
            if (this.predicate.apply(new PredicateKey(server))) {
                return server;
            }
        }

        // 如果循环结束后仍然没有找到满足谓词的服务实例，则调用父类的choose方法
        return super.choose(key);
    }

    public AbstractServerPredicate getPredicate() {
        return this.predicate;
    }
}

父类 PredicateBasedRule 的 choose 方法：

public Server choose(Object key) {
    // 获取负载均衡器实例，负载均衡器负责管理一组服务实例，并提供选择服务实例的方法
    ILoadBalancer lb = this.getLoadBalancer();
    // 获取谓词，然后通过获取可用服务器，以轮询策略获取可用服务
    Optional<Server> server = this.getPredicate().chooseRoundRobinAfterFiltering(lb.getAllServers(), key);
    return server.isPresent() ? (Server)server.get() : null;
}

查看 AbstractServerPredicate 抽象类的 chooseRoundRobinAfterFiltering 方法：

public Optional<Server> chooseRoundRobinAfterFiltering(List<Server> servers, Object loadBalancerKey) {
    // 获取可用的服务列表
    List<Server> eligible = this.getEligibleServers(servers, loadBalancerKey);
    // 如果可用服务器列表为空，返回值不存在 Optional
    // 如果有，则从中选择一个
    return eligible.size() == 0 ? Optional.absent() : 
        Optional.of(eligible.get(this.incrementAndGetModulo(eligible.size())));
}

// 实现了一个原子性的自增操作并取模的功能。它确保在多线程环境下安全地对一个
// 整数值进行自增操作，同时保证结果在给定的模数值范围内循环
private final AtomicInteger nextIndex = new AtomicInteger();
private int incrementAndGetModulo(int modulo) {
    int current;
    int next;
    do {
        current = this.nextIndex.get();
        next = (current + 1) % modulo;
    } while(!this.nextIndex.compareAndSet(current, next) || current >= modulo);

    return current;
}

下面给出一个简单关于如何使用谓词的例子，你可以将谓词当作一个条件判断。假如我们使用谓词来过滤列表中年龄大于等于 30 的人：

import com.google.common.base.Predicate;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import java.util.List;

public class PredicateDemo {

    public static void main(String[] args) {
        List<Person> people = Lists.newArrayList(
                new Person("Alice", 25),
                new Person("Bob", 30),
                new Person("Charlie", 18),
                new Person("David", 40)
        );

        // 定义一个 Predicate 判断年龄大于等于 30 的人
        Predicate<Person> agePredicate = new Predicate<Person>() {
            @Override
            public boolean apply(Person input) {
                return input.getAge() >= 30;
            }
        };

        // 使用 Iterables.filter 过滤出满足条件的人
        Iterable<Person> filteredPeople = Iterables.filter(people, agePredicate);
        for (Person person : filteredPeople) {
            System.out.println(person.getName() + " is " + person.getAge() + " years old.");
        }
    }

    // 定义一个 Person POJO类
    static class Person {
        private String name;
        private int age;

        public Person(String name, int age) {
            this.name = name;
            this.age = age;
        }

        public String getName() {
            return name;
        }

        public int getAge() {
            return age;
        }
    }
}