[享学Netflix] 四十九、Ribbon的LoadBalancer五大组件之：ServerListFilter服务列表过滤器_飞道的博客

[享学Netflix] 四十九、Ribbon的LoadBalancer五大组件之：ServerListFilter服务列表过滤器

2020-04-01 15:16 344人阅读评论(0)

靠代码行数来衡量开发进度，就像是凭重量来衡量飞机制造的进度------比尔·盖茨

–> 返回专栏总目录 <–
代码下载地址：https://github.com/f641385712/netflix-learning

前言

前面已经介绍了ServerList组件它用于提供服务列表，本文介绍它另外一个重要的组件：ServerListFilter服务列表过滤器。

服务的过滤对负载均衡是非常有意义的，因为在运行过程中，并不是没台Server一直都持续可用，另外多台Server很有可能分部在不同的可用区zone，而很多时候我们希望是获取到同区域的机器以加速访问，这些都是交由由ServerListFilter来完成的。

正文

上文有讲到服务列表的断言器：AbstractServerPredicate。本文的ServerListFilter服务列表过滤器有些便会基于它进行实现，特别是基于Zone区域的过滤逻辑，复用现成的即可。

ServerListFilter

该接口用于过滤Server列表们，接口描述很简单，难的是过滤规则。

public interface ServerListFilter<T extends Server> {
	// 返回的是一个过滤后的列表
	// 可能是原列表，也可能是新的列表~~~~~~~
    public List<T> getFilteredListOfServers(List<T> servers);
}

它的继承图谱如下（Spring Cloud环境下新增了一个）：

AbstractServerListFilter

从负载均衡器LoadBalancer的服务器列表里面筛选出可用的Server，它无非就是规定了Server来源：来自于负载均衡器LB，这种可用/不可用是通过指标收集库/存储库LoadBalancerStats计算出来的。

public abstract class AbstractServerListFilter<T extends Server> implements ServerListFilter<T> {

	private volatile LoadBalancerStats stats;
    public void setLoadBalancerStats(LoadBalancerStats stats) {
        this.stats = stats;
    } 
    public LoadBalancerStats getLoadBalancerStats() {
        return stats;
    }
}

该抽象类仅指明了，服务列表过滤的指标参考来自于LoadBalancerStats，下面看具体实现类如何使用。

ZoneAffinityServerListFilter

它借助于ZoneAffinityPredicate来过滤出和zone相关的服务器，即：只留下指定zone下的Server们。

成员属性

// 小细节：它的泛型也没确定
public class ZoneAffinityServerListFilter<T extends Server> extends AbstractServerListFilter<T> implements IClientConfigAware {

    private volatile boolean zoneAffinity = DefaultClientConfigImpl.DEFAULT_ENABLE_ZONE_AFFINITY;
    private volatile boolean zoneExclusive = DefaultClientConfigImpl.DEFAULT_ENABLE_ZONE_EXCLUSIVITY;
    private DynamicDoubleProperty activeReqeustsPerServerThreshold;
    private DynamicDoubleProperty blackOutServerPercentageThreshold;
    private DynamicIntProperty availableServersThreshold;
    private Counter overrideCounter;
    private ZoneAffinityPredicate zoneAffinityPredicate = new ZoneAffinityPredicate();
    
	String zone;

	// 构造器通过initWithNiwsConfig为成员属性赋值~~~~
    public ZoneAffinityServerListFilter() {      
    }
    public ZoneAffinityServerListFilter(IClientConfig niwsClientConfig) {
        initWithNiwsConfig(niwsClientConfig);
    }

}

zoneAffinity：控制是否要开启ZoneAffinity的开关，默认是false
- 可以通过EnableZoneAffinity来配置。也就是xxx.ribbon.EnableZoneAffinity或者全局默认ribbon.EnableZoneAffinity
zoneExclusive：同样是可以控制是否要开启ZoneAffinity的开关。同时它在Filter过滤Server的时候还起到开关的通，默认是false
- 可以通过EnableZoneExclusivity这个key进行配置（全局or定制）
activeReqeustsPerServerThreshold：最大负载阈值，默认值是0.6d
- 可通过<clientName>.ribbon.zoneAffinity.maxLoadPerServer = xxx来配置
- 对比ZoneAvoidancePredicate里面的ZoneAwareNIWSDiscoveryLoadBalancer.triggeringLoadPerServerThreshold属性你会发现默认值是及其不合理的，后面会用例子解释
blackOutServerPercentageThreshold：默认值0.8d
- 可通过<clientName>.ribbon.zoneAffinity.maxBlackOutServesrPercentage= xxx来配置
- 同样的你对标ZoneAvoidancePredicate的triggeringBlackoutPercentage属性值吧
availableServersThreshold：可用Server的阈值，默认值是2。
- 可通过<clientName>.ribbon.zoneAffinity.minAvailableServers = xxx来配置
overrideCounter：servo里用于统计的，略。
zoneAffinityPredicate：断言器，用于筛选出配置的指定zone的server们。这在上篇文章中重点阐述过
zone：当前的zone。来自于配置的上下文：ConfigurationManager.getDeploymentContext().getValue(ContextKey.zone)

过滤逻辑

过滤逻辑中，最重要的乃shouldEnableZoneAffinity()这个方法：

ZoneAffinityServerListFilter：

	// 是否开启根据zone进行过滤
	// 说明：filtered是已经经过zone过滤后，肯定是同一个zone里面的server们了
	private boolean shouldEnableZoneAffinity(List<T> filtered) {
		// 如果zoneAffinity=false 并且 zoneExclusive = false才表示不开启zone过滤
		// 默认两个都是false哦
        if (!zoneAffinity && !zoneExclusive) {
            return false;
        }
        // 若显示开启zone排除，那就直接返回true
        // 否则会计算，根据负载情况动态判断
        if (zoneExclusive) {
            return true;
        }
		
		LoadBalancerStats stats = getLoadBalancerStats();
        if (stats == null) {
            return zoneAffinity;
        } else {
			// 拿到zone的快照，从而拿到zone的实例总数、负载、熔断总数等
			ZoneSnapshot snapshot = stats.getZoneSnapshot(filtered);
            double loadPerServer = snapshot.getLoadPerServer();
            int instanceCount = snapshot.getInstanceCount();    
            int circuitBreakerTrippedCount = snapshot.getCircuitTrippedCount();

			// 开始判断
            if (((double) circuitBreakerTrippedCount) / instanceCount >= blackOutServerPercentageThreshold.get() 
                    || loadPerServer >= activeReqeustsPerServerThreshold.get()
                    || (instanceCount - circuitBreakerTrippedCount) < availableServersThreshold.get()) {
            	return false;
			} else {
				return true;
			}
        }
	}

将此方法的执行步骤使用文字版总结如下：

若你配置了zoneAffinity或者zoneExclusive任何一个为true，则将开启此筛选逻辑
1. 若你是zoneExclusive=true，说明你同意这种排除逻辑，那就直接生效开启返回true喽
2. 否则，进入根据动态指标的计算逻辑
下面复杂的逻辑计算，有如下情况均会返回false(不执行过滤，而是返回全部Server)：
1. circuitBreakerTrippedCount/instanceCount >= blackOutServerPercentageThreshold，也就是说呗熔断的占比率超过0.8，也就是80%的机器都被熔断了，那就返回false（毕竟此zone已基本不可用了，那还是返回所有Server保险点）
2. loadPerServer >= activeReqeustsPerServerThreshold，若平均负载超过0.6，那就返回fasle（因为没必要把负载过高的zone返回出去，还是返回所有Server较好）
3. (instanceCount - circuitBreakerTrippedCount) < availableServersThreshold，如果“活着的（没熔断的）”实例总数不足2个（仅有1个了），那就返回false
若以上三种情况均没发生，那就返回true

该方法返回值释义：

true：最终只留下本zone的Server们
false，返回所有Server，相当于忽略此Filter的操作

这么做的目的是：担心你配置的zone里面的Server情况不乐观，如果这个时候只返回该zone的Server的话，反倒不好，还不如把所有Server都返回更为合适。下面就是它实现接口方法代码实现：

ZoneAffinityServerListFilter：

	// 只有你自己指定了，配置了zone
	// 并且显示的开启了过滤逻辑zoneAffinity/zoneExclusive任何一个为true
	// 并且servers不为空
    @Override
    public List<T> getFilteredListOfServers(List<T> servers) {
        if (zone != null && (zoneAffinity || zoneExclusive) && servers !=null && servers.size() > 0){
            List<T> filteredServers = Lists.newArrayList(Iterables.filter(servers, this.zoneAffinityPredicate.getServerOnlyPredicate()));
            if (shouldEnableZoneAffinity(filteredServers)) {
                return filteredServers;
            } else if (zoneAffinity) {
                overrideCounter.increment();
            }
        }
        return servers;
    }

说明：次数使用的过滤断言器是ZoneAffinityPredicate，关于它的详解你可以参考这篇文章：[享学Netflix] 四十八、Ribbon服务器过滤逻辑的基础组件：AbstractServerPredicate

ZoneAffinityServerListFilter它有两个实现类：一个是自己实现的ServerListSubsetFilter，一个是Spring Cloud实现的ZonePreferenceServerListFilter。

ServerListSubsetFilter

一种服务器列表筛选器实现。它将负载均衡器使用的Server数量限制为所有服务器的子集。比如通过父类筛选出了一个zone里的server如果非常多的话（比如上百、上千台），那么你是木有必要把这上千台全部返回出去的，自需要返回其一个子集即可。

因为全部返回出去，比如上千台，那么都需要保留其httpclient链接在连接池中，挺耗资源的

在server list非常多的场景下，没有必要在连接池的保持这么多的连接，ServerListSubsetFilter可以在这种场景下对server [list进行精简，通过剔除相对不健康(failureCount、activeRequestCount)的server来达到此目标。

ZonePreferenceServerListFilter

它是Spring Cloud默认使用的筛选器。它的特点是：能够优先过滤出与请求调用方处于同区域的服务实例。

public class ZonePreferenceServerListFilter extends ZoneAffinityServerListFilter<Server> {
	...
	@Override
	public List<Server> getFilteredListOfServers(List<Server> servers) {
		List<Server> output = super.getFilteredListOfServers(servers);
		
		// 若指定了zone，并且output.size() == servers.size()
		// 也就说父类没有根据zone进行过滤的话，那这里就会继续处理逻辑
		if (this.zone != null && output.size() == servers.size()) {
			List<Server> local = new ArrayList<>();
			// 只会选取和当前设置的zone一样的Server
			for (Server server : output) {
				if (this.zone.equalsIgnoreCase(server.getZone())) {
					local.add(server);
				}
			}

			// 哪怕只有一台Server都返回
			if (!local.isEmpty()) {
				return local;
			}
		}
		return output;
	}
	...
}

虽然它作为Spring Cloud的默认筛选器，但我认为它是父类筛选逻辑的阉割版。若父类没筛选出来，它简单的粗暴的仅根据zone进行选择，其实这么做是可能会有问题的：万一这台Server负载很高？万一熔断了呢？万一只有一个Server实例呢？？？

所以我个人觉得生产环境下默认使用它不算一个很好的方案，可以尝试自己定制。

默认配置不合理

还是负载的阈值问题，ZoneAffinityServerListFilter.activeReqeustsPerServerThreshold的默认值是0.6，显然是及其不合理的。具体原因前面有讲过，本处略。可参考：[享学Netflix] 四十七、Ribbon多区域选择：ZoneAvoidanceRule.getAvailableZones()获取可用区

代码示例

@Test
public void fun8() throws InterruptedException {
    String clientName = "YourBatman";
    // 负载均衡器状态信息   后面模拟请求来增加指标数据
    LoadBalancerStats lbs = new LoadBalancerStats(clientName);

    // 添加Server
    List<Server> serverList = new ArrayList<>();
    serverList.add(createServer("华南", 1));
    serverList.add(createServer("华东", 1));
    serverList.add(createServer("华东", 2));

    serverList.add(createServer("华北", 1));
    serverList.add(createServer("华北", 2));
    serverList.add(createServer("华北", 3));
    serverList.add(createServer("华北", 4));
    lbs.updateServerList(serverList);

    Map<String, List<Server>> zoneServerMap = new HashMap<>();
    // 模拟向每个Server发送请求  记录ServerStatus数据
    serverList.forEach(server -> {
        ServerStats serverStat = lbs.getSingleServerStat(server);
        request(serverStat);

        // 顺便按照zone分组
        String zone = server.getZone();
        if (zoneServerMap.containsKey(zone)) {
            zoneServerMap.get(zone).add(server);
        } else {
            List<Server> servers = new ArrayList<>();
            servers.add(server);
            zoneServerMap.put(zone, servers);
        }
    });
    lbs.updateZoneServerMapping(zoneServerMap);

    // 指定当前的zone
    DeploymentContext deploymentContext = ConfigurationManager.getDeploymentContext();
    deploymentContext.setValue(DeploymentContext.ContextKey.zone, "华北");
    // 准备一个服务列表过滤器
    IClientConfig config = DefaultClientConfigImpl.getClientConfigWithDefaultValues(clientName);
    config.set(CommonClientConfigKey.EnableZoneAffinity, true);
    ZoneAffinityServerListFilter serverListFilter = new ZoneAffinityServerListFilter();
    serverListFilter.setLoadBalancerStats(lbs);
    serverListFilter.initWithNiwsConfig(config);

    // 从lbs里拿到一些监控数据
    monitor(lbs, serverListFilter);

    TimeUnit.SECONDS.sleep(500);
}


private void monitor(LoadBalancerStats lbs, ZoneAffinityServerListFilter serverListFilter) {
    List<String> zones = Arrays.asList("华南", "华东", "华北");
    new Thread(() -> {
        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
        executorService.scheduleWithFixedDelay(() -> {
            // 打印当前可用区
            // 获取可用区
            Set<String> availableZones = ZoneAvoidanceRule.getAvailableZones(lbs, 0.2d, 0.99999d);
            System.out.println("=====当前可用区为：" + availableZones);

            List<Server> filteredListOfServers = serverListFilter.getFilteredListOfServers(new ArrayList(lbs.getServerStats().keySet()));
            System.out.println("=====过滤后可用的服务列表：" + filteredListOfServers);

            zones.forEach(zone -> {
                System.out.printf("区域[" + zone + "]概要：");
                int instanceCount = lbs.getInstanceCount(zone);
                int activeRequestsCount = lbs.getActiveRequestsCount(zone);
                double activeRequestsPerServer = lbs.getActiveRequestsPerServer(zone);
                // ZoneSnapshot zoneSnapshot = lbs.getZoneSnapshot(zone);

                System.out.printf("实例总数：%s，活跃请求总数：%s，平均负载：%s\n", instanceCount, activeRequestsCount, activeRequestsPerServer);
                // System.out.println(zoneSnapshot);
            });
            System.out.println("======================================================");
        }, 5, 5, TimeUnit.SECONDS);
    }).start();
}

// 请注意：请必须保证Server的id不一样，否则放不进去List的（因为Server的equals hashCode方法仅和id有关）
// 所以此处使用index作为port，以示区分
private Server createServer(String zone, int index) {
    Server server = new Server("www.baidu" + zone + ".com", index);
    server.setZone(zone);
    return server;
}


// 多线程，模拟请求
private void request(ServerStats serverStats) {
    new Thread(() -> {
        // 每10ms发送一个请求（每个请求处理10-200ms的时间），持续不断
        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
        executorService.scheduleWithFixedDelay(() -> {
            new Thread(() -> {
                // 请求之前 记录活跃请求数
                serverStats.incrementActiveRequestsCount();
                serverStats.incrementNumRequests();
                long rt = doSomething();
                // 请求结束， 记录响应耗时
                serverStats.noteResponseTime(rt);
                serverStats.decrementActiveRequestsCount();
            }).start();
        }, 10, 10, TimeUnit.MILLISECONDS);
    }).start();
}

// 模拟请求耗时，返回耗时时间
private long doSomething() {
    try {
        int rt = randomValue(10, 200);
        TimeUnit.MILLISECONDS.sleep(rt);
        return rt;
    } catch (InterruptedException e) {
        e.printStackTrace();
        return 0L;
    }
}

// 本地使用随机数模拟数据收集
private int randomValue(int min, int max) {
    return min + (int) (Math.random() * ((max - min) + 1));
}

运行程序，控制台打印：

=====当前可用区为：[华南, 华北]
=====过滤后可用的服务列表：[www.baidu华北.com:1, www.baidu华东.com:2, www.baidu华北.com:3, www.baidu华南.com:1, www.baidu华北.com:2, www.baidu华东.com:1, www.baidu华北.com:4]
区域[华南]概要：实例总数：1，活跃请求总数：10，平均负载：10.0
区域[华东]概要：实例总数：2，活跃请求总数：20，平均负载：10.0
区域[华北]概要：实例总数：4，活跃请求总数：31，平均负载：7.75
======================================================
=====当前可用区为：[华南, 华东]
=====过滤后可用的服务列表：[www.baidu华北.com:1, www.baidu华东.com:2, www.baidu华北.com:3, www.baidu华南.com:1, www.baidu华北.com:2, www.baidu华东.com:1, www.baidu华北.com:4]
区域[华南]概要：实例总数：1，活跃请求总数：10，平均负载：10.0
区域[华东]概要：实例总数：2，活跃请求总数：21，平均负载：10.5
区域[华北]概要：实例总数：4，活跃请求总数：42，平均负载：10.5

what a fuck，竟然还是把所有Server给我打印出来了，过滤木有生效？？？那铁定是shouldEnableZoneAffinity()方法返回false喽，而因为我已经EnableZoneAffinity=true了，所以肯定是计算逻辑那块判断没过，这就是Ribbon的一个天坑：负载的计算问题。

本例中【华北】的负载是10左右，而默认的负载阈值是0.6，所以铁定loadPerServer >= activeReqeustsPerServerThreshold这个条件永远成立，所以返回false，过滤失效喽。

现在调整此阈值为：ConfigurationManager.getConfigInstance().setProperty(clientName + "." + config.getNameSpace() + ".zoneAffinity.maxLoadPerServer", 100);再次运行程序，控制台输出为：

=====当前可用区为：[华北, 华东]
=====过滤后可用的服务列表：[www.baidu华北.com:1, www.baidu华北.com:3, www.baidu华北.com:2, www.baidu华北.com:4]
区域[华南]概要：实例总数：1，活跃请求总数：12，平均负载：12.0
区域[华东]概要：实例总数：2，活跃请求总数：23，平均负载：11.5
区域[华北]概要：实例总数：4，活跃请求总数：31，平均负载：7.75
======================================================

这便达到了期望的过滤效果。

总结

关于Ribbon的LoadBalancer五大组件之：ServerListFilter服务列表过滤器就先介绍到这，下文将继续介绍其其它核心组件。

声明

原创不易，码字不易，多谢你的点赞、收藏、关注。把本文分享到你的朋友圈是被允许的，但拒绝抄袭。你也可【左边扫码/或加wx：fsx641385712】邀请你加入我的 Java高工、架构师 系列群大家庭学习和交流。

转载：https://blog.csdn.net/f641385712/article/details/104890138

查看评论

飞道的博客

飞道的博客

个人资料

文章分类

文章存档

阅读排行

评论排行

推荐文章