Spring Cloud Ribbon 全解（4） - 基本组件实现源码（2）

本文基于SpringCloud-Dalston.SR5

我们继续逐个分析

所有Ribbon负载均衡器需要实现的接口IClient
服务实例列表维护机制实现的接口ServerList
负载均衡数据记录LoadBalancerStats
负责选取Server的接口ILoadBalancer
负载均衡选取规则实现的接口IRule
检查实例是否存活实现的接口IPing
服务实例列表更新机制实现的接口ServerListUpdater
服务实例列表过滤机制ServerListFilter

4. 负责选取Server的接口ILoadBalancer

ILoadBalancer负责存储并更新服务实例列表，并调用IRule（即根据配置的负载均衡规则）来返回Server以供于服务调用

这里，我们只看默认的ZoneAwareLoadBalancer相关的

AbstractLoadBalancer.java

public abstract class AbstractLoadBalancer implements ILoadBalancer {public enum ServerGroup{ALL,STATUS_UP,STATUS_NOT_UP        }public Server chooseServer() {return chooseServer(null);}public abstract List<Server> getServerList(ServerGroup serverGroup);public abstract LoadBalancerStats getLoadBalancerStats();
}

AbstractLoadBalancer在原有ILoadBalancer接口基础上，增加了按照分组获取Server的方法，有ALL,STATUS_UP,STATUS_NOT_UP三种组别。同时还增加了LoadBalancerStats，记录每次请求的Server的负载均衡统计数据ServerStat，以及其他一些的实时记录信息。之后在介绍具体的LoadBalancer实现的时候，会用到

BaseLoadBalancer.java

BaseLoadBalancer是负载均衡的基本实现，包含如下元素：

1)两个列表：所有Server列表，还有所有Up Server的列表：

protected volatile List<Server> allServerList = Collections.synchronizedList(new ArrayList<Server>());
protected volatile List<Server> upServerList = Collections.synchronizedList(new ArrayList<Server>());//更新两个列表的锁
protected ReadWriteLock allServerLock = new ReentrantReadWriteLock();
protected ReadWriteLock upServerLock = new ReentrantReadWriteLock();

2)定时PING任务相关的元素，为了定时检查Server是否UP的任务：

private final static SerialPingStrategy DEFAULT_PING_STRATEGY = new SerialPingStrategy();
protected IPingStrategy pingStrategy = DEFAULT_PING_STRATEGY;
protected IPing ping = null;protected Timer lbTimer = null;
protected int pingIntervalSeconds = 10;
protected int maxTotalPingTimeSeconds = 5;
protected Comparator<Server> serverComparator = new ServerComparator();
protected AtomicBoolean pingInProgress = new AtomicBoolean(false);

一般的在构造一个BaseLoadBalancer时候，会调用setupPingTask构造一个定时ping的任务：


void setupPingTask() {//判断是否需要pingif (canSkipPing()) {return;}//关闭之前已经开启的定时ping的任务if (lbTimer != null) {lbTimer.cancel();}//设置PingTask，默认是每10秒一次lbTimer = new ShutdownEnabledTimer("NFLoadBalancer-PingTimer-" + name,true);lbTimer.schedule(new PingTask(), 0, pingIntervalSeconds * 1000);//这个可能有些多余，因为定时任务也是立即执行，就会pingforceQuickPing();
}//如果ping为null，或者为DummyPing，就不用定时Ping了
private boolean canSkipPing() {if (ping == null|| ping.getClass().getName().equals(DummyPing.class.getName())) {// default ping, no need to set up timerreturn true;} else {return false;}
}

Pinger包含了如何去Ping所有的Server的逻辑：

class Pinger {//ping每个server的方式，默认就是普通遍历private final IPingStrategy pingerStrategy;public Pinger(IPingStrategy pingerStrategy) {this.pingerStrategy = pingerStrategy;}public void runPinger() throws Exception {//如果设置失败，则证明，当前Ping的任务正在执行中（执行时间大于定时任务的周期）if (!pingInProgress.compareAndSet(false, true)) { return; // Ping in progress - nothing to do}// we are "in" - we get to PingServer[] allServers = null;boolean[] results = null;Lock allLock = null;Lock upLock = null;try {//读取所有Server的列表，需要获取读锁allLock = allServerLock.readLock();allLock.lock();//先读取之后再遍历，减少锁时间allServers = allServerList.toArray(new Server[allServerList.size()]);allLock.unlock();//ping每个Serverint numCandidates = allServers.length;results = pingerStrategy.pingServers(ping, allServers);final List<Server> newUpList = new ArrayList<Server>();final List<Server> changedServers = new ArrayList<Server>();//更新所有Server的状态for (int i = 0; i < numCandidates; i++) {boolean isAlive = results[i];Server svr = allServers[i];boolean oldIsAlive = svr.isAlive();svr.setAlive(isAlive);if (oldIsAlive != isAlive) {changedServers.add(svr);logger.debug("LoadBalancer [{}]:  Server [{}] status changed to {}", name, svr.getId(), (isAlive ? "ALIVE" : "DEAD"));}if (isAlive) {newUpList.add(svr);}}//更新Up Server列表需要获取写锁upLock = upServerLock.writeLock();upLock.lock();upServerList = newUpList;upLock.unlock();//通知监听Server状态变化的ListenernotifyServerStatusChangeListener(changedServers);} finally {//任务结束，需要设置状态位pingInProgress.set(false);}}
}

IPingStrategy目前只有一种默认实现，就是SerialPingStrategy，依次串行遍历Ping每个Server：

private static class SerialPingStrategy implements IPingStrategy {@Overridepublic boolean[] pingServers(IPing ping, Server[] servers) {int numCandidates = servers.length;boolean[] results = new boolean[numCandidates];logger.debug("LoadBalancer:  PingTask executing [{}] servers configured", numCandidates);for (int i = 0; i < numCandidates; i++) {results[i] = false; /* Default answer is DEAD. */try {//因为在SpringCloud的环境下，默认ping是基于访问本地内存的Eureka缓存的列表，所有串行挨个ping也不会有太大的性能影响if (ping != null) {results[i] = ping.isAlive(servers[i]);}} catch (Exception e) {logger.error("Exception while pinging Server: '{}'", servers[i], e);}}return results;}
}

3)记录每次负载均衡统计数据的LoadBalancerStats
这个暂且不表，在后面具体介绍负载均衡选取规则的时候，会用到这里面的统计数据

protected LoadBalancerStats lbStats;

4)用于预热连接的逻辑类PrimeConnections

private PrimeConnections primeConnections;
private volatile boolean enablePrimingConnections = false;

但是这个配置默认是不开启的。

5)监听Server列表变化和Server状态变化的Listener

private List<ServerListChangeListener> changeListeners = new CopyOnWriteArrayList<ServerListChangeListener>();private List<ServerStatusChangeListener> serverStatusListeners = new CopyOnWriteArrayList<ServerStatusChangeListener>();

选择Server的方法非常简单，就是调用IRule来选取：

public Server chooseServer(Object key) {if (counter == null) {counter = createCounter();}counter.increment();if (rule == null) {return null;} else {try {return rule.choose(key);} catch (Exception e) {logger.warn("LoadBalancer [{}]:  Error choosing server for key {}", name, key, e);return null;}}
}

DynamicServerListLoadBalancer.java

DynamicServerListLoadBalancer在BaseLoadBalancer的基础上，增加了之前提到的：服务实例列表维护机制实现的接口ServerList、服务实例列表更新机制实现的接口ServerListUpdater和服务实例列表过滤机制ServerListFilter。利用这些元素实现服务实例列表的更新

ZoneAwareLoadBalancer.java

ZoneAwareLoadBalancer则是进一步增加了对于Zone的感知。利用一个ConcurrentHashMap来维护不同Zone下的负载均衡数据，并且不同的Zone可以设置不同的Rule：

 private ConcurrentHashMap<String, BaseLoadBalancer> balancers = new ConcurrentHashMap<String, BaseLoadBalancer>();

我们可以通过setRule方法给不同的Zone设置不同的IRule

public void setRule(IRule rule) {super.setRule(rule);if (balancers != null) {for (String zone: balancers.keySet()) {balancers.get(zone).setRule(cloneRule(rule));}}
}

查看负载均衡方法：

public Server chooseServer(Object key) {//如果未启用（默认是启用的，可以通过ZoneAwareNIWSDiscoveryLoadBalancer.enabled修改），或者可用区域不大于1，则调用BaseLoadBalancer的choose方法选取if (!ENABLED.get() || getLoadBalancerStats().getAvailableZones().size() <= 1) {logger.debug("Zone aware logic disabled or there is only one zone");return super.chooseServer(key);}Server server = null;try {LoadBalancerStats lbStats = getLoadBalancerStats();//获取当前负载均衡数据的快照Map<String, ZoneSnapshot> zoneSnapshot = ZoneAvoidanceRule.createSnapshot(lbStats);logger.debug("Zone snapshots: {}", zoneSnapshot);//获取最大负载阈值配置if (triggeringLoad == null) {triggeringLoad = DynamicPropertyFactory.getInstance().getDoubleProperty("ZoneAwareNIWSDiscoveryLoadBalancer." + this.getName() + ".triggeringLoadPerServerThreshold", 0.2d);}//获取熔断实例比例配置if (triggeringBlackoutPercentage == null) {triggeringBlackoutPercentage = DynamicPropertyFactory.getInstance().getDoubleProperty("ZoneAwareNIWSDiscoveryLoadBalancer." + this.getName() + ".avoidZoneWithBlackoutPercetage", 0.99999d);}//获取当前有效的可用区域（利用最大负载阈值配置和熔断实例比例配置）Set<String> availableZones = ZoneAvoidanceRule.getAvailableZones(zoneSnapshot, triggeringLoad.get(), triggeringBlackoutPercentage.get());logger.debug("Available zones: {}", availableZones);if (availableZones != null &&  availableZones.size() < zoneSnapshot.keySet().size()) {//随机选取zoneString zone = ZoneAvoidanceRule.randomChooseZone(zoneSnapshot, availableZones);logger.debug("Zone chosen: {}", zone);if (zone != null) {BaseLoadBalancer zoneLoadBalancer = getLoadBalancer(zone);//在选取的zone下选取serverserver = zoneLoadBalancer.chooseServer(key);}}} catch (Exception e) {logger.error("Error choosing server using zone aware logic for load balancer={}", name, e);}if (server != null) {return server;} else {logger.debug("Zone avoidance logic is not invoked.");return super.chooseServer(key);}
}

如何筛选的AvailableZone：

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;for (Map.Entry<String, ZoneSnapshot> zoneEntry : snapshot.entrySet()) {String zone = zoneEntry.getKey();ZoneSnapshot zoneSnapshot = zoneEntry.getValue();int instanceCount = zoneSnapshot.getInstanceCount();//可用区域数量为0，这个可用区需要被移除if (instanceCount == 0) {availableZones.remove(zone);limitedZoneAvailability = true;} else {double loadPerServer = zoneSnapshot.getLoadPerServer();//熔断的实例个数/实例数量大于triggeringBlackoutPercentage，这个可用区需要被移除if (((double) zoneSnapshot.getCircuitTrippedCount())/ instanceCount >= triggeringBlackoutPercentage|| loadPerServer < 0) {availableZones.remove(zone);limitedZoneAvailability = true;} else {//寻找平均负载最高的可用区，将它添加到worstZones集合if (Math.abs(loadPerServer - maxLoadPerServer) < 0.000001d) {// they are the same considering double calculation// round errorworstZones.add(zone);} else if (loadPerServer > maxLoadPerServer) {maxLoadPerServer = loadPerServer;worstZones.clear();worstZones.add(zone);}}}}//如果最大负载小于负载阈值，并且没有被移除过可用区，就直接返回当前结果if (maxLoadPerServer < triggeringLoad && !limitedZoneAvailability) {// zone override is not needed herereturn availableZones;}String zoneToAvoid = randomChooseZone(snapshot, worstZones);if (zoneToAvoid != null) {availableZones.remove(zoneToAvoid);}return availableZones;}