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