Spring Cloud Ribbon 负载均衡器

上一篇中进行负载均衡选择具体服务时，是使用的Ribbon 的 LoadBalancer 的 chooseServer 方法，方法中是使用的 Ribbon的 Rule 的 choose 方法。

看一下Ribbon 的 LoadBalancer

//com.netflix.loadbalancer.ILoadBalancer
public interface ILoadBalancer {
   //添加server
   public void addServers(List newServers);
	//选择server
   public Server chooseServer(Object key);
	//标记server 挂掉
   public void markServerDown(Server server);
	//丢弃的方法，获取serverList，根据是否可用
   @Deprecated
   public List getServerList(boolean availableOnly);
	//获取可用的servers
    public List getReachableServers();
	//获取所有的servers
   public List getAllServers();
}

对比接口，添加了一个枚举，添加了三个方法

public abstract class AbstractLoadBalancer implements ILoadBalancer {
    public enum ServerGroup{
        ALL,
        STATUS_UP,
        STATUS_NOT_UP        
    }
	//选择server
    public Server chooseServer() {
       return chooseServer(null);
    }
	//根据分组获取serverList
    public abstract List getServerList(ServerGroup serverGroup);
	//获取负载均衡状态
    public abstract LoadBalancerStats getLoadBalancerStats();    
}

AbstractLoadBalancer有两个实现类BaseLoadBalancer，NoOpLoadBalancer

public class BaseLoadBalancer extends AbstractLoadBalancer implements
        PrimeConnections.PrimeConnectionListener, IClientConfigAware {
	//默认的rule是轮询	在这个类中才 组合了 IRule
    private final static IRule DEFAULT_RULE = new RoundRobinRule();
    //实现接口的方法  添加servers
    public void addServers(List newServers) {
        if (newServers != null && newServers.size() > 0) {
            try {
                ArrayList newList = new ArrayList();
                newList.addAll(allServerList);
                newList.addAll(newServers);
                //这个方法里面使用了一个读写锁
                setServersList(newList);
            } catch (Exception e) {
                logger.error("LoadBalancer [{}]: Exception while adding Servers", name, e);
            }
        }
    }
    //实现接口的方法，选择server	调用rule的choose方法
    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;
            }
        }
    }
    //实现接口的方法，标记server下线
    public void markServerDown(Server server) {
        if (server == null || !server.isAlive()) {
            return;
        }

        logger.error("LoadBalancer [{}]:  markServerDown called on [{}]", name, server.getId());
        server.setAlive(false);
        // forceQuickPing();
		//通知server状态的变化
        notifyServerStatusChangeListener(singleton(server));
    }
    //实现接口的方法，获取可用servers
    public List getReachableServers() {
        return Collections.unmodifiableList(upServerList);
    }
    //实现接口的方法，获取所有的server
    public List getAllServers() {
        return Collections.unmodifiableList(allServerList);
    }
    //实现抽象类的方法，获取serverList
    public List getServerList(ServerGroup serverGroup) {
        switch (serverGroup) {
        case ALL:
            return allServerList;
        case STATUS_UP:
            return upServerList;
        case STATUS_NOT_UP:
            ArrayList notAvailableServers = new ArrayList(
                    allServerList);
            ArrayList upServers = new ArrayList(upServerList);
            notAvailableServers.removeAll(upServers);
            return notAvailableServers;
        }
        return new ArrayList();
    }
    //抽象类的实现方法，获取负载均衡状态
    public LoadBalancerStats getLoadBalancerStats() {
        return lbStats;
    }
}

public class NoOpLoadBalancer extends AbstractLoadBalancer {
    @Override
    public void addServers(List newServers) {
        logger.info("addServers to NoOpLoadBalancer ignored");
    }
    @Override
    public Server chooseServer(Object key) {       
        return null;
    }
    @Override
    public LoadBalancerStats getLoadBalancerStats() {        
        return null;
    }
    @Override
    public List getServerList(ServerGroup serverGroup) {     
        return Collections.emptyList();
    }
    @Override
    public void markServerDown(Server server) {
        logger.info("markServerDown to NoOpLoadBalancer ignored");
    }
   @Override
   public List getServerList(boolean availableOnly) {
      // TODO Auto-generated method stub
      return null;
   }
    @Override
    public List getReachableServers() {
        return null;
    }
    @Override
    public List getAllServers() {
        return null;
    }
}

BaseLoadBalancer有个实现类DynamicServerListLoadBalancer

public class DynamicServerListLoadBalancer extends BaseLoadBalancer {
	//是否https
    boolean isSecure = false;
    boolean useTunnel = false;
    // server状态是否正在更新
    protected AtomicBoolean serverListUpdateInProgress = new AtomicBoolean(false);
    volatile ServerList serverListImpl;
    volatile ServerListFilter filter;
    protected final ServerListUpdater.UpdateAction updateAction = new ServerListUpdater.UpdateAction() {
        @Override
        public void doUpdate() {
            updateListOfServers();
        }
    };
    protected volatile ServerListUpdater serverListUpdater;
	...
	//更新servers
    @VisibleForTesting
    public void updateListOfServers() {
        List servers = new ArrayList();
        if (serverListImpl != null) {
            servers = serverListImpl.getUpdatedListOfServers();
            LOGGER.debug("List of Servers for {} obtained from Discovery client: {}",
                    getIdentifier(), servers);

            if (filter != null) {
                servers = filter.getFilteredListOfServers(servers);
                LOGGER.debug("Filtered List of Servers for {} obtained from Discovery client: {}",
                        getIdentifier(), servers);
            }
        }
        updateAllServerList(servers);
    }
	//更新所有servers
    protected void updateAllServerList(List ls) {
        // other threads might be doing this - in which case, we pass
        if (serverListUpdateInProgress.compareAndSet(false, true)) {
            try {
                for (T s : ls) {
                    s.setAlive(true); // set so that clients can start using these
                                      // servers right away instead
                                      // of having to wait out the ping cycle.
                }
                setServersList(ls);
                super.forceQuickPing();
            } finally {
                serverListUpdateInProgress.set(false);
            }
        }
    }
    ...
}

DynamicServerListLoadBalancer的唯一实现类，也是Ribbon负载均衡默认的负载具衡器

public class ZoneAwareLoadBalancer extends DynamicServerListLoadBalancer {
	//key:zone, value:LoadBalancer
    private ConcurrentHashMap balancers = new ConcurrentHashMap();
    private static final Logger logger = LoggerFactory.getLogger(ZoneAwareLoadBalancer.class);
    private volatile DynamicDoubleProperty triggeringLoad;
    private volatile DynamicDoubleProperty triggeringBlackoutPercentage; 
    private static final DynamicBooleanProperty ENABLED = DynamicPropertyFactory.getInstance().getBooleanProperty("ZoneAwareNIWSDiscoveryLoadBalancer.enabled", true);
	...
	//设置区域和服务
    @Override
    protected void setServerListForZones(Map> zoneServersMap) {
        super.setServerListForZones(zoneServersMap);
        if (balancers == null) {
            balancers = new ConcurrentHashMap();
        }
        for (Map.Entry> entry: zoneServersMap.entrySet()) {
           String zone = entry.getKey().toLowerCase();
            getLoadBalancer(zone).setServersList(entry.getValue());
        }
        // check if there is any zone that no longer has a server
        // and set the list to empty so that the zone related metrics does not
        // contain stale data
        for (Map.Entry existingLBEntry: balancers.entrySet()) {
            if (!zoneServersMap.keySet().contains(existingLBEntry.getKey())) {
                existingLBEntry.getValue().setServersList(Collections.emptyList());
            }
        }
    }    
    //选择server，上一篇讲过    
    @Override
    public Server chooseServer(Object key) {
        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 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 availableZones = ZoneAvoidanceRule.getAvailableZones(zoneSnapshot, triggeringLoad.get(), triggeringBlackoutPercentage.get());
            logger.debug("Available zones: {}", availableZones);
            if (availableZones != null &&  availableZones.size() < zoneSnapshot.keySet().size()) {
                String zone = ZoneAvoidanceRule.randomChooseZone(zoneSnapshot, availableZones);
                logger.debug("Zone chosen: {}", zone);
                if (zone != null) {
                    BaseLoadBalancer zoneLoadBalancer = getLoadBalancer(zone);
                    server = 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);
        }
    }
     //根据zone获取负载均衡器
    @VisibleForTesting
    BaseLoadBalancer getLoadBalancer(String zone) {
        zone = zone.toLowerCase();
        BaseLoadBalancer loadBalancer = balancers.get(zone);
        if (loadBalancer == null) {
           // We need to create rule object for load balancer for each zone
           IRule rule = cloneRule(this.getRule());
            loadBalancer = new BaseLoadBalancer(this.getName() + "_" + zone, rule, this.getLoadBalancerStats());
            BaseLoadBalancer prev = balancers.putIfAbsent(zone, loadBalancer);
            if (prev != null) {
               loadBalancer = prev;
            }
        } 
        return loadBalancer;        
    }
	//克隆rule
    private IRule cloneRule(IRule toClone) {
       IRule rule;
       if (toClone == null) {
          rule = new AvailabilityFilteringRule();
       } else {
          String ruleClass = toClone.getClass().getName();              
          try {
            rule = (IRule) ClientFactory.instantiateInstanceWithClientConfig(ruleClass, this.getClientConfig());
         } catch (Exception e) {
            throw new RuntimeException("Unexpected exception creating rule for ZoneAwareLoadBalancer", e);
         }
       }
       return rule;
    }
    //设置rule  
    @Override
    public void setRule(IRule rule) {
        super.setRule(rule);
        if (balancers != null) {
            for (String zone: balancers.keySet()) {
                balancers.get(zone).setRule(cloneRule(rule));
            }
        }
    }
}