微服务组件源码2——Spring Ribbon原理（基于RibbonLoadBalancerClient）
1、基本原理（LoadBalancerClient + 拦截器）
Ribbon 负载组件的内部就是集成了 LoadBalancerClient 负载均衡客户端，所以 Ribbon 负载均衡的原理本质也跟上面介绍的 LoadBalancerClient 原理一致，负载均衡器 Ribbon 默认会通过 Eureka Client 向 Eureka 服务端的服务注册列表中获取服务的信息，并缓存一份在本地 JVM 中，根据缓存的服务注册列表信息，可以通过 LoadBalancerClient 来选择不同的服务实例，从而实现负载均衡。主要做如下的增强处理：
基本用法就是注入一个 RestTemplate，并使用@LoadBalance注解标注 RestTemplate，从而使RestTemplate具备负载均衡的能力。当 Spring 容器启动时，使用@LoadBalanced 注解修饰的 RestTemplate 会被添加拦截器 LoadBalancerInterceptor，拦截器会拦截 RestTemplate 发送的请求，转而执行LoadBalancerInterceptor 中的 intercept() 方法，并在 intercept()方法中使用LoadBalancerClient 处理请求，从而达到负载均衡的目的。
那么 RestTemplate 添加 @LoadBalanced 注解后，为什么会被拦截呢？这是因为 LoadBalancerAutoConfiguration 类维护了一个被 @LoadBalanced 修饰的 RestTemplate 列表，在初始化过程中，通过调用 customizer.customize(restTemplate) 方法为 RestTemplate 添加了 LoadBalancerInterceptor 拦截器，该拦截器中的方法将远程服务调用的方法交给了 LoadBalancerClient 去处理，从而达到了负载均衡的目的。
Ribbon会提供一个serviceId对应的服务实例，此外就不会做其他事了，基于此服务实例进行远程调用、返回结果处理等都是由RestTemplate去处理的
2、自动配置概述
2.1、RibbonAutoConfiguration
* 收集@RibbonClients注解信息封装为RibbonClientSpecification（功能和@LoadBalancerClients注解类似）
会收集注册@RibbonClients、RibbonClient注解，name和指定的配置类（包括默认）为RibbonClientSpecification(String name, Class<?>[] configuration)类
@Configuration
@RibbonClient(name = "orderService",configuration = HelloRibbonConfiguration.class)
@RibbonClients(defaultConfiguration = MyRibbonConfiguration.class)
public class RibbonConfiguration {}
* 创建SpringClientFactory extends NamedContextFactory<RibbonClientSpecification>命名上下文
①会基于容器中List<RibbonClientSpecification> configurations（上一步注册的配置集合），为每一个指定的name 创建一个上下文，父上下文为顶层Spring容器
每个上下文里包含可独自指定的RibbonClientSpecification配置和默认配置类RibbonClientConfiguration，如IClientConfig客户端配置、IRule策略的配置、超时配置等
②其中在RibbonClientConfiguration里的IClientConfig客户端配置的默认配置为DefaultClientConfigImpl，它还会加载文件配置的信息
从Spring env中加载“ribbon.配置项”这类全局默认配置、和加载“client名.ribbon.配置项”这类针对某个Client的配置信息
* 创建RibbonLoadBalancerClient(springClientFactory()) 用户使用的顶级LoadBalancerClient 接口对象
①List<RibbonClientSpecification>收集全局配置和单个服务的配置信息
②SpringClientFactory会为每一个服务创建独立的上下文
③new RibbonLoadBalancerClient(springClientFactory())
@Configuration
@Conditional(org.springframework.cloud.netflix.ribbon.RibbonAutoConfiguration.RibbonClassesConditions.class)
//其内RibbonClientConfigurationRegistrar会收集注册@RibbonClients、RibbonClient注解内的那么，name和指定的配置类（包括默认）为RibbonClientSpecification
@RibbonClients
@AutoConfigureAfter(name = "org.springframework.cloud.netflix.eureka.EurekaClientAutoConfiguration")
//把@LoadBalanced 注解修饰的 RestTemplate 会被添加拦截器LoadBalancerInterceptor(loadBalancerClient, requestFactory)
@AutoConfigureBefore({ LoadBalancerAutoConfiguration.class, AsyncLoadBalancerAutoConfiguration.class })
@EnableConfigurationProperties({ RibbonEagerLoadProperties.class, ServerIntrospectorProperties.class })
public class RibbonAutoConfiguration {
@Autowired(required = false)
private List<RibbonClientSpecification> configurations = new ArrayList<>();
@Autowired
private RibbonEagerLoadProperties ribbonEagerLoadProperties;
@Bean
public HasFeatures ribbonFeature() {
return HasFeatures.namedFeature("Ribbon", Ribbon.class);
}
//客户端配置容器工厂：会为每一个ClientName和指定的配置类(包含了默认配置类RibbonClientConfiguration)创建一个独立的上下文
@Bean
public SpringClientFactory springClientFactory() {
SpringClientFactory factory = new SpringClientFactory();
factory.setConfigurations(this.configurations);
return factory;
}
//使用客户端配置容器工厂创建顶层对象，RibbonLoadBalancerClient，实现了LoadBalancerClient接口
@Bean
@ConditionalOnMissingBean(LoadBalancerClient.class)
public LoadBalancerClient loadBalancerClient() {
return new RibbonLoadBalancerClient(springClientFactory());
}
//略
}
//SpringClientFactory
public class SpringClientFactory extends NamedContextFactory<RibbonClientSpecification> {
static final String NAMESPACE = "ribbon";
public SpringClientFactory() {
super(RibbonClientConfiguration.class, NAMESPACE, "ribbon.client.name");
}
public ILoadBalancer getLoadBalancer(String name) {
return getInstance(name, ILoadBalancer.class);
}
//通过NamedContextFactory获取指定容器内的某个type实例，获取不到时，会通过IClientConfig进行创建，
//调用容器的autowireBean进属性注入，但是最终实例不会放入容器中
@Override
public <C> C getInstance(String name, Class<C> type) {
C instance = super.getInstance(name, type);
if (instance != null) {
return instance;
}
IClientConfig config = getInstance(name, IClientConfig.class);
return instantiateWithConfig(getContext(name), type, config);
}
//略
}
2.2、RibbonClientConfiguration默认配置(serviceId维度)
* 是默认的Ribbon客户端配置类，在创建SpringClientFactory是指定的默认配置类
* 每个独立的服务ID上下文中都会注入此类配置的信息
- 指定了robbin会用到的所有组件的bean
- 同时还import指定了具体的httpClient的自动配置
@Import({ HttpClientConfiguration.class, OkHttpRibbonConfiguration.class,
RestClientRibbonConfiguration.class, HttpClientRibbonConfiguration.class })
2.2.1、Name容器中6大可配置组件
- Spring Ribbon的6大可配置组件类及默认配置
|
自动化配置接口 |
描述 |
默认实现 |
说明 |
|
IClientConfig |
Ribbon的客户端配置 |
com.netflix.client.config.DefaultClientConfigImpl |
对如下组件及其他信息的配置项 |
|
IRule |
Ribbon的负载均衡策略 |
com.netflix.loadbalancer.ZoneAvoidanceRule |
该策略能在多区域环境下选出最佳区域的实例进行访问 |
|
IPing |
Ribbon的实例检查策略 |
com.netflix.loadbalancer.NoOpPing |
该检查策略是一个特殊的实现，实际上它并不会检查实例是否可用，而是始终返回true，默认所有的实例都是可用的 |
|
ServerList<Server> |
服务实例清单维护机制 |
com.netflix.loadbalancer.ConfigurationBasedServerList |
|
|
ServerListFilter<Server> |
服务实例清单过滤机制 |
org.springframework.cloud.netflix.ribbon.ZonePreferenceServerListFilter |
该策略能够优先过滤出与请求调用方处于同一个区域的服务清单 |
|
ILoadBalancer |
（负载均衡器主类） |
com.netflix.loadbalancer.ZoneAwareLoadBalancer |
该策略具备服务感知能力，封装了如上几个组件 |
- 与Eureka、Ncaos集成
* 当在Spring Cloud中同时引入Spring Cloud Eureka 和 Spring Cloud Ribbon 时，会触发Eureka对于Ribbon的自动化配置，那么Ribbon的相关默认实现类就会有所变化。
|
自动化配置接口 |
描述 |
默认实现 |
说明 |
|
IPing |
Ribbon的实例检查策略 |
com.netflix.niws.loadbalancer.NIWSDiscoveryPing |
该实现将实例检查的任务交给服务治理框架来进行维护 |
|
ServerList<Server> |
服务实例清单维护机制 |
com.netflix.niws.loadbalancer.DiscoveryEnabledNIWSServerList |
该实现会将服务清单列表交给Eureka来维护 |
在与Spring Cloud Eureka结合使用时，我们的配置会更简单，例如上一步中提到的客户端配置EUREKA-CLIENT.ribbon.listOfServers，就不需要再这么麻烦的进行配置，
因为Eureka的自动配置类会为我们维护所有实例的清单。
我们也可以通过参数配置来禁用Eureka对Ribbon服务实例的维护实现：ribbon.eureka.enabled=false
* 同样的，当在Spring Cloud中同时引入Spring Cloud Nacos和 Spring Cloud Ribbon 时，且禁用了Eureka，此时也会只有Nacos自定义的实现类覆盖关默认实现类
例如我们希望使用Nacos的IRule实现，那么在配置类中加上
@Bean
public IRule ribbonRule() {
return new NacosRule();
}
// Spring Cloud Ribbon中对RibbonClient的默认配置类
@Configuration(proxyBeanMethods = false)
@EnableConfigurationProperties
//具体通信工具配置HttpClient
@Import({ HttpClientConfiguration.class, OkHttpRibbonConfiguration.class,
RestClientRibbonConfiguration.class, HttpClientRibbonConfiguration.class })
public class RibbonClientConfiguration {
public static final int DEFAULT_CONNECT_TIMEOUT = 1000;
public static final int DEFAULT_READ_TIMEOUT = 1000;
public static final boolean DEFAULT_GZIP_PAYLOAD = true;
@RibbonClientName
private String name = "client";
@Autowired
private PropertiesFactory propertiesFactory;
@Bean
@ConditionalOnMissingBean
public IClientConfig ribbonClientConfig() {
DefaultClientConfigImpl config = new DefaultClientConfigImpl();
config.loadProperties(this.name);
config.set(CommonClientConfigKey.ConnectTimeout, DEFAULT_CONNECT_TIMEOUT);
config.set(CommonClientConfigKey.ReadTimeout, DEFAULT_READ_TIMEOUT);
config.set(CommonClientConfigKey.GZipPayload, DEFAULT_GZIP_PAYLOAD);
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 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);
}
//略
}
2.2.2、OkHttpRibbonConfiguration （例）
配置具体的通信工具，例如当项目存在OkHttpClient类且存在配置属性ribbon.okhttp.enabled，会加载这个配置类
* OkHttpLoadBalancingClient和RetryableOkHttpLoadBalancingClient实现类bean
是实现负载均衡功能中顶级接口com.netflix.client.IClient
* OkHttpClientConfiguration
是存粹的OkHttpClient的配置，包括配置连接池、生成OkHttpClient对象
注意这里OkHttpClient的生成，会从IClientConfig config配置中获取Ribbon相关的配置（超时配置），设置在OkHttpClient中
注：基于RestTemplate使用ribbon的方法内，是没有用到此配置的bean。是直接借助RestTemplate的通信功能进行访问。Ribbon值提供serverInstan的选择
@Configuration(proxyBeanMethods = false)
@ConditionalOnProperty("ribbon.okhttp.enabled")
@ConditionalOnClass(name = "okhttp3.OkHttpClient")
public class OkHttpRibbonConfiguration {
@RibbonClientName
private String name = "client";
@Bean
@ConditionalOnMissingBean(AbstractLoadBalancerAwareClient.class)
@ConditionalOnClass(name = "org.springframework.retry.support.RetryTemplate")
public RetryableOkHttpLoadBalancingClient retryableOkHttpLoadBalancingClient(
IClientConfig config, ServerIntrospector serverIntrospector,
ILoadBalancer loadBalancer, RetryHandler retryHandler,
LoadBalancedRetryFactory loadBalancedRetryFactory, OkHttpClient delegate,
RibbonLoadBalancerContext ribbonLoadBalancerContext) {
RetryableOkHttpLoadBalancingClient client = new RetryableOkHttpLoadBalancingClient(
delegate, config, serverIntrospector, loadBalancedRetryFactory);
client.setLoadBalancer(loadBalancer);
client.setRetryHandler(retryHandler);
client.setRibbonLoadBalancerContext(ribbonLoadBalancerContext);
Monitors.registerObject("Client_" + this.name, client);
return client;
}
@Bean
@ConditionalOnMissingBean(AbstractLoadBalancerAwareClient.class)
@ConditionalOnMissingClass("org.springframework.retry.support.RetryTemplate")
public OkHttpLoadBalancingClient okHttpLoadBalancingClient(IClientConfig config,
ServerIntrospector serverIntrospector, ILoadBalancer loadBalancer,
RetryHandler retryHandler, OkHttpClient delegate) {
OkHttpLoadBalancingClient client = new OkHttpLoadBalancingClient(delegate, config,
serverIntrospector);
client.setLoadBalancer(loadBalancer);
client.setRetryHandler(retryHandler);
Monitors.registerObject("Client_" + this.name, client);
return client;
}
@Configuration(proxyBeanMethods = false)
protected static class OkHttpClientConfiguration {
private OkHttpClient httpClient;
@Bean
@ConditionalOnMissingBean(ConnectionPool.class)
public ConnectionPool httpClientConnectionPool(IClientConfig config,
OkHttpClientConnectionPoolFactory connectionPoolFactory) {
RibbonProperties ribbon = RibbonProperties.from(config);
int maxTotalConnections = ribbon.maxTotalConnections();
long timeToLive = ribbon.poolKeepAliveTime();
TimeUnit ttlUnit = ribbon.getPoolKeepAliveTimeUnits();
return connectionPoolFactory.create(maxTotalConnections, timeToLive, ttlUnit);
}
@Bean
@ConditionalOnMissingBean(OkHttpClient.class)
public OkHttpClient client(OkHttpClientFactory httpClientFactory,
ConnectionPool connectionPool, IClientConfig config) {
RibbonProperties ribbon = RibbonProperties.from(config);
this.httpClient = httpClientFactory.createBuilder(false)
.connectTimeout(ribbon.connectTimeout(), TimeUnit.MILLISECONDS)
.readTimeout(ribbon.readTimeout(), TimeUnit.MILLISECONDS)
.followRedirects(ribbon.isFollowRedirects())
.connectionPool(connectionPool).build();
return this.httpClient;
}
@PreDestroy
public void destroy() {
if (httpClient != null) {
httpClient.dispatcher().executorService().shutdown();
httpClient.connectionPool().evictAll();
}
}
}
}
2.3、LoadBalancerAutoConfiguration
* 把标注了@LoadBalanced注解的所有RestTemplate实例，添加一个LoadBalancerInterceptor(loadBalancerClient, requestFactory)
- loadBalancerClient是在RibbonAutoConfiguration中指定的RibbonLoadBalancerClient(springClientFactory()) 类型bean
- requestFactory是本配置类中指定的LoadBalancerRequestFactory类型bean
- @LoadBalanced注解就是@Qualifier类注解，用于匹配bean的标志
@Qualifier
public @interface LoadBalanced {}
* 执行RestTemplates时，会被这个拦截器拦截，
①LoadBalancerRequestFactory会根据传递进来的ClientHttpRequestExecution创建具体通信工具封装后的LoadBalancerRequest
②RestTemplates可以设置ClientHttpRequestFactory，默认为SimpleClientHttpRequestFactory来创建ClientHttpRequest
③LoadBalancerClient负载均衡器会根据serviceName选择合适的路径，进行请求
@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(RestTemplate.class)
@ConditionalOnBean(LoadBalancerClient.class)
@EnableConfigurationProperties(LoadBalancerRetryProperties.class)
public class LoadBalancerAutoConfiguration {
@LoadBalanced
@Autowired(required = false)
private List<RestTemplate> restTemplates = Collections.emptyList();
@Autowired(required = false)
private List<LoadBalancerRequestTransformer> transformers = Collections.emptyList();
//把标注的了@LoadBalanced所有RestTemplate，添加一个LoadBalancerInterceptor(loadBalancerClient, requestFactory)
@Bean
public SmartInitializingSingleton loadBalancedRestTemplateInitializerDeprecated(
final ObjectProvider<List<RestTemplateCustomizer>> restTemplateCustomizers) {
return () -> restTemplateCustomizers.ifAvailable(customizers -> {
for (RestTemplate restTemplate : LoadBalancerAutoConfiguration.this.restTemplates) {
for (RestTemplateCustomizer customizer : customizers) {
customizer.customize(restTemplate);
}
}
});
}
@Bean
@ConditionalOnMissingBean
public LoadBalancerRequestFactory loadBalancerRequestFactory(
LoadBalancerClient loadBalancerClient) {
return new LoadBalancerRequestFactory(loadBalancerClient, this.transformers);
}
@Configuration(proxyBeanMethods = false)
@ConditionalOnMissingClass("org.springframework.retry.support.RetryTemplate")
static class LoadBalancerInterceptorConfig {
@Bean
public LoadBalancerInterceptor ribbonInterceptor(
LoadBalancerClient loadBalancerClient,
LoadBalancerRequestFactory requestFactory) {
return new LoadBalancerInterceptor(loadBalancerClient, requestFactory);
}
@Bean
@ConditionalOnMissingBean
public RestTemplateCustomizer restTemplateCustomizer(
final LoadBalancerInterceptor loadBalancerInterceptor) {
return restTemplate -> {
List<ClientHttpRequestInterceptor> list = new ArrayList<>(
restTemplate.getInterceptors());
list.add(loadBalancerInterceptor);
restTemplate.setInterceptors(list);
};
}
}
//略
}
public class LoadBalancerInterceptor implements ClientHttpRequestInterceptor {
private LoadBalancerClient loadBalancer;
private LoadBalancerRequestFactory requestFactory;
public LoadBalancerInterceptor(LoadBalancerClient loadBalancer,
LoadBalancerRequestFactory requestFactory) {
this.loadBalancer = loadBalancer;
this.requestFactory = requestFactory;
}
public LoadBalancerInterceptor(LoadBalancerClient loadBalancer) {
// for backwards compatibility
this(loadBalancer, new LoadBalancerRequestFactory(loadBalancer));
}
//执行RestTemplates时，会被这个拦截器拦截，
//LoadBalancerRequestFactory会根据传递进来的ClientHttpRequestExecution创建具体通信工具封装后的LoadBalancerRequest
// RestTemplates可以设置ClientHttpRequestFactory，默认为SimpleClientHttpRequestFactory来创建ClientHttpRequest
//LoadBalancerClient负载均衡器会根据serviceName选择合适的路径，进行请求
//
@Override
public ClientHttpResponse intercept(final HttpRequest request, final byte[] body,
final ClientHttpRequestExecution execution) throws IOException {
final URI originalUri = request.getURI();
String serviceName = originalUri.getHost();
return this.loadBalancer.execute(serviceName,
this.requestFactory.createRequest(request, body, execution));
}
3、Name容器中6大可配置组件
3.1、概述
- Spring Ribbon的6大可配置组件类及默认配置
|
自动化配置接口 |
描述 |
默认实现 |
说明 |
|
IClientConfig |
Ribbon的客户端配置 |
com.netflix.client.config.DefaultClientConfigImpl |
对如下组件及其他信息的配置项 |
|
IRule |
Ribbon的负载均衡策略 |
com.netflix.loadbalancer.ZoneAvoidanceRule |
该策略能在多区域环境下选出最佳区域的实例进行访问 |
|
IPing |
Ribbon的实例检查策略 |
com.netflix.loadbalancer.NoOpPing |
该检查策略是一个特殊的实现，实际上它并不会检查实例是否可用，而是始终返回true，默认所有的实例都是可用的 |
|
ServerList<Server> |
服务实例清单维护机制 |
com.netflix.loadbalancer.ConfigurationBasedServerList |
|
|
ServerListFilter<Server> |
服务实例清单过滤机制 |
org.springframework.cloud.netflix.ribbon.ZonePreferenceServerListFilter |
该策略能够优先过滤出与请求调用方处于同一个区域的服务清单 |
|
ILoadBalancer |
（负载均衡器主类） |
com.netflix.loadbalancer.ZoneAwareLoadBalancer |
该策略具备服务感知能力，封装了如上几个组件 |
3.2、IClientConfig配置项组件
3.2.1、IClientConfig接口
* ClientName和NameSpace的获取
这是为了隔离不同的服务调用端的配置
* 其他接口就是属性的增删改查
public interface IClientConfig {
public String getClientName();
public String getNameSpace();
public void loadProperties(String clientName);
public void loadDefaultValues();
public Map<String, Object> getProperties();
public boolean containsProperty(IClientConfigKey key);
public String resolveDeploymentContextbasedVipAddresses();
public int getPropertyAsInteger(IClientConfigKey key, int defaultValue);
public String getPropertyAsString(IClientConfigKey key, String defaultValue);
public boolean getPropertyAsBoolean(IClientConfigKey key, boolean defaultValue);
public <T> T get(IClientConfigKey<T> key);
public <T> T get(IClientConfigKey<T> key, T defaultValue);
public <T> IClientConfig set(IClientConfigKey<T> key, T value);
}
3.2.2、DefaultClientConfigImpl实现类
* 作用是保存了所有的配置项，基于Archaius实现，支持动态的获取配置项的最新值
* 默认实现是DefaultClientConfigImpl，此类作用：
①保存各个组件的默认实现类，比如
public static final String DEFAULT_NFLOADBALANCER_RULE_CLASSNAME = "com.netflix.loadbalancer.AvailabilityFilteringRule";
public static final String DEFAULT_NFLOADBALANCER_CLASSNAME = "com.netflix.loadbalancer.ZoneAwareLoadBalancer";
注：此默认配置在DefaultClientConfigImpl和RibbonClientConfiguration （配置bean加入到独立的服务上下文中）均有指定
②保存各种属性的默认配置，支持动态比如
public static final int DEFAULT_READ_TIMEOUT = 5000;
public static final int DEFAULT_CONNECTION_MANAGER_TIMEOUT = 2000;
public static final int DEFAULT_CONNECT_TIMEOUT = 2000;
* loadProperties会加载环境对象中如下属性，保存到本地缓存Map<String, Object> properties，或者支持动态刷新的Map<String, DynamicStringProperty> dynamicProperties
①从Spring env中加载“ribbon.配置项”这类，作为默认配置
使用 ribbon.<key>=<value> 的形式配置，例如全局配置连接超时时间：
ribbon:
ConnectTimeout: 250
②和加载“client名.ribbon.配置项”这类针对某个Client的配置信息
指定客户端配置方式采用 <client>.ribbon.<key>=<value>，使用样例如下所示，同时，如果同时配置了全局配置和指定客户端配置，那么以指定客户端的配置为准。
EUREKA-CLIENT:
ribbon:
listOfServers: localhost:8001,localhost:8002
* 基本原理（更多原理见Archaius文档）
- 基于Archaius作为数据源根据，对于Springboot项目来说，由ArchaiusAutoConfiguration进行自动配置，这个数据源一般为ConfigurableEnvironmentConfiguration
此AbstractConfiguration实现类，以Spring容器的环境Environmen对象作为数据源进行获取，而不会对其进行反向设置
也会组合其他的数据源，一并配置到ConfigurationManager.install(config);
- 同时也注入了一个监听器ApplicationListener<EnvironmentChangeEvent>
当监听 到环境配置修改时，会获取ConfigurableEnvironmentConfiguration中设置好的自动的动态刷新监听器，执行
listener.configurationChanged(new ConfigurationEvent(source, type,key, value, beforeUpdate));
这样Map<String, Object> properties，或者支持动态刷新的Map<String, DynamicStringProperty> dynamicProperties里对应的值就会被刷新
- config.loadProperties(clientname)方法会从ConfigurableEnvironmentConfiguration中在抽取出指定格式key的键值对作为新的AbstractConfiguration实现类
即SubsetConfiguration以实现数据源的隔离，例如SubsetConfiguration对抽象方法的实现，parent就是原始ConfigurableEnvironmentConfiguration配置
getParentKey(key)会拼接clientname前缀
public void addPropertyDirect(String key, Object value) {
parent.addProperty(getParentKey(key), value);
}
@Bean
@ConditionalOnMissingBean
public IClientConfig ribbonClientConfig() {
DefaultClientConfigImpl config = new DefaultClientConfigImpl();
config.loadProperties(this.name);
config.set(CommonClientConfigKey.ConnectTimeout, DEFAULT_CONNECT_TIMEOUT);
config.set(CommonClientConfigKey.ReadTimeout, DEFAULT_READ_TIMEOUT);
config.set(CommonClientConfigKey.GZipPayload, DEFAULT_GZIP_PAYLOAD);
return config;
}
public class DefaultClientConfigImpl implements IClientConfig {
public static final Boolean DEFAULT_PRIORITIZE_VIP_ADDRESS_BASED_SERVERS = Boolean.TRUE;
public static final String DEFAULT_NFLOADBALANCER_PING_CLASSNAME = "com.netflix.loadbalancer.DummyPing"; // DummyPing.class.getName();
public static final String DEFAULT_NFLOADBALANCER_RULE_CLASSNAME = "com.netflix.loadbalancer.AvailabilityFilteringRule";
public static final String DEFAULT_NFLOADBALANCER_CLASSNAME = "com.netflix.loadbalancer.ZoneAwareLoadBalancer";
public static final boolean DEFAULT_USEIPADDRESS_FOR_SERVER = Boolean.FALSE;
public static final String DEFAULT_CLIENT_CLASSNAME = "com.netflix.niws.client.http.RestClient";
public static final String DEFAULT_VIPADDRESS_RESOLVER_CLASSNAME = "com.netflix.client.SimpleVipAddressResolver";
public static final int DEFAULT_MAX_TOTAL_TIME_TO_PRIME_CONNECTIONS = 30000;
//略
public void loadProperties(String restClientName) {
//启用动态属性 如果开启属性将具有动态 性 这里也是唯一一处将该值设置成true的地方
//所以如果仅仅是默认值 不支持动态属性的
enableDynamicProperties = true;
//设置clientName
setClientName(restClientName);
//加载默认属性
//依赖Archaius获取对应key的配置值。但是 但是有一点需要特别注意。这里的 ConfigurationManager.getConfigInstance().getString()方法获取的配置和
//例如我们配置的值是a,b,c 那么getStringValue获取的值是a,b,c 而 getString()获取的却是 a！ 所以默认值字符串类型的 配置里面出现逗号那就有问题了。
//
loadDefaultValues();
//这里可以看到底层的配置还是通过 Archaius来配置的 所以我们把配置写在classpath的config.properties中是生效的
//对于subset这个方法 举个例子可能会更清楚 例如我们在config文件中的配置是 coredy.ribbon.ReadTimeout
//我们调用subset(”coredy“) 那就会给我们返回所有以coredy开头的Configuration
Configuration props = ConfigurationManager.getConfigInstance().subset(restClientName);
//遍历得到的Configuration 将属性放到全局的配置里面properties
for (Iterator<String> keys = props.getKeys(); keys.hasNext(); ) {
String key = keys.next();
String prop = key;
try {
//指定了NameSpace的情况，默认你为ribbon，需要截断这个前缀
if (prop.startsWith(getNameSpace())) {
prop = prop.substring(getNameSpace().length() + 1);
}
//特别注意这里是使用getStringValue(props, key)来获取值得
//意为着 你配置的属性是a,b,c 那么最终在全局配置properties里面的值也是a,b,c
setPropertyInternal(prop, getStringValue(props, key));
} catch (Exception ex) {
throw new RuntimeException(String.format("Property %s is invalid", prop));
}
}
//略
}
3.3、ServerList<Server>获取指定serviceId服务列表组件
3.3.1、ServerList<T extends Server>接口
* 定义获取指定serviceId服务列表的方法
public interface ServerList<T extends Server> {
//初始可用的服务列表
public List<T> getInitialListOfServers();
//经过ping进行心跳正常过滤后的最新可用的服务列表
public List<T> getUpdatedListOfServers();
}
* 一个Server对象表示一个服务实例的信息
public class Server {
public static final String UNKNOWN_ZONE = "UNKNOWN";
private String host;
private int port = 80;
private String scheme;
private volatile String id;//通常为 http域名+端口
private volatile boolean isAliveFlag;
private String zone = UNKNOWN_ZONE;
private volatile boolean readyToServe = true;
private MetaInfo simpleMetaInfo = new MetaInfo() {
/**
* 服务实例对应服务器的名称和服务组
*/
@Override
public String getAppName() {
return null;
}
@Override
public String getServerGroup() {
return null;
}
/**
* 服务实例的别名
*/
@Override
public String getServiceIdForDiscovery() {
return null;
}
@Override
public String getInstanceId() {
return id;
}
};
public Server(String host, int port) {
this(null, host, port);
}
public Server(String scheme, String host, int port) {
this.scheme = scheme;
this.host = host;
this.port = port;
this.id = host + ":" + port;
isAliveFlag = false;
}
//..
}
3.3.2、NacosServerList
@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;
}
* 继承AbstractServerList，这里值初始化了过滤器实现类
- 优先取NIWSServerListFilterClassName配置的，
- 如果为null，取NIWSServerListFilterClassName类
- 在RibbonClientConfiguration自动配置类中，默认配置为ZonePreferenceServerListFilter
* NacosServerList实现类
- 逻辑简单，直接借助Nacos提供的客户端NacosDiscoveryProperties 获取服务列表
List<Instance> instances = discoveryProperties.namingServiceInstance().selectInstances(serviceId, true);
- 再适配为ribbon需要的Server类型即可
- 一个iClientConfig.getClientName()，即一个ServiceId对应一个NacosServerList
public class NacosServerList extends AbstractServerList<NacosServer> {
private NacosDiscoveryProperties discoveryProperties;
private String serviceId;
public NacosServerList(NacosDiscoveryProperties discoveryProperties) {
this.discoveryProperties = discoveryProperties;
}
@Override
public List<NacosServer> getInitialListOfServers() {
return getServers();
}
@Override
public List<NacosServer> getUpdatedListOfServers() {
return getServers();
}
private List<NacosServer> getServers() {
try {
List<Instance> instances = discoveryProperties.namingServiceInstance()
.selectInstances(serviceId, true);
return instancesToServerList(instances);
}
catch (Exception e) {
throw new IllegalStateException(
"Can not get service instances from nacos, serviceId=" + serviceId,
e);
}
}
private List<NacosServer> instancesToServerList(List<Instance> instances) {
List<NacosServer> result = new ArrayList<>();
if (null == instances) {
return result;
}
for (Instance instance : instances) {
result.add(new NacosServer(instance));
}
return result;
}
public String getServiceId() {
return serviceId;
}
@Override
public void initWithNiwsConfig(IClientConfig iClientConfig) {
this.serviceId = iClientConfig.getClientName();
}
}
3.4、ServerListFilter<T extends Server>服务列表过滤组件
3.4.1、ServerListFilter接口
public interface ServerListFilter<T extends Server> {
public List<T> getFilteredListOfServers(List<T> servers);
}
3.4.2、ZoneAffinityServerListFilter实现类
* getFilteredListOfServers的实现逻辑，分两步过滤
- 初步过滤：
对指定serverId下的所有实例列表默认使用ZoneAffinityPredicate使用过滤：过滤配置指定zone的服务实例
一般的服务发现客户端会配置好这个zone值
- 对初步过滤的结果再次判断
①配置zoneAffinity（默认flase）和zoneExclusive（默认flase）均没有true时，不过滤zone，选择所有实例列表
②当zoneExclusive为true，使用初步过滤的（指定zone过滤）
③当zoneAffinity为true，表示需要再次过滤判断：基于LoadBalancerStats判断这些初步过滤的服务实例的状态是否符合配置的阈值
符合就使用过滤的，否则不过滤
((double) circuitBreakerTrippedCount) / instanceCount >= blackOutServerPercentageThreshold.get()
|| loadPerServer >= activeReqeustsPerServerThreshold.get()
|| (instanceCount - circuitBreakerTrippedCount) < availableServersThreshold.get())
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();
private static Logger logger = LoggerFactory.getLogger(ZoneAffinityServerListFilter.class);
String zone;
public ZoneAffinityServerListFilter() {
}
public ZoneAffinityServerListFilter(IClientConfig niwsClientConfig) {
initWithNiwsConfig(niwsClientConfig);
}
//从配置组件中获取属性
@Override
public void initWithNiwsConfig(IClientConfig niwsClientConfig) {
String sZoneAffinity = "" + niwsClientConfig.getProperty(CommonClientConfigKey.EnableZoneAffinity, false);
if (sZoneAffinity != null){
zoneAffinity = Boolean.parseBoolean(sZoneAffinity);
}
String sZoneExclusive = "" + niwsClientConfig.getProperty(CommonClientConfigKey.EnableZoneExclusivity, false);
if (sZoneExclusive != null){
zoneExclusive = Boolean.parseBoolean(sZoneExclusive);
}
if (ConfigurationManager.getDeploymentContext() != null) {
zone = ConfigurationManager.getDeploymentContext().getValue(ContextKey.zone);
}
activeReqeustsPerServerThreshold = DynamicPropertyFactory.getInstance().getDoubleProperty(niwsClientConfig.getClientName() + "." + niwsClientConfig.getNameSpace() + ".zoneAffinity.maxLoadPerServer", 0.6d);
blackOutServerPercentageThreshold = DynamicPropertyFactory.getInstance().getDoubleProperty(niwsClientConfig.getClientName() + "." + niwsClientConfig.getNameSpace() + ".zoneAffinity.maxBlackOutServesrPercentage", 0.8d);
availableServersThreshold = DynamicPropertyFactory.getInstance().getIntProperty(niwsClientConfig.getClientName() + "." + niwsClientConfig.getNameSpace() + ".zoneAffinity.minAvailableServers", 2);
overrideCounter = Monitors.newCounter("ZoneAffinity_OverrideCounter");
Monitors.registerObject("NIWSServerListFilter_" + niwsClientConfig.getClientName());
}
@Override
public List<T> getFilteredListOfServers(List<T> servers) {
if (zone != null && (zoneAffinity || zoneExclusive) && servers !=null && servers.size() > 0){
//初步过滤：使用指定zone过滤的
List<T> filteredServers = Lists.newArrayList(Iterables.filter(
servers, this.zoneAffinityPredicate.getServerOnlyPredicate()));
//再次过滤判断，符合才使用初步过滤的，否则放行所有
if (shouldEnableZoneAffinity(filteredServers)) {
return filteredServers;
} else if (zoneAffinity) {
overrideCounter.increment();
}
}
return servers;
}
//zoneAffinity和zoneExclusive均没有true时，过滤
//当zoneExclusive为true，使用初步过滤的（指定zone过滤）
//当zoneAffinity为true，表示需要再次过滤：基于LoadBalancerStats判断这些初步过滤的服务实例的状态是否符合配置的阈值
// 符合就使用过滤的，否则不过滤
private boolean shouldEnableZoneAffinity(List<T> filtered) {
if (!zoneAffinity && !zoneExclusive) {
return false;
}
if (zoneExclusive) {
return true;
}
LoadBalancerStats stats = getLoadBalancerStats();
if (stats == null) {
return zoneAffinity;
} else {
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;
}
}
}
}
3.4.3、ZonePreferenceServerListFilter实现类
@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;
}
* 在RibbonClientConfiguration自动配置中，默认实现为ZonePreferenceServerListFilter
* ZonePreferenceServerListFilter继承ZoneAffinityServerListFilter
当ZoneAffinityServerListFilter过滤失败（即调用父类过滤后，的实例数不变）)时，使用指定zone（如有）进行过滤
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<>();
for (Server server : output) {
if (this.zone.equalsIgnoreCase(server.getZone())) {
local.add(server);
}
}
if (!local.isEmpty()) {
return local;
}
}
return output;
}
}
3.5、IPing连通性检测组件
//各个组件的默认实现
@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();
}
* 默认的逻辑始终为true,即不会进行ping处理
这时为了性能考虑，不需要主动的去ping每一个实例的连通性，因为在ZoneAwareLoadBalancer中会进行过滤，
如果实例有问题，那么必定存在socket相关的异常，那么就会触发此实例的断路，
public interface IPing {
public boolean isAlive(Server server);
}
public class DummyPing extends AbstractLoadBalancerPing {
public DummyPing() {
}
public boolean isAlive(Server server) {
return true;
}
@Override
public void initWithNiwsConfig(IClientConfig clientConfig) {
}
}
3.6、IRule选取规则组件
3.6.1、IRule接口
由接口方法即可知道：核心方法为choose选择一个serverId下的一个服务实例，而且逻辑或基于ILoadBalancer实现类
public interface IRule{
public Server choose(Object key);
public void setLoadBalancer(ILoadBalancer lb);
public ILoadBalancer getLoadBalancer();
}
3.6.2、ZoneAvoidanceRule实现类
3.6.2.1、主要流程
@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;
}
* 是RibbonAutoConfiguration指定的默认规则
* 其核心逻辑是：当前serviceId下的每一个服务实例经过其内配置的compositePredicate进行过滤后的候选服务实例集合，再进行轮询即可
注：每一服务实例都需要走一遍下面的流程，比如ZoneAvoidanceRule.getAvailableZones方法会调用多此（其实没必要）
- ZoneAvoidancePredicate校验
①基于LoadBalancerStats获取当前服务实例对应的zoneSnapshot
使用ZoneAvoidanceRule.getAvailableZones计算出可用有效的分区availableZones，再进行下一步AvailabilityPredicate的校验
getAvailableZones具体算法见下
②如果此服务实例对应的zone或者zoneSnapshot 不存在，那么会直接放行，进行下一步AvailabilityPredicate的校验
- AvailabilityPredicate校验
①所有实例不处于断路状态
②所有实例的活跃请求数小于niws.loadbalancer.availabilityFilteringRule.activeConnectionsLimit配置，默认为int最大值
- 两个Predicate校验都满足后，最为候选加入List<Server> eligible
如果最终候选为0，那么再遍历一次，符合AvailabilityPredicate这个校验即可
如果二次校验AvailabilityPredicate还是失败，那么默认全部实例放行
- 对最终结果List<Server> eligible进行轮询
public abstract class PredicateBasedRule extends ClientConfigEnabledRoundRobinRule {
public abstract AbstractServerPredicate getPredicate();
@Override
public Server choose(Object key) {
ILoadBalancer lb = getLoadBalancer();
//getPredicate()就是子类ZoneAvoidanceRule 的CompositePredicate compositePredicate
Optional<Server> server = getPredicate().chooseRoundRobinAfterFiltering(lb.getAllServers(), key);
if (server.isPresent()) {
return server.get();
} else {
return null;
}
}
}
public class ZoneAvoidanceRule extends PredicateBasedRule {
private static final Random random = new Random();
private CompositePredicate compositePredicate;
public ZoneAvoidanceRule() {
super();
ZoneAvoidancePredicate zonePredicate = new ZoneAvoidancePredicate(this);
AvailabilityPredicate availabilityPredicate = new AvailabilityPredicate(this);
compositePredicate = createCompositePredicate(zonePredicate, availabilityPredicate);
}
private CompositePredicate createCompositePredicate(ZoneAvoidancePredicate p1, AvailabilityPredicate p2) {
return CompositePredicate.withPredicates(p1, p2)
.addFallbackPredicate(p2)
.addFallbackPredicate(AbstractServerPredicate.alwaysTrue())
.build();
}
//..
}
//CompositePredicate的父类，其中chooseRoundRobinAfterFiltering如下
public abstract class AbstractServerPredicate implements Predicate<PredicateKey> {
protected IRule rule;
private volatile LoadBalancerStats lbStats;
private final Random random = new Random();
private final AtomicInteger nextIndex = new AtomicInteger();
private final Predicate<Server> serverOnlyPredicate = new Predicate<Server>() {
@Override
public boolean apply(@Nullable Server input) {
return AbstractServerPredicate.this.apply(new PredicateKey(input));
}
};
public static AbstractServerPredicate alwaysTrue() {
return new AbstractServerPredicate() {
@Override
public boolean apply(@Nullable PredicateKey input) {
return true;
}
};
}
//同理：在过滤之后进行随机
public Optional<Server> chooseRandomlyAfterFiltering(List<Server> servers) {
List<Server> eligible = getEligibleServers(servers);
if (eligible.size() == 0) {
return Optional.absent();
}
return Optional.of(eligible.get(random.nextInt(eligible.size())));
}
//在过滤之后进行随机轮询
public Optional<Server> chooseRoundRobinAfterFiltering(List<Server> servers) {
List<Server> eligible = getEligibleServers(servers);
if (eligible.size() == 0) {
return Optional.absent();
}
return Optional.of(eligible.get(incrementAndGetModulo(eligible.size())));
}
public List<Server> getEligibleServers(List<Server> servers) {
return getEligibleServers(servers, null);
}
//遍历每一服务实例，使用子类的apply进行校验过滤，返回符合的候选集合
public List<Server> getEligibleServers(List<Server> servers, Object loadBalancerKey) {
if (loadBalancerKey == null) {
return ImmutableList.copyOf(Iterables.filter(servers, this.getServerOnlyPredicate()));
} else {
List<Server> results = Lists.newArrayList();
for (Server server: servers) {
if (this.apply(new PredicateKey(loadBalancerKey, server))) {
results.add(server);
}
}
return results;
}
}
//轮询
private int incrementAndGetModulo(int modulo) {
for (;;) {
int current = nextIndex.get();
int next = (current + 1) % modulo;
if (nextIndex.compareAndSet(current, next) && current < modulo)
return current;
}
}
//...
}
public class ZoneAvoidancePredicate extends AbstractServerPredicate {
private volatile DynamicDoubleProperty triggeringLoad = new DynamicDoubleProperty("ZoneAwareNIWSDiscoveryLoadBalancer.triggeringLoadPerServerThreshold", 0.2d);
private volatile DynamicDoubleProperty triggeringBlackoutPercentage = new DynamicDoubleProperty("ZoneAwareNIWSDiscoveryLoadBalancer.avoidZoneWithBlackoutPercetage", 0.99999d);
private static final DynamicBooleanProperty ENABLED = DynamicPropertyFactory
.getInstance().getBooleanProperty(
"niws.loadbalancer.zoneAvoidanceRule.enabled", true);
public ZoneAvoidancePredicate(LoadBalancerStats lbStats,
IClientConfig clientConfig) {
super(lbStats, clientConfig);
initDynamicProperties(clientConfig);
}
private void initDynamicProperties(IClientConfig clientConfig) {
if (clientConfig != null) {
triggeringLoad = DynamicPropertyFactory.getInstance().getDoubleProperty(
"ZoneAwareNIWSDiscoveryLoadBalancer." + clientConfig.getClientName() + ".triggeringLoadPerServerThreshold", 0.2d);
triggeringBlackoutPercentage = DynamicPropertyFactory.getInstance().getDoubleProperty(
"ZoneAwareNIWSDiscoveryLoadBalancer." + clientConfig.getClientName() + ".avoidZoneWithBlackoutPercetage", 0.99999d);
}
}
@Override
public boolean apply(@Nullable PredicateKey input) {
if (!ENABLED.get()) {
return true;
}
String serverZone = input.getServer().getZone();
if (serverZone == null) {
return true;
}
LoadBalancerStats lbStats = getLBStats();
if (lbStats == null) {
return true;
}
if (lbStats.getAvailableZones().size() <= 1) {
return true;
}
Map<String, ZoneSnapshot> zoneSnapshot = ZoneAvoidanceRule.createSnapshot(lbStats);
if (!zoneSnapshot.keySet().contains(serverZone)) {
return true;
}
Set<String> availableZones = ZoneAvoidanceRule.getAvailableZones(zoneSnapshot, triggeringLoad.get(), triggeringBlackoutPercentage.get());
if (availableZones != null) {
return availableZones.contains(input.getServer().getZone());
} else {
return false;
}
}
}
public class AvailabilityPredicate extends AbstractServerPredicate {
private static final DynamicBooleanProperty CIRCUIT_BREAKER_FILTERING =
DynamicPropertyFactory.getInstance().getBooleanProperty("niws.loadbalancer.availabilityFilteringRule.filterCircuitTripped", true);
private static final DynamicIntProperty ACTIVE_CONNECTIONS_LIMIT =
DynamicPropertyFactory.getInstance().getIntProperty("niws.loadbalancer.availabilityFilteringRule.activeConnectionsLimit", Integer.MAX_VALUE);
private ChainedDynamicProperty.IntProperty activeConnectionsLimit = new ChainedDynamicProperty.IntProperty(ACTIVE_CONNECTIONS_LIMIT);
public AvailabilityPredicate(LoadBalancerStats lbStats, IClientConfig clientConfig) {
super(lbStats, clientConfig);
initDynamicProperty(clientConfig);
}
private void initDynamicProperty(IClientConfig clientConfig) {
String id = "default";
if (clientConfig != null) {
id = clientConfig.getClientName();
activeConnectionsLimit = new ChainedDynamicProperty.IntProperty(id + "." + clientConfig.getNameSpace() + ".ActiveConnectionsLimit", ACTIVE_CONNECTIONS_LIMIT);
}
}
@Override
public boolean apply(@Nullable PredicateKey input) {
LoadBalancerStats stats = getLBStats();
if (stats == null) {
return true;
}
return !shouldSkipServer(stats.getSingleServerStat(input.getServer()));
}
private boolean shouldSkipServer(ServerStats stats) {
if ((CIRCUIT_BREAKER_FILTERING.get() && stats.isCircuitBreakerTripped())
|| stats.getActiveRequestsCount() >= activeConnectionsLimit.get()) {
return true;
}
return false;
}
}
3.6.2.2、getAvailableZones获取有效分区
getAvailableZones(Map<String, ZoneSnapshot> snapshot, double triggeringLoad, double triggeringBlackoutPercentage)
* 参数：
①Map<String, ZoneSnapshot> snapshot每一个分区的快照，是基于LoadBalancerStats获取的
对一个分区中的所有服务实例进行计算后封装为new ZoneSnapshot(instanceCount, circuitBreakerTrippedCount, activeConnectionsCount, loadPerServer)
- instanceCount：当前服务数
- circuitBreakerTrippedCount：当前滑动窗口内断路的服务数
- activeConnectionsCount：当前滑动窗口内此分区所有服务实例的有效连接总数数
- LoadPerServer = activeConnectionsCount / (instanceCount - circuitBreakerTrippedCount)--此值越大表示，此分区剩余可用连接越少，负载就越大
②triggeringLoad和triggeringBlackoutPercentage都是配置的阈值
triggeringLoadPerServerThreshold，默认为0.2
avoidZoneWithBlackoutPercetage，默认为0.99999d
* getAvailableZones流程
①获取Set<String> availableZones当前可用的候选的Zones集合，取Map<String, ZoneSnapshot> snapshot的keys
②获取配置值
triggeringLoadPerServerThreshold，默认为0.2
avoidZoneWithBlackoutPercetage，默认为0.99999d
③去除服务实例数为0的分区
④去除 loadPerServer小于0 或者 断路数/服务实例数>= avoidZoneWithBlackoutPercetage的分区
⑤计算LoadPerServer 最大的几个分区到worstZones集合，这个worstZones集合最大LoadPerServer 和最下LoadPerServer 相差必须小于0.000001d，
如果最大的LoadPerServer 小于triggeringLoadPerServerThreshold且上面步骤没有排除过任何一个分区不存在，那么就直接availableZones，而不用过滤worstZones
⑥对worstZones进行过滤，
获取在worstZones中所有分区的实例数总数，在此数范围内随机选择一个数，遍历worstZones，第一个实例数小于此数的分区为选中服务
从availableZones中去除此分区
⑦最终得到Set<String> availableZones
static Map<String, ZoneSnapshot> createSnapshot(LoadBalancerStats lbStats) {
Map<String, ZoneSnapshot> map = new HashMap<String, ZoneSnapshot>();
for (String zone : lbStats.getAvailableZones()) {
ZoneSnapshot snapshot = lbStats.getZoneSnapshot(zone);
map.put(zone, snapshot);
}
return map;
}
//LoadBalancerStats
public ZoneSnapshot getZoneSnapshot(List<? extends Server> servers) {
if (servers == null || servers.size() == 0) {
return new ZoneSnapshot();
}
int instanceCount = servers.size();
int activeConnectionsCount = 0;
int activeConnectionsCountOnAvailableServer = 0;
int circuitBreakerTrippedCount = 0;
double loadPerServer = 0;
long currentTime = System.currentTimeMillis();
for (Server server: servers) {
ServerStats stat = getSingleServerStat(server);
if (stat.isCircuitBreakerTripped(currentTime)) {
circuitBreakerTrippedCount++;
} else {
activeConnectionsCountOnAvailableServer += stat.getActiveRequestsCount(currentTime);
}
activeConnectionsCount += stat.getActiveRequestsCount(currentTime);
}
if (circuitBreakerTrippedCount == instanceCount) {
if (instanceCount > 0) {
// should be NaN, but may not be displayable on Epic
loadPerServer = -1;
}
} else {
loadPerServer = ((double) activeConnectionsCountOnAvailableServer) / (instanceCount - circuitBreakerTrippedCount);
}
return new ZoneSnapshot(instanceCount, circuitBreakerTrippedCount, activeConnectionsCount, loadPerServer);
}
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();
if (instanceCount == 0) {
availableZones.remove(zone);
limitedZoneAvailability = true;
} else {
double loadPerServer = zoneSnapshot.getLoadPerServer();
if (((double) zoneSnapshot.getCircuitTrippedCount())
/ instanceCount >= triggeringBlackoutPercentage
|| loadPerServer < 0) {
availableZones.remove(zone);
limitedZoneAvailability = true;
} else {
if (Math.abs(loadPerServer - maxLoadPerServer) < 0.000001d) {
// they are the same considering double calculation
// round error
worstZones.add(zone);
} else if (loadPerServer > maxLoadPerServer) {
maxLoadPerServer = loadPerServer;
worstZones.clear();
worstZones.add(zone);
}
}
}
}
if (maxLoadPerServer < triggeringLoad && !limitedZoneAvailability) {
// zone override is not needed here
return availableZones;
}
String zoneToAvoid = randomChooseZone(snapshot, worstZones);
if (zoneToAvoid != null) {
availableZones.remove(zoneToAvoid);
}
return availableZones;
}
static String randomChooseZone(Map<String, ZoneSnapshot> snapshot,
Set<String> chooseFrom) {
if (chooseFrom == null || chooseFrom.size() == 0) {
return null;
}
String selectedZone = chooseFrom.iterator().next();
if (chooseFrom.size() == 1) {
return selectedZone;
}
int totalServerCount = 0;
for (String zone : chooseFrom) {
totalServerCount += snapshot.get(zone).getInstanceCount();
}
int index = random.nextInt(totalServerCount) + 1;
int sum = 0;
for (String zone : chooseFrom) {
sum += snapshot.get(zone).getInstanceCount();
if (index <= sum) {
selectedZone = zone;
break;
}
}
return selectedZone;
}
3.7、ILoadBalancer总控组件
3.7.1、ILoadBalancer接口
接口作用是服务实例的增改查
public interface ILoadBalancer {
//添加服务实例
public void addServers(List<Server> newServers);
//选择一个服务实例
public Server chooseServer(Object key);
//LB这边把一个指定的服务标为下线
public void markServerDown(Server server);
@Deprecated
public List<Server> getServerList(boolean availableOnly);
//只返回可用的服务实例
public List<Server> getReachableServers();
//所有服务实例
public List<Server> getAllServers();
}
3.7.1.1、接口作用
* 是负载均衡器主类ILoadBalancer接口的一个实现，ILoadBalancer是面向客户调用的接口
该策略具备复制均衡的所有能力，因为封装了必要的几个组件
* 一个serviceId维度有一个自己的ILoadBalancer实现类
* 在RibbonClientConfiguration 中的默认配置为ZoneAwareLoadBalancer，从构造方法看，这个类封装所有相关的组件
@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);
}
①IClientConfig：Ribbon的客户端配置，对如下组件及其他信息的配置，默认为DefaultClientConfigImpl
②IRule：Ribbon的负载均衡策略，该策略能在多区域环境下选出最佳区域的实例进行访问，默认为ZoneAvoidanceRule
③IPing Ribbon：实例连通性检查策略，默认NoOpPing该检查策略是一个特殊的实现，实际上它并不会检查实例是否可用，而是始终返回true，默认所有的实例都是可用的
④ServerList<Server> ：服务实例清单维护机制，默认为.ConfigurationBasedServerList
⑤ServerListFilter<Server>：服务实例清单过滤机制，默认为ZonePreferenceServerListFilter 该策略能够优先过滤出与请求调用方处于同一个区域的服务清单
⑥ServerListUpdater：进行动态服务列表的更新，默认为PollingServerListUpdater，这时一个定时拉取原始服务列表的策略
3.7.1.2、接口体系
①ILoadBalancer接口的，定义了service实例的增删查改
如addServers(List<Server> newServers)、chooseServer(Object key)等
②BaseLoadBlance，实现了整体的功能，如下
- 抽象父类AbstractLoadBalancer，仅仅是定义了枚举ServerGroup：ALL,STATUS_UP,STATUS_NOT_UP
- 两个List保存此服务的可用的所有的服务实例列表，以读写锁来处理线程安全
此serviceId的所有的服务实例：List<Server> allServerList = Collections .synchronizedList(new ArrayList<Server>());
此serviceId的上线的服务实例：List<Server> upServerList = Collections.synchronizedList(new ArrayList<Server>());
- 通过IPing定时的对服务实例列表的可用性进行检查
- 通过IRule实现ILoadBalancer接口chooseServer方法以选择指定策略的服务实例
- 通过LoadBalancerStats记录了每个服务实例的状态ServerStats和每个区域的状态zoneStatsMap，
包括访问次数、响应时间、熔断（时间）控制等等，通过zone划分了服务实例
③LoadBalancerContext
- 此类的作用类似一个记录工具类，所有方法都需要指定LoadBalancerStats的ServerStats参数
- 穿插在请调用的整个流程，提供记录和更新服务状态ServerStats的功能
比如在打开请求时，会调用noteOpenConnectio(ServerStats serverStats)n方法，此方法会把对应的ServerStats的活跃请求指标activeRequestsCount + 1
④DynamicServerListLoadBalancer
- 从指定的ServerList<T>实现类，如Eruka的DiscoveryEnabledNIWSServerList实现类，会获取所有服务列表，执行BaseLoadBlance父类的setServersList
就会写入BaseLoadBlance的两个List保存此服务的可用的所有的服务实例列表
- 基于服务列表构建好LoadBalancerStats的两个以Zone划分的Map
Map<String, ZoneStats> zoneStatsMap ;
Map<String, List<? extends Server>> upServerListZoneMap ;
- 会基于关联的ServerListUpdater进行动态服务列表的更新，默认为PollingServerListUpdater，这时一个定时拉取默认1秒)原始服务列表的策略
会从serverListImpl.getUpdatedListOfServers()获取最新的数据，使用ServerListFilter.getFilteredListOfServers(servers)过滤以下，调用setServersList会更新各种地表
BaseLoadBlance的allServerList 和upServerList
DynamicServerListLoadBalancer的Map<String, List<? extends Server>> upServerListZoneMap
ZoneAwareLoadBalancer的每个zone对应的BaseLoadBalancer的allServerList 和upServerList
拉取到最新的服务列表后，重写走上面④的流程
⑤ZoneAwareLoadBalancer
- 为每一个分区创建一个BaseLoadBlance，即ConcurrentHashMap<String, BaseLoadBalancer> balancers
主要是BaseLoadBalancer类型，其内的
- 重写chooseServer方法
如果只有一个分区，使用super.chooseServer(key)；如果有多个分区，先通过LoadBalancerStats获取所有分区的状态快照，
使用ZoneAvoidanceRule.getAvailableZones计算出可用有效的分区
再使用ZoneAvoidanceRule.randomChooseZone随机选择一个分区，使用balancers 表对应BaseLoadBalancer的chooseServer选择最终的服务实例
- 注：此分区过滤的功能最终基于ZoneAvoidanceRule进行选择的逻辑是部分重复的，那么这里为什么还需要进行重写
- 用户这可以自定义的ZoneAwareLoadBalancer的rule，这样ZoneAwareLoadBalancer会帮我们先过滤好分区，然后选择自己设置的rule进行处理
这样就复用了分区的功能
例如可以使用WeightedResponseTimeRule以平均响应时间作为权重进行随机选择
3.7.2、BaseLoadBlance
3.7.2.1、提供整体的功能
实现了整体的功能，如下
* 在构造方法在对主要的成员属性进行初始化
- config = clientConfig;
- setRule(rule);
- setPing(ping);
- setLoadBalancerStats(stats); //一般会new LoadBalancerStats(clientConfig.getClientName());
- 设置pingIntervalTime默认30s、maxTotalPingTime默认2
其中一个间隔pingIntervalTime的PingTask
* 维护存此服务的可用的所有的服务实例列表synchronizedList(new ArrayList<Server>()) List<Server> allServerList和upServerList，
- 其中一个重要的方法setServersList(List<Server> lsrv)用于新增或者初始化服务实例列表，流程如下
①写锁加锁writeLock.lock
②遍历参数List<Server> lsr，添加到临时List allServers
③和原allServerList比较，如果个数不同，通知配置List<ServerListChangeListener> changeListeners监听器，触发serverListChanged(oldList, newList)方法
④是否与实例提前创建连接，在isEnablePrimingConnections为true时（默认为false），使用PrimeConnections对象
会为每一个服务实例创建执行一个异步Future<Boolean>会去访问实例服务器，如果成功值为true，否则为false
⑤复制到最新的服务列表allServerList
如果ping == null || ping.getClass().getName().equals(DummyPing.class.getName())，也直接赋值到upServerList
注：这里直接覆盖没有关系，因为server重写了equal方法： svc.getId().equals(this.getId());
⑥写锁解锁writeLock.unlock();
- 线程安全保障
底表allServerList和upServerList是synchronizedList
获取allServerList和upServerList信息时，一般就是直接返回两个集合成员，并使用 Collections.unmodifiableList
setServersList以读写锁来处理线程安全，是为了与Pinger时获取待检查列表，进行安全控制
进一步，在DynamicServerListLoadBalancer使用的Map<String, List<? extends Server>> upServerListZoneMap在setServersList里会同步更新，获取是是直接访问upServerListZoneMap
* 通过IPing实现类定时的对服务实例列表的可用性进行检查
- 默认始终返回true，对于Eureka的实现类为NIWSDiscoveryPing
- 定时或者快速执行的PingTask任务为 new Pinger(pingStrategy).runPinger();，其中pingStrategy为SerialPingStrategy，流程如下
注：ping == null || ping.getClass().getName().equals(DummyPing.class.getName())时此任务不执行
①加锁的获取allServerList的快照集合
②使用pingStrategy的pingServers(快照集合)，获取每一个实例的ping结果，封装为result[]
③用result[]最新的isAlive结果覆盖原allServerList的isAlive状态，其中刷选为true的还赋值到upServerList
* 通过IRule实现chooseServer方法以选择指定策略的服务实例
rule.choose(key);
* 保存对应的LoadBalancerStats
- 此值可以是构造方法传递，如果没有传递使用IClientConfig进行默认创建。一般会new LoadBalancerStats(clientConfig.getClientName());
- 记录了每个服务实例的状态ServerStats和每个区域的状态zoneStatsMap，包括访问次数、响应时间、熔断（时间）控制等等，
通过zone划分了服务实例
public class BaseLoadBalancer extends AbstractLoadBalancer implements PrimeConnections.PrimeConnectionListener, IClientConfigAware {
private final static IRule DEFAULT_RULE = new RoundRobinRule();
private final static SerialPingStrategy DEFAULT_PING_STRATEGY = new SerialPingStrategy();
private static final String DEFAULT_NAME = "default";
private static final String PREFIX = "LoadBalancer_";
protected IRule rule = DEFAULT_RULE;
protected IPingStrategy pingStrategy = DEFAULT_PING_STRATEGY;
protected IPing ping = null;
protected String name = DEFAULT_NAME;
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();
protected Timer lbTimer = null;
protected int pingIntervalSeconds = 10;
protected int maxTotalPingTimeSeconds = 5;
protected Comparator<Server> serverComparator = new ServerComparator();
protected AtomicBoolean pingInProgress = new AtomicBoolean(false);
protected LoadBalancerStats lbStats;
private IClientConfig config;
private volatile Counter counter = Monitors.newCounter("LoadBalancer_ChooseServer");
private PrimeConnections primeConnections;
private volatile boolean enablePrimingConnections = false;
private List<ServerListChangeListener> changeListeners = new CopyOnWriteArrayList<ServerListChangeListener>();
private List<ServerStatusChangeListener> serverStatusListeners = new CopyOnWriteArrayList<ServerStatusChangeListener>();
//配置默认组件，进行ping都是任务的启动
public BaseLoadBalancer() {
this.name = DEFAULT_NAME;
this.ping = null;
setRule(DEFAULT_RULE);
setupPingTask();
lbStats = new LoadBalancerStats(DEFAULT_NAME);
}
//其他构造器，略
public BaseLoadBalancer(IClientConfig config) {
initWithNiwsConfig(config);
}
public BaseLoadBalancer(IClientConfig config, IRule rule, IPing ping) {
initWithConfig(config, rule, ping, createLoadBalancerStatsFromConfig(config));
}
//根据配置IClientConfig类进行构造
void initWithConfig(IClientConfig clientConfig, IRule rule, IPing ping, LoadBalancerStats stats) {
this.config = clientConfig;
String clientName = clientConfig.getClientName();
this.name = clientName;
int pingIntervalTime = Integer.parseInt(""+ clientConfig.getProperty(CommonClientConfigKey.NFLoadBalancerPingInterval, Integer.parseInt("30")));
int maxTotalPingTime = Integer.parseInt(""+ clientConfig.getProperty(CommonClientConfigKey.NFLoadBalancerMaxTotalPingTime,Integer.parseInt("2")));
setPingInterval(pingIntervalTime);
setMaxTotalPingTime(maxTotalPingTime);
setRule(rule);
setPing(ping);
setLoadBalancerStats(stats);
rule.setLoadBalancer(this);
if (ping instanceof AbstractLoadBalancerPing) {
((AbstractLoadBalancerPing) ping).setLoadBalancer(this);
}
boolean enablePrimeConnections = clientConfig.get(CommonClientConfigKey.EnablePrimeConnections, DefaultClientConfigImpl.DEFAULT_ENABLE_PRIME_CONNECTIONS);
if (enablePrimeConnections) {
this.setEnablePrimingConnections(true);
PrimeConnections primeConnections = new PrimeConnections(
this.getName(), clientConfig);
this.setPrimeConnections(primeConnections);
}
init();
}
//setter/setter略
public void addServerListChangeListener(ServerListChangeListener listener) {
changeListeners.add(listener);
}
public void removeServerListChangeListener(ServerListChangeListener listener) {
changeListeners.remove(listener);
}
public void addServerStatusChangeListener(ServerStatusChangeListener listener) {serverStatusListeners.add(listener);}
public void removeServerStatusChangeListener(ServerStatusChangeListener listener) {serverStatusListeners.remove(listener); }
public IClientConfig getClientConfig() {
return config;
}
void setupPingTask() {
//Iping为空或者为DummyPing实现类，可跳过
if (canSkipPing()) {return;}
if (lbTimer != null) {
lbTimer.cancel();
}
lbTimer = new ShutdownEnabledTimer("NFLoadBalancer-PingTimer-" + name, true);
lbTimer.schedule(new PingTask(), 0, pingIntervalSeconds * 1000);
//手动触发一下PingTask
forceQuickPing();
}
//添加服务
@Override
public void addServers(List<Server> newServers) {
if (newServers != null && newServers.size() > 0) {
try {
ArrayList<Server> newList = new ArrayList<Server>();
newList.addAll(allServerList);
newList.addAll(newServers);
setServersList(newList);
} catch (Exception e) { }
}
}
//使用写锁，锁住allServerLock
public void setServersList(List lsrv) {
Lock writeLock = allServerLock.writeLock();
ArrayList<Server> newServers = new ArrayList<Server>();
//添加写锁
writeLock.lock();
try {
ArrayList<Server> allServers = new ArrayList<Server>();
for (Object server : lsrv) {
if (server == null) {continue;}
if (server instanceof String) {server = new Server((String) server);}
if (server instanceof Server) {
allServers.add((Server) server);
}
else {throw new IllegalArgumentException();
}
}
//触发器处理
boolean listChanged = false;
if (!allServerList.equals(allServers)) {
listChanged = true;
if (changeListeners != null && changeListeners.size() > 0) {
List<Server> oldList = ImmutableList.copyOf(allServerList);
List<Server> newList = ImmutableList.copyOf(allServers);
for (ServerListChangeListener l: changeListeners) {
try {
l.serverListChanged(oldList, newList);
} catch (Exception e) {}
}
}
}
//是否提前建立连接
if (isEnablePrimingConnections()) {
for (Server server : allServers) {
if (!allServerList.contains(server)) {
server.setReadyToServe(false);
newServers.add((Server) server);
}
}
if (primeConnections != null) {
primeConnections.primeConnectionsAsync(newServers, this);
}
}
//复制最新的服务列表
allServerList = allServers;
if (canSkipPing()) {
for (Server s : allServerList) {
s.setAlive(true);
}
upServerList = allServerList;
} else if (listChanged) {
forceQuickPing();
}
} finally {
writeLock.unlock();
}
}
//PingTask任务
class PingTask extends TimerTask {
public void run() {
try {
new Pinger(pingStrategy).runPinger();
} catch (Exception e) { }
}
}
//
class Pinger {
private final IPingStrategy pingerStrategy;
public Pinger(IPingStrategy pingerStrategy) {this.pingerStrategy = pingerStrategy;}
public void runPinger() throws Exception {
//幂等处理
if (!pingInProgress.compareAndSet(false, true)) { return; }
// we are "in" - we get to Ping
Server[] allServers = null;
boolean[] results = null;
Lock allLock = null;
Lock upLock = null;
try {
//加把读锁，获取当前的服务列表的数组快照allServers，接着就释放锁了
allLock = allServerLock.readLock();
allLock.lock();
allServers = allServerList.toArray(new Server[allServerList.size()]);
allLock.unlock();
//使用指定的ping策略对allServers里的每一个服务进行ping
int numCandidates = allServers.length;
//默认为SerialPingStrategy，会遍历服务，调用IPing实现类ping， Results[i] = ping.isAlive(servers[i]);
//默认的IPing实现类为NoOpPing，始终返回true
results = pingerStrategy.pingServers(ping, allServers);
final List<Server> newUpList = new ArrayList<Server>();
final List<Server> changedServers = new ArrayList<Server>();
//根据ping的结果形成新的newUpList，使用读写多进行upServerList复制
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); }
if (isAlive) {newUpList.add(svr);}
}
upLock = upServerLock.writeLock();
upLock.lock();
upServerList = newUpList;
upLock.unlock();
notifyServerStatusChangeListener(changedServers);
} finally {
pingInProgress.set(false);
}
}
}
private final Counter createCounter() {
return Monitors.newCounter("LoadBalancer_ChooseServer");
}
//基于rule实现chooseServer
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) {return null;}
}
}
public void markServerDown(Server server) {
if (server == null || !server.isAlive()) {return;}
server.setAlive(false);
notifyServerStatusChangeListener(singleton(server));
}
}
3.7.2.2、LoadBalancerStats
负载均衡LB需要依赖这些统计信息做为判断的策略，负载均衡器的统计类主要是LoadBalancerStats，一个serverId下所有服务实例对应一个LoadBalancerStats。
其内部持有:
* LoadingCache<Server, ServerStats> serverStatsCache
①ServerStats对每个Server的运行情况做了相关统计如：平均响应时间、累计失败数、熔断（时间）控制等
②ServerStats的增删改查的方法
③基于ServerStats实现的方法，如noteResponseTime方法实现为ServerStatsnoteResponseTime(msecs)，记录响应时间
* Map<String, ZoneStats> zoneStatsMap和Map<String, List<? extends Server>> upServerListZoneMap
①zoneStatsMap是在在每一个Servrt基础上又进行区域zone的划分，ZoneStats保存着每一个zone的状态（主要用于记录zone名称）
②upServerListZoneMap直接就是以zone的划分Server
注：子类DynamicServerListLoadBalancer重写setServersList会更新这个两个map，即在实例列表初始化或者定时拉取的时候会更新
* 基于ServerStats获取一些报表：
①获取一个区域快照getZoneSnapshot(List<? extends Server> servers)
- 对一个分区中的所有服务实例进行计算后封装为new ZoneSnapshot(instanceCount, circuitBreakerTrippedCount, activeConnectionsCount, loadPerServer)
- instanceCount：当前服务数
- circuitBreakerTrippedCount：当前滑动窗口内断路的服务实例数
- activeConnectionsCount：当前滑动窗口内此分区所有服务实例的有效连接总数数
- LoadPerServer = activeConnectionsCount / (instanceCount - circuitBreakerTrippedCount)
new ZoneSnapshot(instanceCount, circuitBreakerTrippedCount, activeConnectionsCount, loadPerServer)
②略
public class LoadBalancerStats implements IClientConfigAware {
private static final String PREFIX = "LBStats_";
String name;
//区域状态表，ZoneStats里面封装了zone名称和一个计数器
volatile Map<String, ZoneStats> zoneStatsMap = new ConcurrentHashMap<String, ZoneStats>();
//区域名称和服务列表的映射
volatile Map<String, List<? extends Server>> upServerListZoneMap = new ConcurrentHashMap<String, List<? extends Server>>();
//serverStats断路器相关
private volatile CachedDynamicIntProperty connectionFailureThreshold;
private volatile CachedDynamicIntProperty circuitTrippedTimeoutFactor;
private volatile CachedDynamicIntProperty maxCircuitTrippedTimeout;
//serverStats表
private static final DynamicIntProperty SERVERSTATS_EXPIRE_MINUTES =
DynamicPropertyFactory.getInstance().getIntProperty("niws.loadbalancer.serverStats.expire.minutes", 30);
private final LoadingCache<Server, ServerStats> serverStatsCache =
CacheBuilder.newBuilder()
.expireAfterAccess(SERVERSTATS_EXPIRE_MINUTES.get(), TimeUnit.MINUTES)
.removalListener(new RemovalListener<Server, ServerStats>() {
@Override
public void onRemoval(RemovalNotification<Server, ServerStats> notification) {
notification.getValue().close();
}
})
.build(
new CacheLoader<Server, ServerStats>() {
public ServerStats load(Server server) {
return createServerStats(server);
}
});
protected ServerStats createServerStats(Server server) {
ServerStats ss = new ServerStats(this);
ss.setBufferSize(1000);
ss.setPublishInterval(1000);
ss.initialize(server);
return ss;
}
public LoadBalancerStats(){
zoneStatsMap = new ConcurrentHashMap<String, ZoneStats>();
upServerListZoneMap = new ConcurrentHashMap<String, List<? extends Server>>();
}
public LoadBalancerStats(String name){
this();
this.name = name;
Monitors.registerObject(name, this);
}
//getter/setter略
@Override
public void initWithNiwsConfig(IClientConfig clientConfig)
{
this.name = clientConfig.getClientName();
Monitors.registerObject(name, this);
}
//更新serverStats表
public void updateServerList(List<Server> servers){
for (Server s: servers){
addServer(s);
}
}
public void addServer(Server server) {
try {
serverStatsCache.get(server);
} catch (ExecutionException e) {
ServerStats stats = createServerStats(server);
serverStatsCache.asMap().putIfAbsent(server, stats);
}
}
protected ServerStats getServerStats(Server server) {
try {
return serverStatsCache.get(server);
} catch (ExecutionException e) {
ServerStats stats = createServerStats(server);
serverStatsCache.asMap().putIfAbsent(server, stats);
return serverStatsCache.asMap().get(server);
}
}
//基于ServerStats内暴露的方法
public void noteResponseTime(Server server, double msecs){
ServerStats ss = getServerStats(server);
ss.noteResponseTime(msecs);
}
public void incrementActiveRequestsCount(Server server) {
ServerStats ss = getServerStats(server);
ss.incrementActiveRequestsCount();
}
public boolean isCircuitBreakerTripped(Server server) {
ServerStats ss = getServerStats(server);
return ss.isCircuitBreakerTripped();
}
//略
//zoneStatsMap表相关方法
private ZoneStats getZoneStats(String zone) {
zone = zone.toLowerCase();
ZoneStats zs = zoneStatsMap.get(zone);
if (zs == null){
zoneStatsMap.put(zone, new ZoneStats(this.getName(), zone, this));
zs = zoneStatsMap.get(zone);
}
return zs;
}
public void updateZoneServerMapping(Map<String, List<Server>> map) {
upServerListZoneMap = new ConcurrentHashMap<String, List<? extends Server>>(map);
for (String zone: map.keySet()) {
getZoneStats(zone);
}
}
public int getActiveRequestsCount(String zone) {
return getZoneSnapshot(zone).getActiveRequestsCount();
}
public double getActiveRequestsPerServer(String zone) {
return getZoneSnapshot(zone).getLoadPerServer();
}
//获取某个地区的zoneStatsMap快照
public ZoneSnapshot getZoneSnapshot(String zone) {
if (zone == null) {return new ZoneSnapshot();}
zone = zone.toLowerCase();
List<? extends Server> currentList = upServerListZoneMap.get(zone);
return getZoneSnapshot(currentList);
}
//根据每个服务的断路、可用状态，计算了loadPerServer = 当前滑动窗口内有效连接数（包括了当前时刻状态为断路的） / (当前服务数 - 当前滑动窗口内有效连接数且服务没有断路)
public ZoneSnapshot getZoneSnapshot(List<? extends Server> servers) {
if (servers == null || servers.size() == 0) {
return new ZoneSnapshot();
}
int instanceCount = servers.size();
int activeConnectionsCount = 0;
int activeConnectionsCountOnAvailableServer = 0;
int circuitBreakerTrippedCount = 0;
double loadPerServer = 0;
long currentTime = System.currentTimeMillis();
for (Server server: servers) {
ServerStats stat = getSingleServerStat(server);
if (stat.isCircuitBreakerTripped(currentTime)) {
circuitBreakerTrippedCount++;
} else {
activeConnectionsCountOnAvailableServer += stat.getActiveRequestsCount(currentTime);
}
activeConnectionsCount += stat.getActiveRequestsCount(currentTime);
}
if (circuitBreakerTrippedCount == instanceCount) {
if (instanceCount > 0) {
// should be NaN, but may not be displayable on Epic
loadPerServer = -1;
}
} else {
loadPerServer = ((double) activeConnectionsCountOnAvailableServer) / (instanceCount - circuitBreakerTrippedCount);
}
return new ZoneSnapshot(instanceCount, circuitBreakerTrippedCount, activeConnectionsCount, loadPerServer);
}
//获取指定的zone内的累计请求数
public long getMeasuredZoneHits(String zone) {
if (zone == null) {
return 0;
}
zone = zone.toLowerCase();
long count = 0;
List<? extends Server> currentList = upServerListZoneMap.get(zone);
if (currentList == null) {
return 0;
}
for (Server server: currentList) {
ServerStats stat = getSingleServerStat(server);
count += stat.getMeasuredRequestsCount();
}
return count;
}
//计算指定zone的 (活跃请求数量（周期） + 断路数 ) / 服务数
public int getCongestionRatePercentage(String zone) {
if (zone == null) {
return 0;
}
zone = zone.toLowerCase();
List<? extends Server> currentList = upServerListZoneMap.get(zone);
if (currentList == null || currentList.size() == 0) {
return 0;
}
int serverCount = currentList.size();
int activeConnectionsCount = 0;
int circuitBreakerTrippedCount = 0;
for (Server server: currentList) {
ServerStats stat = getSingleServerStat(server);
activeConnectionsCount += stat.getActiveRequestsCount();
if (stat.isCircuitBreakerTripped()) {
circuitBreakerTrippedCount++;
}
}
return (int) ((activeConnectionsCount + circuitBreakerTrippedCount) * 100L / serverCount);
}
}
3.7.2.3、ServerStats
- 用于记录一个Server实例的统计数据/属性
CircuitBreaker断路器相关属性
*connectionFailureThreshold：连接失败阈值，默认值3（超过就熔断）
默认值配置：niws.loadbalancer.default.connectionFailureCountThreshold此key指定；个性化配置："niws.loadbalancer." + name + ".connectionFailureCountThreshold"
* circuitTrippedTimeoutFactor：断路器超时因子，默认值10s。
默认值配置： niws.loadbalancer.default.circuitTripTimeoutFactorSeconds；个性化配置："niws.loadbalancer." + name + ".circuitTripTimeoutFactorSeconds"
* maxCircuitTrippedTimeout：断路器最大超时秒数（默认使用超时因子计算出来），默认值是30s。
默认值配置：niws.loadbalancer.default.circuitTripMaxTimeoutSeconds；个性化配置："niws.loadbalancer." + name + ".circuitTripMaxTimeoutSeconds"
* successiveConnectionFailureCount：连续（successive）请求异常数量（这个连续发生在Retry重试期间）。
在重试期间，但凡有一次成功了，就会把此参数置为0（失败的话此参数就一直加）
说明：只有通常callErrorHandler.isCircuitTrippingException(e)方法判定才会算作失败，才会+1，
默认情况下只有SocketException/SocketTimeoutException这两种异常才算失败哦~
* totalCircuitBreakerBlackOutPeriod：断路器断电总时长（连续失败>=3次，增加20~30秒。具体增加多少秒，后面有计算逻辑）。
* totalRequests：总请求数量。每次请求结束/错误时就会+1。
* activeRequestsCount：活跃请求数量（正在请求的数量，它能反应该Server的负载、压力）。
但凡只要开始执行Sever了，就+1；但凡只要请求完成了/出错了，就-1；注意：它有时间窗口的概念，后面讲具体逻辑
* lastActiveRequestsCountChangeTimestamp：简单的说就是activeRequestsCount的值最后变化的时间戳
* activeRequestsCountTimeout：活跃请求数量的统计周期，如果lastActiveRequestsCountChangeTimestamp距离now超过了此周期，activeRequestsCount会重新计算
* openConnectionsCount：暂无任何使用处，可忽略。
* lastConnectionFailedTimestamp：最后一次失败的时间戳。至于什么叫失败，参考successiveConnectionFailureCount对失败的判断逻辑
* lastAccessedTimestamp：最后访问时间戳。和lastActiveRequestsCountChangeTimestamp的区别是，它增/减都update一下，
而lastAccessedTimestamp只有在增的时候才会update一下。
* firstConnectionTimestamp：首次连接时间戳，只会记录首次请求进来时的时间。
* failureCountSlidingWindowInterval：失败次数统计时间窗。默认值1000ms
* serverFailureCounts：上一秒失败次数（上一秒是因为failureCountSlidingWindowInterval默认1000ms）,
successiveConnectionFailureCount增它就增，只不过它有时间窗口(1s)
* requestCountInWindow：一个窗口期内的请求总数，窗口期默认为5分钟（300秒）,activeRequestsCount增它就增，只不过它有时间窗口(300s)
响应时间的累计预计
* dataDist：它是一个DataAccumulator，数据累加器。内部维护这一个样本区，是一个数值类型的数组，每次调用noteValue都会在数组依次填充记录，数组下标++，
如果满了会从头填充，之前的数据不会清空
* publisher：定时publish发布数据，默认1分钟发布一次，会计算数组中数据10,20…,90…,99.5百分比数据的统计值。统计之后会清空数据
比如数组buf大小为10，数为 [ 1,2,3,4,5, null,null,null,null,null]，开始统计，表示在最近的1分钟内，只记录了5个数据 , 90%的统计值为：5 * 90% = 4.5 ,取low = 4 , high = 5
返回 buf[iLow] + (index - iLow) * (buf[iHigh] - buf[iLow]) = 2.5
* responseTimeDist：它是个Distribution类型，因为它仅仅只需要持续累加数据，然后提供最大最小值、平均值的访问而已
dataDist和responseTimeDist统一通过noteResponseTime(double msecs)来记录每个请求的响应时间，dataDist按照时间窗口统计，responseTimeDist一直累加。
（2）CircuitBreaker断路器的原理
* 本处的断路器解释
①当有某个服务存在多个实例时，在请求的过程中，负载均衡器会统计每次请求的情况（请求响应时间，是否发生网络异常等），当出现了请求出现累计重试时，负载均衡器会标识当前服务实例，
设置当前服务实例的断 路的时间区间，在此区间内，当请求过来时，负载均衡器会将此服务实例从可用服务实例列表中暂时剔除（其实就是暂时忽略此Server），优先选择其他服务实例。
②该断路器和Hystrix无任何关系，无任何关系，无任何关系。它是ServerStats内部维护的一套熔断机制，体现在如下getCircuitBreakerTimeout方法上
该断路器规则非常简单，开启与否完全由连续失败来决定，而是否算失败由RetryHandler#isCircuitTrippingException来决定，默认它只认为SocketException/SocketTimeoutException（或者其子类异常）属于该种类型的异常
任何业务异常（如NPE）和此断路器没有半毛钱关系。本断路器断的是Server，也就是远程服务器， Hystrix断路器断的是Client，也就是客户端的调用
* 熔断结束时间的计算
目前断路器统计失败是靠连续失败次数去判断断路逻辑的。此方法逻辑可总结如下：
①若连续失败次数还小于阈值（默认3次），那么就不用断路。返回0即可，否则执行计算要断开多久的逻辑
连续（successive）请求异常数量（这个连续发生在Retry重试期间），在重试期间，但凡有一次成功了，就会把此参数置为0（失败的话此参数就一直加）
说明：只有通常callErrorHandler.isCircuitTrippingException(e)方法判定才会算作失败，才会+1，默认情况下只有SocketException/SocketTimeoutException这两种异常
②计算失败基数，最大不能超过16（就算你连续失败100次，此基数也是16）
③根据超时因子circuitTrippedTimeoutFactor（默认是10）计算出时间值blackOutSeconds，该值不能大于上限connectionFailureCircuitTimeout（默认30s），
也就是说保证了断路器最长不能打开超过30s
④断路结束计算公式：（连续失败次数 - 失败基数） * 2 * 超时因子
// 看看该断路器到哪个时间点（关闭）的时刻时间戳，在这个时间段内，该服务是断路的，在Server被熔断期间，负载均衡器都将忽略此Server。
// 比如断路器要从0点开30s，那么返回值就是00:00:30s这个时间戳呗，= 最后一次连接失败时间 + 断路周期
private long getCircuitBreakerTimeout() {
long blackOutPeriod = getCircuitBreakerBlackoutPeriod();
if (blackOutPeriod <= 0) {
return 0;
}
return lastConnectionFailedTimestamp + blackOutPeriod;
}
// 返回需要中断的持续时间（毫秒值），如果大于0表示需要断路
private long getCircuitBreakerBlackoutPeriod() {
int failureCount = successiveConnectionFailureCount.get();
int threshold = connectionFailureThreshold.get();
if (failureCount < threshold) {
return 0;
}
int diff = (failureCount - threshold) > 16 ? 16 : (failureCount - threshold);
int blackOutSeconds = (1 << diff) * circuitTrippedTimeoutFactor.get();
if (blackOutSeconds > maxCircuitTrippedTimeout.get()) {
blackOutSeconds = maxCircuitTrippedTimeout.get();
}
return blackOutSeconds * 1000L;
}
* 断路器如何闭合？
倘若断路器打开了，它如何恢复呢？有如下3种情形它会恢复到正常状态：
①不是连续失败了，也就是成功了一次，那么successiveConnectionFailureCount就会立马归0，所以熔断器就闭合了
②即使请求失败了，但是并非是断路器类异常，即不是RetryHandler#isCircuitTrippingException这种类型的异常时（比如RuntimeException就不是这种类型的异常），那就也不算连续失败，所以也就闭合了
③到时间了，断路器自然就自动闭合了
（3）有窗口性质的统计
* serverFailureCounts：上一秒失败次数（上一秒是因为failureCountSlidingWindowInterval默认1000ms）,
successiveConnectionFailureCount增它就增，只不过它有时间窗口(1s)
* requestCountInWindow：一个窗口期内的请求总数，窗口期默认为5分钟（300秒）,activeRequestsCount增它就增，只不过它有时间窗口(300s)
* 他们都是MeasuredRate类型的
内部使用一个_threshold表示当前窗口的结尾边界，= System.currentTimeMillis() + sampleInterval;
在操作统计值时，如果此时还处于窗口内就累加，否就重置
public class MeasuredRate {
private final AtomicLong _lastBucket = new AtomicLong(0);
private final AtomicLong _currentBucket = new AtomicLong(0);
private final long _sampleInterval;
private volatile long _threshold;
public MeasuredRate(long sampleInterval){
_sampleInterval = sampleInterval;
_threshold = System.currentTimeMillis() + sampleInterval;
}
public long getCount() {
checkAndResetWindow();
return _lastBucket.get();
}
public long getCurrentCount() {
checkAndResetWindow();
return _currentBucket.get();
}
public void increment() {
checkAndResetWindow();
_currentBucket.incrementAndGet();
}
private void checkAndResetWindow() {
long now = System.currentTimeMillis();
if(_threshold < now) {
_lastBucket.set(_currentBucket.get());
_currentBucket.set(0);
_threshold = now + _sampleInterval;
}
}
}
3.7.3、DynamicServerListLoadBalancer
主要是对服务列表的处理，基于几个组件，重写父类的方法
* 在构造器后新增restOfInit处理，以初始化当前ServerId的服务实例
①调用serverListUpdater.start，为服务列表的更新器，默认为PollingServerListUpdater会定时（默认30s）调用updateListOfServers方法，
而Eureka对应的为EurekaNotificationServerListUpdater，为在eurekaClient.registerEventListener(updateListener)，
只有监听到CacheRefreshedEvent事件才会调用updateListOfServers方法
②调用serverListImpl.getUpdatedListOfServers()，再使用filter过滤下，
③再调用setServersList方法，更新好父类的服务列表allServerList，再处理LoadBalancerStats的serverStatsCache和upServerListZoneMap
- 每一个Server都封装为ServerStats在存于父类BaseLoadBlance中的LoadingCache<Server, ServerStats> serverStatsCache
- 把所有Server按照zone进行划分存于父类BaseLoadBlance中的 Map<String, List<? extends Server> upServerListZoneMap
- 把每一个zone封装为ZoneStats在存于父类BaseLoadBlance中的Map<String, ZoneStats> zoneStatsMap
ZoneStats只是对zone的简单封装，存有LoadBalancerStats loadBalancerStats和 String zone;
public class DynamicServerListLoadBalancer<T extends Server> extends BaseLoadBalancer {
private static final Logger LOGGER = LoggerFactory.getLogger(DynamicServerListLoadBalancer.class);
boolean isSecure = false;
boolean useTunnel = false;
protected AtomicBoolean serverListUpdateInProgress = new AtomicBoolean(false);
//基于serverListImpl获取服务列表
volatile ServerList<T> serverListImpl;
volatile ServerListFilter<T> filter;
//处理ServerList的个更新器，默认为PollingServerListUpdater会都是调用updateListOfServers方法，
//而Eureka对应的为EurekaNotificationServerListUpdater，为在eurekaClient.registerEventListener(updateListener)，
//只有监听到CacheRefreshedEvent事件才会调用updateListOfServers方法
protected final ServerListUpdater.UpdateAction updateAction = new ServerListUpdater.UpdateAction() {
@Override
public void doUpdate() {
updateListOfServers();
}
};
protected volatile ServerListUpdater serverListUpdater;
public DynamicServerListLoadBalancer() {
super();
}
//构造器，初始化组件之后，调用restOfInit
public DynamicServerListLoadBalancer(IClientConfig clientConfig, IRule rule, IPing ping,
ServerList<T> serverList, ServerListFilter<T> filter,
ServerListUpdater serverListUpdater) {
super(clientConfig, rule, ping);
this.serverListImpl = serverList;
this.filter = filter;
this.serverListUpdater = serverListUpdater;
if (filter instanceof AbstractServerListFilter) {
((AbstractServerListFilter) filter).setLoadBalancerStats(getLoadBalancerStats());
}
restOfInit(clientConfig);
}
public DynamicServerListLoadBalancer(IClientConfig clientConfig) {
initWithNiwsConfig(clientConfig);
}
//基于配置类初始化默认的组件实现
@Override
public void initWithNiwsConfig(IClientConfig clientConfig) {
try {
super.initWithNiwsConfig(clientConfig);
ServerList<T> niwsServerListImpl = (ServerList<T>) ClientFactory.instantiateInstanceWithClientConfig(niwsServerListClassName, clientConfig);
this.serverListImpl = niwsServerListImpl;
//略
restOfInit(clientConfig);
} catch (Exception e) {throw new RuntimeException();}
}
void restOfInit(IClientConfig clientConfig) {
boolean primeConnection = this.isEnablePrimingConnections();
this.setEnablePrimingConnections(false);
//调用serverListUpdater.start
enableAndInitLearnNewServersFeature();
//调用serverListImpl.getUpdatedListOfServers()，再使用filter过滤下，再调用setServersList
updateListOfServers();
if (primeConnection && this.getPrimeConnections() != null) {
this.getPrimeConnections().primeConnections(getReachableServers());
}
this.setEnablePrimingConnections(primeConnection);
}
@Override
public void setServersList(List lsrv) {
//更新好父类的服务列表allServerList，再处理LoadBalancerStats的serverStatsCache和upServerListZoneMap
super.setServersList(lsrv);
List<T> serverList = (List<T>) lsrv;
Map<String, List<Server>> serversInZones = new HashMap<String, List<Server>>();
for (Server server : serverList) {
// make sure ServerStats is created to avoid creating them on hot
// path
getLoadBalancerStats().getSingleServerStat(server);
String zone = server.getZone();
if (zone != null) {
zone = zone.toLowerCase();
List<Server> servers = serversInZones.get(zone);
if (servers == null) {
servers = new ArrayList<Server>();
serversInZones.put(zone, servers);
}
servers.add(server);
}
}
//getLoadBalancerStats().updateZoneServerMapping(zoneServersMap)
setServerListForZones(serversInZones);
}
protected void setServerListForZones(Map<String, List<Server>> zoneServersMap) {
getLoadBalancerStats().updateZoneServerMapping(zoneServersMap);
}
public void stopServerListRefreshing() {
if (serverListUpdater != null) {
serverListUpdater.stop();
}
}
@Override
public void shutdown() {
super.shutdown();
stopServerListRefreshing();
}
//略
}
3.7.4、ZoneAwareLoadBalancer重写chooseServer
* 为每一个分区创建一个BaseLoadBlance，即ConcurrentHashMap<String, BaseLoadBalancer> balancers
* 重写chooseServer，如果只有一个分区，使用super.chooseServer(key)；如果有多个分区，先通过LoadBalancerStats获取所有分区的状态快照，
使用ZoneAvoidanceRule.getAvailableZones计算出可用有效的分区集合availableZones（具体逻辑见上ZoneAvoidanceRule实现类）
* 再使用ZoneAvoidanceRule.randomChooseZone随机选择一个分区，使用其ZoneAvoidanceRule的chooseServer选择最终的服务实例
即最终使用ZoneAwareLoadBalancer配置的rule的choose方法进行选定，默认就是ZoneAvoidanceRule的choose方法
注：默认ZoneAvoidanceRule也会进行分区处理，当时一个BaseLoadBlance里没有upServerListZoneMap分区概念，其内的upServerList都是同一个分区的
所以会第二个AvailabilityPredicate校验，会筛选不处于断路状态且活跃请求数小于niws.loadbalancer.availabilityFilteringRule.activeConnectionsLimit配置，默认为int最大值
对最终结果List<Server> eligible进行轮询即可
* 为什么给每一个每一个分区创建一个BaseLoadBlance
- 用户这可以自定义的ZoneAwareLoadBalancer的rule，这样ZoneAwareLoadBalancer会帮我们先过滤好分区，然后选择自己设置的rule进行处理
这样就复用了分区的功能
例如可以使用WeightedResponseTimeRule以平均响应时间作为权重进行随机选择
-
public class ZoneAwareLoadBalancer<T extends Server> extends DynamicServerListLoadBalancer<T> {
private ConcurrentHashMap<String, com.netflix.loadbalancer.BaseLoadBalancer> balancers = new ConcurrentHashMap<String, com.netflix.loadbalancer.BaseLoadBalancer>();
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);
void setUpServerList(List<Server> upServerList) {
this.upServerList = upServerList;
}
public ZoneAwareLoadBalancer() {
super();
}
@Deprecated
public ZoneAwareLoadBalancer(IClientConfig clientConfig, IRule rule,
IPing ping, ServerList<T> serverList, ServerListFilter<T> filter) {
super(clientConfig, rule, ping, serverList, filter);
}
@Override
public Server chooseServer(Object key) {
if (!ENABLED.get() || getLoadBalancerStats().getAvailableZones().size() <= 1) {
return super.chooseServer(key);
}
Server server = null;
try {
LoadBalancerStats lbStats = getLoadBalancerStats();
//调用LoadBalancerStats每一个分区的状态快照ZoneSnapshot，ZoneSnapshot包括了此分区的可用连接，断路数的相关数据
//loadPerServer = ((double) activeConnectionsCountOnAvailableServer) / (instanceCount - circuitBreakerTrippedCount);
// = 当前滑动窗口内有效连接数（包括了当前时刻状态为断路的） / (当前服务数 - 当前滑动窗口内有效连接数且服务没有断路)
Map<String, ZoneSnapshot> zoneSnapshot = ZoneAvoidanceRule.createSnapshot(lbStats);
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);
}
//选择有效的分区，再随机选择一个分区zone，使用zoneLoadBalancer.chooseServer(key)选择一个服务实例
Set<String> availableZones = ZoneAvoidanceRule.getAvailableZones(zoneSnapshot, triggeringLoad.get(), triggeringBlackoutPercentage.get());
if (availableZones != null && availableZones.size() < zoneSnapshot.keySet().size()) {
String zone = ZoneAvoidanceRule.randomChooseZone(zoneSnapshot, availableZones);
if (zone != null) {
//在balancers找到分区对应的BaseLoadBalancer，没有会创建，在基于默认的ZoneAvoidanceRule寻找实例
com.netflix.loadbalancer.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);
}
}
@VisibleForTesting
com.netflix.loadbalancer.BaseLoadBalancer getLoadBalancer(String zone) {
zone = zone.toLowerCase();
com.netflix.loadbalancer.BaseLoadBalancer loadBalancer = balancers.get(zone);
if (loadBalancer == null) {
IRule rule = cloneRule(this.getRule());
loadBalancer = new com.netflix.loadbalancer.BaseLoadBalancer(this.getName() + "_" + zone, rule, this.getLoadBalancerStats());
BaseLoadBalancer prev = balancers.putIfAbsent(zone, loadBalancer);
if (prev != null) {
loadBalancer = prev;
}
}
return loadBalancer;
}
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;
}
@Override
public void setRule(IRule rule) {
super.setRule(rule);
if (balancers != null) {
for (String zone: balancers.keySet()) {
balancers.get(zone).setRule(cloneRule(rule));
}
}
}
}
//ZoneAvoidanceRule
public class ZoneAvoidanceRule extends PredicateBasedRule {
private static final Random random = new Random();
private CompositePredicate compositePredicate;
public ZoneAvoidanceRule() {
super();
ZoneAvoidancePredicate zonePredicate = new ZoneAvoidancePredicate(this);
AvailabilityPredicate availabilityPredicate = new AvailabilityPredicate(this);
compositePredicate = createCompositePredicate(zonePredicate, availabilityPredicate);
}
private CompositePredicate createCompositePredicate(ZoneAvoidancePredicate p1, AvailabilityPredicate p2) {
return CompositePredicate.withPredicates(p1, p2)
.addFallbackPredicate(p2)
.addFallbackPredicate(AbstractServerPredicate.alwaysTrue())
.build();
}
@Override
public void initWithNiwsConfig(IClientConfig clientConfig) {
ZoneAvoidancePredicate zonePredicate = new ZoneAvoidancePredicate(this, clientConfig);
AvailabilityPredicate availabilityPredicate = new AvailabilityPredicate(this, clientConfig);
compositePredicate = createCompositePredicate(zonePredicate, availabilityPredicate);
}
static Map<String, ZoneSnapshot> createSnapshot(LoadBalancerStats lbStats) {
Map<String, ZoneSnapshot> map = new HashMap<String, ZoneSnapshot>();
for (String zone : lbStats.getAvailableZones()) {
ZoneSnapshot snapshot = lbStats.getZoneSnapshot(zone);
map.put(zone, snapshot);
}
return map;
}
static String randomChooseZone(Map<String, ZoneSnapshot> snapshot,
Set<String> chooseFrom) {
if (chooseFrom == null || chooseFrom.size() == 0) {
return null;
}
String selectedZone = chooseFrom.iterator().next();
if (chooseFrom.size() == 1) {
return selectedZone;
}
int totalServerCount = 0;
for (String zone : chooseFrom) {
totalServerCount += snapshot.get(zone).getInstanceCount();
}
int index = random.nextInt(totalServerCount) + 1;
int sum = 0;
for (String zone : chooseFrom) {
sum += snapshot.get(zone).getInstanceCount();
if (index <= sum) {
selectedZone = zone;
break;
}
}
return selectedZone;
}
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();
if (instanceCount == 0) {
availableZones.remove(zone);
limitedZoneAvailability = true;
} else {
double loadPerServer = zoneSnapshot.getLoadPerServer();
if (((double) zoneSnapshot.getCircuitTrippedCount())
/ instanceCount >= triggeringBlackoutPercentage
|| loadPerServer < 0) {
availableZones.remove(zone);
limitedZoneAvailability = true;
} else {
if (Math.abs(loadPerServer - maxLoadPerServer) < 0.000001d) {
// they are the same considering double calculation
// round error
worstZones.add(zone);
} else if (loadPerServer > maxLoadPerServer) {
maxLoadPerServer = loadPerServer;
worstZones.clear();
worstZones.add(zone);
}
}
}
}
if (maxLoadPerServer < triggeringLoad && !limitedZoneAvailability) {
// zone override is not needed here
return availableZones;
}
String zoneToAvoid = randomChooseZone(snapshot, worstZones);
if (zoneToAvoid != null) {
availableZones.remove(zoneToAvoid);
}
return availableZones;
}
public static Set<String> getAvailableZones(LoadBalancerStats lbStats,
double triggeringLoad, double triggeringBlackoutPercentage) {
if (lbStats == null) {
return null;
}
Map<String, ZoneSnapshot> snapshot = createSnapshot(lbStats);
return getAvailableZones(snapshot, triggeringLoad,
triggeringBlackoutPercentage);
}
@Override
public AbstractServerPredicate getPredicate() {
return compositePredicate;
}
}
3.8、RibbonLoadBalancerContext统计根据
* RibbonLoadBalancerContext和LoadBalancerContext一致
* 此类的作用类似一个记录工具类，所有方法都需要指定LoadBalancerStats的ServerStats参数
穿插在请调用的整个流程，提供记录和更新服务状态ServerStats的功能，由cliect客户端自己决定何时调用
比如在打开请求时，会调用noteOpenConnectio(ServerStats serverStats)n方法，此方法会把对应的ServerStats的活跃请求指标activeRequestsCount + 1
* 核心方法：
① noteRequestCompletion：请求正常响应或者异常时都会调用
- 活跃请求-1 ； 总请求数+1 ；记录响应时间
stats.decrementActiveRequestsCount();
stats.incrementNumRequests();
stats.noteResponseTime(responseTime);
- 正常响应，清空连续失败记录：stats.clearSuccessiveConnectionFailureCount();
- 如果时符合断路器异常(由RetryHandler默认SocketException.class, SocketTimeoutException.class)：
连续失败次数+1 ；stats.incrementSuccessiveConnectionFailureCount();
服务总失败次数+1：stats.addToFailureCount();
- 其他不符合的异常清空连续失败记录：stats.clearSuccessiveConnectionFailureCount();
②noteError：抛出error的时候调用（同noteRequestCompletion）
protected void
③noteResponse：正常响应的时候调用（同noteRequestCompletion）
protected void
④noteOpenConnection：client开始执行请求的是调用：活跃请求+1
public class LoadBalancerContext implements IClientConfigAware {
private static final Logger logger = LoggerFactory.getLogger(LoadBalancerContext.class);
protected String clientName = "default";
protected String vipAddresses;
//用于判断一个异常类型可否重试或断路
protected int maxAutoRetriesNextServer = DefaultClientConfigImpl.DEFAULT_MAX_AUTO_RETRIES_NEXT_SERVER;
protected int maxAutoRetries = DefaultClientConfigImpl.DEFAULT_MAX_AUTO_RETRIES;
protected RetryHandler defaultRetryHandler = new DefaultLoadBalancerRetryHandler();
protected boolean okToRetryOnAllOperations = DefaultClientConfigImpl.DEFAULT_OK_TO_RETRY_ON_ALL_OPERATIONS.booleanValue();
private ILoadBalancer lb;
private volatile Timer tracer;
public LoadBalancerContext(ILoadBalancer lb) {
this.lb = lb;
}
//活跃请求-1 ； 总请求数+1 ； 记录响应时间
private void recordStats(ServerStats stats, long responseTime) {
if (stats == null) {
return;
}
stats.decrementActiveRequestsCount();
stats.incrementNumRequests();
stats.noteResponseTime(responseTime);
}
//请求正常响应或者异常时都会调用
protected void noteRequestCompletion(ServerStats stats, Object response, Throwable e, long responseTime) {
if (stats == null) {
return;
}
noteRequestCompletion(stats, response, e, responseTime, null);
}
public void noteRequestCompletion(ServerStats stats, Object response, Throwable e, long responseTime, RetryHandler errorHandler) {
if (stats == null) {
return;
}
try {
// 活跃请求-1 ； 总请求数+1 ； 记录响应时间
recordStats(stats, responseTime);
RetryHandler callErrorHandler = errorHandler == null ? getRetryHandler() : errorHandler;
if (callErrorHandler != null && response != null) {
//正常响应，清空连续失败记录
stats.clearSuccessiveConnectionFailureCount();
} else if (callErrorHandler != null && e != null) {
if (callErrorHandler.isCircuitTrippingException(e)) {
//如果时符合断路器异常(默认SocketException.class, SocketTimeoutException.class)：连续失败次数+1 ； 服务总失败次数+1
stats.incrementSuccessiveConnectionFailureCount();
stats.addToFailureCount();
} else {
//其他不符合的异常清空连续失败记录
stats.clearSuccessiveConnectionFailureCount();
}
}
} catch (Exception ex) {}
}
//抛出error的时候调用（同noteRequestCompletion）
protected void noteError(ServerStats stats, ClientRequest request, Throwable e, long responseTime) {
if (stats == null) {
return;
}
try {
// 活跃请求-1 ； 总请求数+1 ； 记录响应时间
recordStats(stats, responseTime);
RetryHandler errorHandler = getRetryHandler();
if (errorHandler != null && e != null) {
if (errorHandler.isCircuitTrippingException(e)) {
stats.incrementSuccessiveConnectionFailureCount();
stats.addToFailureCount();
} else {
stats.clearSuccessiveConnectionFailureCount();
}
}
} catch (Exception ex) {}
}
//正常响应的时候调用（同noteRequestCompletion）
protected void noteResponse(ServerStats stats, ClientRequest request, Object response, long responseTime) {
if (stats == null) {
return;
}
try {
recordStats(stats, responseTime);
RetryHandler errorHandler = getRetryHandler();
if (errorHandler != null && response != null) {
stats.clearSuccessiveConnectionFailureCount();
}
} catch (Exception ex) {
logger.error("Error noting stats for client {}", clientName, ex);
}
}
//client开始执行请求的是调用：活跃请求+1
public void noteOpenConnection(ServerStats serverStats) {
if (serverStats == null) {
return;
}
try {
serverStats.incrementActiveRequestsCount();
} catch (Exception ex) {
logger.error("Error noting stats for client {}", clientName, ex);
}
}
//....
}
4、通过SpringClientFactory获取bean（以serviceId容器隔离）
* 在RibbonAutoConfiguration 自动配置类中会创建SpringClientFactory 、configurations为配置表
@Bean
public SpringClientFactory springClientFactory() {
SpringClientFactory factory = new SpringClientFactory();
factory.setConfigurations(this.configurations);
return factory;
}
* SpringClientFactory继承NamedContextFactory，在创建时，构造方法指定了默认的配置信息和名称
public SpringClientFactory() {
super(RibbonClientConfiguration.class, NAMESPACE, "ribbon.client.name");
}
* NamedContextFactory提供以name为key的上下文表（父容器为顶层的Spring容器）
- 可以为每一个服务创建一个容器，内部为各个服务指定配置的实例和默认的配置实例
* 核心作用：获取bean
- SpringClientFactory.getInstance(String name, Class<C> type)
- 通过NamedContextFactory获取指定容器内的某个type实例
获取不到时，会通过IClientConfig进行创建，调用容器的autowireBean进属性注入，但是最终实例不会放入容器中
//SpringClientFactory
public class SpringClientFactory extends NamedContextFactory<RibbonClientSpecification> {
static final String NAMESPACE = "ribbon";
public SpringClientFactory() {
super(RibbonClientConfiguration.class, NAMESPACE, "ribbon.client.name");
}
public ILoadBalancer getLoadBalancer(String name) {
return getInstance(name, ILoadBalancer.class);
}
//通过NamedContextFactory获取指定容器内的某个type实例，获取不到时，会通过IClientConfig进行创建，
//调用容器的autowireBean进属性注入，但是最终实例不会放入容器中
@Override
public <C> C getInstance(String name, Class<C> type) {
C instance = super.getInstance(name, type);
if (instance != null) {
return instance;
}
IClientConfig config = getInstance(name, IClientConfig.class);
return instantiateWithConfig(getContext(name), type, config);
}
//略
}
//NamedContextFactory
public abstract class NamedContextFactory<C extends org.springframework.cloud.context.named.NamedContextFactory.Specification>
implements DisposableBean, ApplicationContextAware {
//默认的配置类信息
private final String propertySourceName;
private final String propertyName;
private Class<?> defaultConfigType;
//容器表
private Map<String, AnnotationConfigApplicationContext> contexts = new ConcurrentHashMap<>();
//@RibbonClients、RibbonClient注解，name和指定的配置类
private Map<String, C> configurations = new ConcurrentHashMap<>();
private ApplicationContext parent;
public NamedContextFactory(Class<?> defaultConfigType, String propertySourceName,
String propertyName) {
this.defaultConfigType = defaultConfigType;
this.propertySourceName = propertySourceName;
this.propertyName = propertyName;
}
//获取指定name上下文中，指定类型的实例
public <T> T getInstance(String name, Class<T> type) {
AnnotationConfigApplicationContext context = getContext(name);
if (BeanFactoryUtils.beanNamesForTypeIncludingAncestors(context,type).length > 0) {
return context.getBean(type);
}
return null;
}
protected AnnotationConfigApplicationContext getContext(String name) {
if (!this.contexts.containsKey(name)) {
synchronized (this.contexts) {
if (!this.contexts.containsKey(name)) {
this.contexts.put(name, createContext(name));
}
}
}
return this.contexts.get(name);
}
//创建指定name的上下文，并注入指定和默认的配置实例
protected AnnotationConfigApplicationContext createContext(String name) {
//创建容器
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
//在配置表中找到对应name作用配置类，进行解析注入
if (this.configurations.containsKey(name)) {
for (Class<?> configuration : this.configurations.get(name)
.getConfiguration()) {
context.register(configuration);
}
}
//在配置表中找到默认的全局配置类（defaultConfiguratiuon指定的），进行解析注入
for (Map.Entry<String, C> entry : this.configurations.entrySet()) {
if (entry.getKey().startsWith("default.")) {
for (Class<?> configuration : entry.getValue().getConfiguration()) {
context.register(configuration);
}
}
}
//在配置表中找到默认的全局配置类（构造器defaultConfigType指定的），进行解析注入
context.register(PropertyPlaceholderAutoConfiguration.class,this.defaultConfigType);
context.getEnvironment().getPropertySources().addFirst(new MapPropertySource(this.propertySourceName,
Collections.<String, Object>singletonMap(this.propertyName, name)));
if (this.parent != null) {
context.setParent(this.parent);
context.setClassLoader(this.parent.getClassLoader());
}
context.setDisplayName(generateDisplayName(name));
context.refresh();
return context;
}
//略
}
5、@LoadBalanced + RestTemplate 的主流程
5.1、实例
（1）使用 @LoadBalanced 注解修饰的 RestTemplate：
@Configuration
public class RestConfig {
//创建restTemplate对象。LoadBalanced注解表示赋予restTemplate使用Ribbon的负载均衡的能力（一定要加上注解，否则无法远程调用）
@Bean
@LoadBalanced
public RestTemplate restTemplate(){
return new RestTemplate();
}
}
（2）通过 RestTemplate 请求远程服务地址并接收返回值
@RestController
@RequestMapping(value = "api/invoke")
public class InvokeController{
@Autowired
private RestTemplate restTemplate;
public String getByRestTemplate(Integer num1, Integer num2){
//第一个cloud-producer-server代表在nacos注册中心中的服务名，第二个cloud-producer-server代表contextPath配置的项目路径
String url = "http://cloud-producer-server/cloud-producer-server/getSum";
MultiValueMap<String, Object> params = new LinkedMultiValueMap<>();
params.add("num1", num1);
params.add("num2", num2);
return restTemplate.postForObject(url, params, String.class);
}
}
5.2、RestTemplate到LoadBalancerClient（RibbonLoadBalancerClient）的流程
* 这里的RestTemplate使用的是原生的RestTemplate，即没有进过任何定制的（可支持自定义超时属性、底层HttpClient 工具）
默认使用JDK自带的HttpURLConnection进行通信的
@Bean
@LoadBalanced
public RestTemplate restTemplate(){
return new RestTemplate();
}
* 标志了@LoadBalanced的restTemplate.postForObject(url, params, String.class) ->doExecute的流程
- 通过restTemplate实例中设置的ClientHttpRequestFactory返回，由于存在拦截器则为InterceptingClientHttpRequestFactory
会创建InterceptingClientHttpRequest(requestFactory, this.interceptors, uri, httpMethod);
- 执行InterceptingClientHttpRequest.execute()
就会执行拦截器链的，最后会调用LoadBalancerClient.execute(serviceName,this.requestFactory.createRequest(request, body, execution))
①在LoadBalancerAutoConfiguration自动配置类中可以指定
- LoadBalancerClient指RibbonLoadBalancerClient类型对象
- requestFactory指LoadBalancerRequestFactory类型对象
②参数serviceName为url路径的域名
③参数request是InterceptingClientHttpRequest：封装了请求对象和通信工具
//RestTemplate
//postForObjec->doExecute
@Nullable
protected <T> T doExecute(URI url, @Nullable HttpMethod method, @Nullable RequestCallback requestCallback,
@Nullable ResponseExtractor<T> responseExtractor) throws RestClientException {
ClientHttpResponse response = null;
try {
//InterceptingClientHttpRequest(requestFactory, this.interceptors, uri, httpMethod);
ClientHttpRequest request = createRequest(url, method);
if (requestCallback != null) {
requestCallback.doWithRequest(request);
}
//InterceptingClientHttpRequest.executeInternal->InterceptingRequestExecution类的.execute
//执行拦截器链的
response = request.execute();
handleResponse(url, method, response);
return (responseExtractor != null ? responseExtractor.extractData(response) : null);
}
catch (IOException ex) {throw new ResourceAccessException();
}
finally {
if (response != null) {response.close(); }
}
}
protected ClientHttpRequest createRequest(URI url, HttpMethod method) throws IOException {
ClientHttpRequest request = getRequestFactory().createRequest(url, method);
initialize(request);
return request;
}
//RestTemplate的父类，InterceptingHttpAccessor
//InterceptingClientHttpRequestFactory底层封装了原始SimpleClientHttpRequestFactory和拦截器，装饰了拦截功能
@Override
public ClientHttpRequestFactory getRequestFactory() {
List<ClientHttpRequestInterceptor> interceptors = getInterceptors();
if (!CollectionUtils.isEmpty(interceptors)) {
ClientHttpRequestFactory factory = this.interceptingRequestFactory;
if (factory == null) {
factory = new InterceptingClientHttpRequestFactory(super.getRequestFactory(), interceptors);
this.interceptingRequestFactory = factory;
}
return factory;
}
else {
return super.getRequestFactory();
}
}
//InterceptingRequestExecution
private class InterceptingRequestExecution implements ClientHttpRequestExecution {
private final Iterator<ClientHttpRequestInterceptor> iterator;
public InterceptingRequestExecution() {
this.iterator = interceptors.iterator();
}
//存在拦截器的：走每一个拦截器的intercept方法
//不存在拦截器：走原始SimpleClientHttpRequestFactory创建的SimpleBufferingClientHttpRequest会设置好超时配置
// setConnectTimeout/setReadTimeout，值是在restTemplate中设置的
// 并且是使用JDK自带的HttpURLConnection进行通信的
@Override
public ClientHttpResponse execute(HttpRequest request, byte[] body) throws IOException {
if (this.iterator.hasNext()) {
ClientHttpRequestInterceptor nextInterceptor = this.iterator.next();
return nextInterceptor.intercept(request, body, this);
}
else {
HttpMethod method = request.getMethod();
ClientHttpRequest delegate = requestFactory.createRequest(request.getURI(), method);
request.getHeaders().forEach((key, value) -> delegate.getHeaders().addAll(key, value));
if (body.length > 0) {
if (delegate instanceof StreamingHttpOutputMessage) {
StreamingHttpOutputMessage streamingOutputMessage = (StreamingHttpOutputMessage) delegate;
streamingOutputMessage.setBody(outputStream -> StreamUtils.copy(body, outputStream));
}
else {
StreamUtils.copy(body, delegate.getBody());
}
}
return delegate.execute();
}
}
}
//最终调用的LoadBalancerClient.execute(serviceName,this.requestFactory.createRequest(request, body, execution))
public class LoadBalancerInterceptor implements ClientHttpRequestInterceptor {
private LoadBalancerClient loadBalancer;
private LoadBalancerRequestFactory requestFactory;
public LoadBalancerInterceptor(LoadBalancerClient loadBalancer,
LoadBalancerRequestFactory requestFactory) {
this.loadBalancer = loadBalancer;
this.requestFactory = requestFactory;
}
public LoadBalancerInterceptor(LoadBalancerClient loadBalancer) {
this(loadBalancer, new LoadBalancerRequestFactory(loadBalancer));
}
@Override
public ClientHttpResponse intercept(final HttpRequest request, final byte[] body,
final ClientHttpRequestExecution execution) throws IOException {
final URI originalUri = request.getURI();
String serviceName = originalUri.getHost();
return this.loadBalancer.execute(serviceName,this.requestFactory.createRequest(request, body, execution));
}
}
5.3、RibbonLoadBalancerClient.execute方法（LoadBalancerClient接口方法）
public class LoadBalancerInterceptor implements ClientHttpRequestInterceptor {
//..
@Override
public ClientHttpResponse intercept(final HttpRequest request, final byte[] body,
final ClientHttpRequestExecution execution) throws IOException {
final URI originalUri = request.getURI();
String serviceName = originalUri.getHost();
return this.loadBalancer.execute(serviceName,this.requestFactory.createRequest(request, body, execution));
}
}
* 核心方法RibbonLoadBalancerClient.execute的实现
- 此方法是LoadBalancerClient接口的核心方法，用于选中一个服务实例后，执行
- 基于SpringClientFactory在serviceId对应的上下文中获取ILoadBalancer实例，默认为ZoneAwareLoadBalancer
SpringClientFactory的作用和LoadBalancerClientFactory一模一样，都是Name容器，以serviceId为一个容器的维度
里面包含了ribbon的所有必要的组件bean。每一个serverId是相互隔离的
- 核心调用其ILoadBalancer的chooseServer方法获取对应目标服务实例
- 最后调用LoadBalancerRequest.apply(serviceInstance)执行请求
执行 LoadBalancerRequestFactory.createRequest(request, body, execution)创建一个函数指针，进行url通信
* LoadBalancerRequestFactory.createRequest(request, body, execution)
- 使用LoadBalancerRequestFactory创建一个函数指针，把选择的服务实例，使用InterceptingRequestExecution.execute(serviceRequest, body);完成通信，
- InterceptingRequestExecution的execute流程见上，由于此时没有拦截器了，就会走默认的execute步骤
走原始SimpleClientHttpRequestFactory创建的SimpleBufferingClientHttpRequest会设置好超时配置，setConnectTimeout/setReadTimeout，值是在restTemplate中设置的
最终使用使用JDK自带的HttpURLConnection进行通信的
public class RibbonLoadBalancerClient implements LoadBalancerClient {
private SpringClientFactory clientFactory;
public RibbonLoadBalancerClient(SpringClientFactory clientFactory) {
this.clientFactory = clientFactory;
}
@Override
public URI reconstructURI(ServiceInstance instance, URI original) {
//略
return context.reconstructURIWithServer(server, uri);
}
public <T> T execute(String serviceId, LoadBalancerRequest<T> request, Object hint)
throws IOException {
//基于SpringClientFactory在serviceId对应的上下文中获取ILoadBalancer实例，默认为ZoneAwareLoadBalancer
ILoadBalancer loadBalancer = getLoadBalancer(serviceId);
//使用ILoadBalancer的chooseServer方法获取最终选定的服务
Server server = getServer(loadBalancer, hint);
if (server == null) {
throw new IllegalStateException("No instances available for " + serviceId);
}
RibbonServer ribbonServer = new RibbonServer(serviceId, server,
isSecure(server, serviceId),
serverIntrospector(serviceId).getMetadata(server));
//核心是调用request.apply(serviceInstance)执行请求
//LoadBalancerRequest<T> request是RestTemplate执行拦截器时调用提供放入
return execute(serviceId, ribbonServer, request);
}
protected ILoadBalancer getLoadBalancer(String serviceId) {
return this.clientFactory.getLoadBalancer(serviceId);
}
//核心是调用loadBalancer.chooseServer方法
protected Server getServer(ILoadBalancer loadBalancer, Object hint) {
if (loadBalancer == null) {
return null;
}
return loadBalancer.chooseServer(hint != null ? hint : "default");
}
//略
}
public class LoadBalancerRequestFactory {
private LoadBalancerClient loadBalancer;
private List<LoadBalancerRequestTransformer> transformers;
public LoadBalancerRequestFactory(LoadBalancerClient loadBalancer,
List<LoadBalancerRequestTransformer> transformers) {
this.loadBalancer = loadBalancer;
this.transformers = transformers;
}
public LoadBalancerRequestFactory(LoadBalancerClient loadBalancer) {
this.loadBalancer = loadBalancer;
}
public LoadBalancerRequest<ClientHttpResponse> createRequest(
final HttpRequest request, final byte[] body,
final ClientHttpRequestExecution execution) {
return instance -> {
HttpRequest serviceRequest = new ServiceRequestWrapper(request, instance,
this.loadBalancer);
if (this.transformers != null) {
for (LoadBalancerRequestTransformer transformer : this.transformers) {
serviceRequest = transformer.transformRequest(serviceRequest,
instance);
}
}
return execution.execute(serviceRequest, body);
};
}
}
汇聚全球AI编程工具，助力开发者即刻编程。
更多推荐
18
0- 0
已为社区贡献2条内容
所有评论(0)