Spring Cloud Ribbon负载均衡器处理方法

spring-cloud-netflix-core-1.3.5.RELEASE.jar
org.springframework.cloud.netflix.ribbon
RibbonClientConfiguration.class

@SuppressWarnings("deprecation")
@Configuration
@EnableConfigurationProperties
//Order is important here, last should be the default, first should be optional
// see https://github.com/spring-cloud/spring-cloud-netflix/issues/2086#issuecomment-316281653
@Import({OkHttpRibbonConfiguration.class, RestClientRibbonConfiguration.class, HttpClientRibbonConfiguration.class})
public class RibbonClientConfiguration {
// 略
 @Bean
 @ConditionalOnMissingBean
 public ILoadBalancer ribbonLoadBalancer(IClientConfig config,
 ServerList<Server> serverList, ServerListFilter<Server> serverListFilter,
 IRule rule, IPing ping, ServerListUpdater serverListUpdater) {
 if (this.propertiesFactory.isSet(ILoadBalancer.class, name)) {
 return this.propertiesFactory.get(ILoadBalancer.class, config, name);
 }
 return new ZoneAwareLoadBalancer<>(config, rule, ping, serverList,
 serverListFilter, serverListUpdater);
 }
// 略
}

@SuppressWarnings("deprecation")
@Configuration
@EnableConfigurationProperties
//Order is important here, last should be the default, first should be optional
// see https://github.com/spring-cloud/spring-cloud-netflix/issues/2086#issuecomment-316281653
@Import({OkHttpRibbonConfiguration.class, RestClientRibbonConfiguration.class, HttpClientRibbonConfiguration.class})
public class RibbonClientConfiguration {
 // 略
 @Bean
 @ConditionalOnMissingBean
 public IClientConfig ribbonClientConfig() {
 DefaultClientConfigImpl config = new DefaultClientConfigImpl();
 config.loadProperties(this.name);
 return config;
 }
 @Bean
 @ConditionalOnMissingBean
 public IRule ribbonRule(IClientConfig config) {
 if (this.propertiesFactory.isSet(IRule.class, name)) {
 return this.propertiesFactory.get(IRule.class, config, name);
 }
 ZoneAvoidanceRule rule = new ZoneAvoidanceRule();
 rule.initWithNiwsConfig(config);
 return rule;
 }
 @Bean
 @ConditionalOnMissingBean
 public IPing ribbonPing(IClientConfig config) {
 if (this.propertiesFactory.isSet(IPing.class, name)) {
 return this.propertiesFactory.get(IPing.class, config, name);
 }
 return new DummyPing();
 }
 @Bean
 @ConditionalOnMissingBean
 @SuppressWarnings("unchecked")
 public ServerList<Server> ribbonServerList(IClientConfig config) {
 if (this.propertiesFactory.isSet(ServerList.class, name)) {
 return this.propertiesFactory.get(ServerList.class, config, name);
 }
 ConfigurationBasedServerList serverList = new ConfigurationBasedServerList();
 serverList.initWithNiwsConfig(config);
 return serverList;
 }
 @Bean
 @ConditionalOnMissingBean
 public ServerListUpdater ribbonServerListUpdater(IClientConfig config) {
 return new PollingServerListUpdater(config);
 }
 @Bean
 @ConditionalOnMissingBean
 public ILoadBalancer ribbonLoadBalancer(IClientConfig config,
 ServerList<Server> serverList, ServerListFilter<Server> serverListFilter,
 IRule rule, IPing ping, ServerListUpdater serverListUpdater) {
 if (this.propertiesFactory.isSet(ILoadBalancer.class, name)) {
 return this.propertiesFactory.get(ILoadBalancer.class, config, name);
 }
 return new ZoneAwareLoadBalancer<>(config, rule, ping, serverList,
 serverListFilter, serverListUpdater);
 }
 @Bean
 @ConditionalOnMissingBean
 @SuppressWarnings("unchecked")
 public ServerListFilter<Server> ribbonServerListFilter(IClientConfig config) {
 if (this.propertiesFactory.isSet(ServerListFilter.class, name)) {
 return this.propertiesFactory.get(ServerListFilter.class, config, name);
 }
 ZonePreferenceServerListFilter filter = new ZonePreferenceServerListFilter();
 filter.initWithNiwsConfig(config);
 return filter;
 }
 @Bean
 @ConditionalOnMissingBean
 public RibbonLoadBalancerContext ribbonLoadBalancerContext(
 ILoadBalancer loadBalancer, IClientConfig config, RetryHandler retryHandler) {
 return new RibbonLoadBalancerContext(loadBalancer, config, retryHandler);
 }
 // 略
}

spring-cloud-netflix-eureka-client-1.3.5.RELEASE.jar
org.springframework.cloud.netflix.ribbon.eureka
EurekaRibbonClientConfiguration.class

@Configuration
@CommonsLog
public class EurekaRibbonClientConfiguration {
 // 略
 @Bean
 @ConditionalOnMissingBean
 public IPing ribbonPing(IClientConfig config) {
 if (this.propertiesFactory.isSet(IPing.class, serviceId)) {
 return this.propertiesFactory.get(IPing.class, config, serviceId);
 }
 NIWSDiscoveryPing ping = new NIWSDiscoveryPing();
 ping.initWithNiwsConfig(config);
 return ping;
 }
 @Bean
 @ConditionalOnMissingBean
 public ServerList<?> ribbonServerList(IClientConfig config, Provider<EurekaClient> eurekaClientProvider) {
 if (this.propertiesFactory.isSet(ServerList.class, serviceId)) {
 return this.propertiesFactory.get(ServerList.class, config, serviceId);
 }
 DiscoveryEnabledNIWSServerList discoveryServerList = new DiscoveryEnabledNIWSServerList(
 config, eurekaClientProvider);
 DomainExtractingServerList serverList = new DomainExtractingServerList(
 discoveryServerList, config, this.approximateZoneFromHostname);
 return serverList;
 }
 // 略
}

 void restOfInit(IClientConfig clientConfig) {
 boolean primeConnection = this.isEnablePrimingConnections();
 // turn this off to avoid duplicated asynchronous priming done in BaseLoadBalancer.setServerList()
 this.setEnablePrimingConnections(false);
 enableAndInitLearnNewServersFeature();
 updateListOfServers();
 if (primeConnection && this.getPrimeConnections() != null) {
 this.getPrimeConnections()
  .primeConnections(getReachableServers());
 }
 this.setEnablePrimingConnections(primeConnection);
 LOGGER.info("DynamicServerListLoadBalancer for client {} initialized: {}", clientConfig.getClientName(), this.toString());
 }

 /**
 * Feature that lets us add new instances (from AMIs) to the list of
 * existing servers that the LB will use Call this method if you want this
 * feature enabled
 */
 public void enableAndInitLearnNewServersFeature() {
 LOGGER.info("Using serverListUpdater {}", serverListUpdater.getClass().getSimpleName());
 serverListUpdater.start(updateAction);
 }

public class PollingServerListUpdater implements ServerListUpdater {
 // 略
 private static long LISTOFSERVERS_CACHE_UPDATE_DELAY = 1000; // msecs;
 private static int LISTOFSERVERS_CACHE_REPEAT_INTERVAL = 30 * 1000; // msecs; 
 // 略
 private final AtomicBoolean isActive = new AtomicBoolean(false);
 private volatile long lastUpdated = System.currentTimeMillis();
 private final long initialDelayMs;
 private final long refreshIntervalMs;
 // 略
 public PollingServerListUpdater(IClientConfig clientConfig) {
 this(LISTOFSERVERS_CACHE_UPDATE_DELAY, getRefreshIntervalMs(clientConfig));
 }
 public PollingServerListUpdater(final long initialDelayMs, final long refreshIntervalMs) {
 this.initialDelayMs = initialDelayMs;
 this.refreshIntervalMs = refreshIntervalMs;
 }
 @Override
 public synchronized void start(final UpdateAction updateAction) {
 if (isActive.compareAndSet(false, true)) {
 final Runnable wrapperRunnable = new Runnable() {
 @Override
 public void run() {
  if (!isActive.get()) {
  if (scheduledFuture != null) {
  scheduledFuture.cancel(true);
  }
  return;
  }
  try {
  updateAction.doUpdate();
  lastUpdated = System.currentTimeMillis();
  } catch (Exception e) {
  logger.warn("Failed one update cycle", e);
  }
 }
 };
 scheduledFuture = getRefreshExecutor().scheduleWithFixedDelay(
  wrapperRunnable,
  initialDelayMs,
  refreshIntervalMs,
  TimeUnit.MILLISECONDS
 );
 } else {
 logger.info("Already active, no-op");
 }
 }
 // 略
}

protected final ServerListUpdater.UpdateAction updateAction = new ServerListUpdater.UpdateAction() {
 @Override
 public void doUpdate() {
 updateListOfServers();
 }
 };

public void updateListOfServers() {
 List<T> servers = new ArrayList<T>();
 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);
 }

public class DiscoveryEnabledNIWSServerList extends AbstractServerList<DiscoveryEnabledServer>{
 // 略
 @Override
 public List<DiscoveryEnabledServer> getInitialListOfServers(){
 return obtainServersViaDiscovery();
 }
 @Override
 public List<DiscoveryEnabledServer> getUpdatedListOfServers(){
 return obtainServersViaDiscovery();
 }
 private List<DiscoveryEnabledServer> obtainServersViaDiscovery() {
 List<DiscoveryEnabledServer> serverList = new ArrayList<DiscoveryEnabledServer>();
 if (eurekaClientProvider == null || eurekaClientProvider.get() == null) {
 logger.warn("EurekaClient has not been initialized yet, returning an empty list");
 return new ArrayList<DiscoveryEnabledServer>();
 }
 EurekaClient eurekaClient = eurekaClientProvider.get();
 if (vipAddresses!=null){
 for (String vipAddress : vipAddresses.split(",")) {
 // if targetRegion is null, it will be interpreted as the same region of client
 List<InstanceInfo> listOfInstanceInfo = eurekaClient.getInstancesByVipAddress(vipAddress, isSecure, targetRegion);
 for (InstanceInfo ii : listOfInstanceInfo) {
  if (ii.getStatus().equals(InstanceStatus.UP)) {
  if(shouldUseOverridePort){
  if(logger.isDebugEnabled()){
  logger.debug("Overriding port on client name: " + clientName + " to " + overridePort);
  }
  // copy is necessary since the InstanceInfo builder just uses the original reference,
  // and we don't want to corrupt the global eureka copy of the object which may be
  // used by other clients in our system
  InstanceInfo copy = new InstanceInfo(ii);
  if(isSecure){
  ii = new InstanceInfo.Builder(copy).setSecurePort(overridePort).build();
  }else{
  ii = new InstanceInfo.Builder(copy).setPort(overridePort).build();
  }
  }
  DiscoveryEnabledServer des = new DiscoveryEnabledServer(ii, isSecure, shouldUseIpAddr);
  des.setZone(DiscoveryClient.getZone(ii));
  serverList.add(des);
  }
 }
 if (serverList.size()>0 && prioritizeVipAddressBasedServers){
  break; // if the current vipAddress has servers, we dont use subsequent vipAddress based servers
 }
 }
 }
 return serverList;
 }
 // 略
}

@Data
@EqualsAndHashCode(callSuper = false)
public class ZonePreferenceServerListFilter extends ZoneAffinityServerListFilter<Server> {
 private String zone;
 @Override
 public void initWithNiwsConfig(IClientConfig niwsClientConfig) {
 super.initWithNiwsConfig(niwsClientConfig);
 if (ConfigurationManager.getDeploymentContext() != null) {
 this.zone = ConfigurationManager.getDeploymentContext().getValue(
  ContextKey.zone);
 }
 }
 @Override
 public List<Server> getFilteredListOfServers(List<Server> servers) {
 List<Server> output = super.getFilteredListOfServers(servers);
 if (this.zone != null && output.size() == servers.size()) {
 List<Server> local = new ArrayList<Server>();
 for (Server server : output) {
 if (this.zone.equalsIgnoreCase(server.getZone())) {
  local.add(server);
 }
 }
 if (!local.isEmpty()) {
 return local;
 }
 }
 return output;
 }
}

public class ZoneAffinityServerListFilter<T extends Server> extends
 AbstractServerListFilter<T> implements IClientConfigAware {
 // 略
 private boolean shouldEnableZoneAffinity(List<T> filtered) { 
 if (!zoneAffinity && !zoneExclusive) {
 return false;
 }
 if (zoneExclusive) {
 return true;
 }
 LoadBalancerStats stats = getLoadBalancerStats();
 if (stats == null) {
 return zoneAffinity;
 } else {
 logger.debug("Determining if zone affinity should be enabled with given server list: {}", filtered);
 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()) {
 logger.debug("zoneAffinity is overriden. blackOutServerPercentage: {}, activeReqeustsPerServer: {}, availableServers: {}", 
  new Object[] {(double) circuitBreakerTrippedCount / instanceCount, loadPerServer, instanceCount - circuitBreakerTrippedCount});
 return false;
 } else {
 return true;
 }
 }
 }
 @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;
 }
 // 略
}

@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<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()) {
 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);
 }
 }

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;
 }
 }
 }

public void setRule(IRule rule) {
 if (rule != null) {
 this.rule = rule;
 } else {
 /* default rule */
 this.rule = new RoundRobinRule();
 }
 if (this.rule.getLoadBalancer() != this) {
 this.rule.setLoadBalancer(this);
 }
 }

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<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()) {
 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);
 }
 }