简介
Spring Cloud Ribbon 是基于Netflix Ribbon实现的一套客户端负载均衡。Ribbon是Netflix发布的一套开源项目, 主要功能是提供客户端的软件负载均衡算法和服务调用。Ribbon客户端提供一系列完善的配置项如连接超时,重试等。
LoadBalance 分为两类, 一种是集中式, 另一种是进程式。
集中式:在服务的提供方和消费方之间提供独立的LoadBalance设施(硬件如F5, 软件如Nginx), 由该设施负责把请求通过某种策略发送到服务提供方。
进程式:将LoadBalance集成到消费方, 消费方从服务注册中心获取到有哪些提供方可以使用, 自己根据策略从提供方地址选择一个, 将请求发送过去。Ribbon就是进程式LoadBalance。
现在在Ribbon git上, 该项目现在进入到维护状态了, 有些模块还在大量使用在项目中。
项目引入
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-netflix-ribbon</artifactId>
<version>2.2.2</version>
</dependency>如果项目中使用到Eureka作为服务注册中心, 引入了spring-cloud-starter-netflix-eureka-client 项目, 则不需要再次引入, eureka已经依赖引入了。
Ribbon LoadBalance机制
客户端启动
上面已经说过Ribbon是一个进程内的LoadBalance, 项目启动时需要启动一个LoadBalancerClient, 用来进行后续的服务请求负载均衡。自动配置加载类RibbonAutoConfiguration, 这个类依赖LoadBalancerAutoConfiguration
和AsyncLoadBalancerAutoConfiguration 这两个都是 ribbon的自动配置(客户端负载均衡) 只不过后者是异步配置这个在后面会说明下。
这里只粘贴一部分代码说明, 有兴趣可直接去看
@Bean
public HasFeatures ribbonFeature() {
return HasFeatures.namedFeature("Ribbon", Ribbon.class);
}
@Bean
public SpringClientFactory springClientFactory() {
SpringClientFactory factory = new SpringClientFactory();
factory.setConfigurations(this.configurations);
return factory;
}
@Bean
@ConditionalOnMissingBean(LoadBalancerClient.class)
public LoadBalancerClient loadBalancerClient() {
return new RibbonLoadBalancerClient(springClientFactory());
}
@Bean
@ConditionalOnClass(name = "org.springframework.retry.support.RetryTemplate")
@ConditionalOnMissingBean
public LoadBalancedRetryFactory loadBalancedRetryPolicyFactory(
final SpringClientFactory clientFactory) {
return new RibbonLoadBalancedRetryFactory(clientFactory);
}
@Bean
@ConditionalOnMissingBean
public PropertiesFactory propertiesFactory() {
return new PropertiesFactory();
}
@Bean
@ConditionalOnProperty("ribbon.eager-load.enabled")
public RibbonApplicationContextInitializer ribbonApplicationContextInitializer() {
return new RibbonApplicationContextInitializer(springClientFactory(),
ribbonEagerLoadProperties.getClients());
}
关键的地方在于LoadBalancerClient Bean实例的获取, 使用ConditionalOnMissingBean注解保证该Bean实例只需要注入一次, 该Bean依赖SpringClientFactory。 SpringClientFactory是创建客户端、负载均衡和客户端配置实例的工厂 , 根据client的名字创建一个Spring ApplicationContext。该类的构造方法中调用了父类的构造方法, 抽象父类为NamedContextFactory, 该类的主要是 创建一组子上下文，允许一组规范定义bean在每个子上下文中。
static final String NAMESPACE = "ribbon";
public SpringClientFactory() {
super(RibbonClientConfiguration.class, NAMESPACE, "ribbon.client.name");
}将 RibbonClientConfiguration 注入的bean 设置到上下文中, 这一步很重要, 配置了ribbon的一些默认配置, 如负载均衡的规则、client、超时时间等。这里的name 启动时默认是client, 当真正的在进行负载均衡时这个name 就是请求的ServiceName, 根据这个serviceName创建个新的clientConfig, 并放到上下文中, 只进行一次初始化。
我们来一个一个看bean的实例:首先是IClientConfig
@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;
}首先是获取客户端配置, 这里会调用DefaultClientConfigImpl.loadProperties方法, 因为默认是使用RestClient, 所以这里的参数名直接叫restClientName。先加载默认配置值, 然后进行配置覆盖。
@Override
public void loadProperties(String restClientName){
enableDynamicProperties = true;
setClientName(restClientName);
loadDefaultValues();
Configuration props = ConfigurationManager.getConfigInstance().subset(restClientName);
for (Iterator<String> keys = props.getKeys(); keys.hasNext(); ){
String key = keys.next();
String prop = key;
try {
if (prop.startsWith(getNameSpace())){
prop = prop.substring(getNameSpace().length() + 1);
}
setPropertyInternal(prop, getStringValue(props, key));
} catch (Exception ex) {
throw new RuntimeException(String.format("Property %s is invalid", prop));
}
}
} public void loadDefaultValues() {
putDefaultIntegerProperty(CommonClientConfigKey.MaxHttpConnectionsPerHost, getDefaultMaxHttpConnectionsPerHost());
putDefaultIntegerProperty(CommonClientConfigKey.MaxTotalHttpConnections, getDefaultMaxTotalHttpConnections());
putDefaultBooleanProperty(CommonClientConfigKey.EnableConnectionPool, getDefaultEnableConnectionPool());
putDefaultIntegerProperty(CommonClientConfigKey.MaxConnectionsPerHost, getDefaultMaxConnectionsPerHost());
putDefaultIntegerProperty(CommonClientConfigKey.MaxTotalConnections, getDefaultMaxTotalConnections());
putDefaultIntegerProperty(CommonClientConfigKey.ConnectTimeout, getDefaultConnectTimeout());
putDefaultIntegerProperty(CommonClientConfigKey.ConnectionManagerTimeout, getDefaultConnectionManagerTimeout());
putDefaultIntegerProperty(CommonClientConfigKey.ReadTimeout, getDefaultReadTimeout());
putDefaultIntegerProperty(CommonClientConfigKey.MaxAutoRetries, getDefaultMaxAutoRetries());
putDefaultIntegerProperty(CommonClientConfigKey.MaxAutoRetriesNextServer, getDefaultMaxAutoRetriesNextServer());
putDefaultBooleanProperty(CommonClientConfigKey.OkToRetryOnAllOperations, getDefaultOkToRetryOnAllOperations());
putDefaultBooleanProperty(CommonClientConfigKey.FollowRedirects, getDefaultFollowRedirects());
putDefaultBooleanProperty(CommonClientConfigKey.ConnectionPoolCleanerTaskEnabled, getDefaultConnectionPoolCleanerTaskEnabled());
putDefaultIntegerProperty(CommonClientConfigKey.ConnIdleEvictTimeMilliSeconds, getDefaultConnectionidleTimeInMsecs());
putDefaultIntegerProperty(CommonClientConfigKey.ConnectionCleanerRepeatInterval, getDefaultConnectionIdleTimertaskRepeatInMsecs());
putDefaultBooleanProperty(CommonClientConfigKey.EnableGZIPContentEncodingFilter, getDefaultEnableGzipContentEncodingFilter());
String proxyHost = ConfigurationManager.getConfigInstance().getString(getDefaultPropName(CommonClientConfigKey.ProxyHost.key()));
if (proxyHost != null && proxyHost.length() > 0) {
setProperty(CommonClientConfigKey.ProxyHost, proxyHost);
}
Integer proxyPort = ConfigurationManager
.getConfigInstance()
.getInteger(
getDefaultPropName(CommonClientConfigKey.ProxyPort),
(Integer.MIN_VALUE + 1)); // + 1 just to avoid potential clash with user setting
if (proxyPort != (Integer.MIN_VALUE + 1)) {
setProperty(CommonClientConfigKey.ProxyPort, proxyPort);
}
putDefaultIntegerProperty(CommonClientConfigKey.Port, getDefaultPort());
putDefaultBooleanProperty(CommonClientConfigKey.EnablePrimeConnections, getDefaultEnablePrimeConnections());
putDefaultIntegerProperty(CommonClientConfigKey.MaxRetriesPerServerPrimeConnection, getDefaultMaxRetriesPerServerPrimeConnection());
putDefaultIntegerProperty(CommonClientConfigKey.MaxTotalTimeToPrimeConnections, getDefaultMaxTotalTimeToPrimeConnections());
putDefaultStringProperty(CommonClientConfigKey.PrimeConnectionsURI, getDefaultPrimeConnectionsUri());
putDefaultIntegerProperty(CommonClientConfigKey.PoolMinThreads, getDefaultPoolMinThreads());
putDefaultIntegerProperty(CommonClientConfigKey.PoolMaxThreads, getDefaultPoolMaxThreads());
putDefaultLongProperty(CommonClientConfigKey.PoolKeepAliveTime, getDefaultPoolKeepAliveTime());
putDefaultTimeUnitProperty(CommonClientConfigKey.PoolKeepAliveTimeUnits, getDefaultPoolKeepAliveTimeUnits());
putDefaultBooleanProperty(CommonClientConfigKey.EnableZoneAffinity, getDefaultEnableZoneAffinity());
putDefaultBooleanProperty(CommonClientConfigKey.EnableZoneExclusivity, getDefaultEnableZoneExclusivity());
putDefaultStringProperty(CommonClientConfigKey.ClientClassName, getDefaultClientClassname());
putDefaultStringProperty(CommonClientConfigKey.NFLoadBalancerClassName, getDefaultNfloadbalancerClassname());
putDefaultStringProperty(CommonClientConfigKey.NFLoadBalancerRuleClassName, getDefaultNfloadbalancerRuleClassname());
putDefaultStringProperty(CommonClientConfigKey.NFLoadBalancerPingClassName, getDefaultNfloadbalancerPingClassname());
putDefaultBooleanProperty(CommonClientConfigKey.PrioritizeVipAddressBasedServers, getDefaultPrioritizeVipAddressBasedServers());
putDefaultFloatProperty(CommonClientConfigKey.MinPrimeConnectionsRatio, getDefaultMinPrimeConnectionsRatio());
putDefaultStringProperty(CommonClientConfigKey.PrimeConnectionsClassName, getDefaultPrimeConnectionsClass());
putDefaultStringProperty(CommonClientConfigKey.NIWSServerListClassName, getDefaultSeverListClass());
putDefaultStringProperty(CommonClientConfigKey.VipAddressResolverClassName, getDefaultVipaddressResolverClassname());
putDefaultBooleanProperty(CommonClientConfigKey.IsClientAuthRequired, getDefaultIsClientAuthRequired());
// putDefaultStringProperty(CommonClientConfigKey.RequestIdHeaderName, getDefaultRequestIdHeaderName());
putDefaultBooleanProperty(CommonClientConfigKey.UseIPAddrForServer, getDefaultUseIpAddressForServer());
putDefaultStringProperty(CommonClientConfigKey.ListOfServers, "");
}加载默认配置时, 参数太多了, 挑几个重要的说下, 其余可自行查阅代码
使用哪种客户端 CommonClientConfigKey.ClientClassName 默认是com.netflix.niws.client.http.RestClient
使用哪个LoadBalance CommonClientConfigKey.NFLoadBalancerClassName 默认是com.netflix.loadbalancer.ZoneAwareLoadBalancer 环境变量或下方创建bean默认会覆盖
使用哪个负载均衡规则 CommonClientConfigKey.NFLoadBalancerRuleClassName 默认是com.netflix.loadbalancer.AvailabilityFilteringRule 环境变量或下方创建bean默认会覆盖
客户端鉴权是否开启CommonClientConfigKey.IsClientAuthRequired 默认 false
线程池最小连接数CommonClientConfigKey.PoolMinThreads 1 最大连接数 CommonClientConfigKey.PoolMaxThreads 200
client的config bean配置完成后, 是其余的bean进行注入。
下面所有从环境变量取ClassName, 环境变量命名规则为 : ServiceName + . + ribbon + . + 接口类对应的名字 如(CLOUD-PAYMENT-SERVICE.ribbon.NFLoadBalancerRuleClassName)
public PropertiesFactory() {
classToProperty.put(ILoadBalancer.class, "NFLoadBalancerClassName");
classToProperty.put(IPing.class, "NFLoadBalancerPingClassName");
classToProperty.put(IRule.class, "NFLoadBalancerRuleClassName");
classToProperty.put(ServerList.class, "NIWSServerListClassName");
classToProperty.put(ServerListFilter.class, "NIWSServerListFilterClassName");
}
//ribbonRule bean注入, 这里会判断环境变量是否已经配置好IRule的实现类, 有的话通过反射获取, 没有的话直接new 个ZoneAvoidanceRule对象。
@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;
}
//ribbonPing bean注入, 这里会判断环境变量是否已经配置好IPing的实现类, 有的话通过反射获取, 没有的话直接new 个DummyPing对象。
@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();
}
//ribbonServerList bean注入, 这里会判断环境变量是否已经配置好ServerList的实现类, 有的话通过反射获取, 没有的话直接new 个ConfigurationBasedServerList对象。
@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;
}
//ribbonServerListUpdater bean注入
@Bean
@ConditionalOnMissingBean
public ServerListUpdater ribbonServerListUpdater(IClientConfig config) {
return new PollingServerListUpdater(config);
}
//ribbonLoadBalancer bean注入, 这里会判断环境变量是否已经配置好ILoadBalancer的实现类, 有的话通过反射获取, 没有的话直接new 个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);
}
//ribbonServerListFilter bean注入, 这里会判断环境变量是否已经配置好ServerListFilter的实现类, 有的话通过反射获取, 没有的话直接new 个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;
}
@Bean
@ConditionalOnMissingBean
public RibbonLoadBalancerContext ribbonLoadBalancerContext(ILoadBalancer loadBalancer,
IClientConfig config, RetryHandler retryHandler) {
return new RibbonLoadBalancerContext(loadBalancer, config, retryHandler);
}
@Bean
@ConditionalOnMissingBean
public RetryHandler retryHandler(IClientConfig config) {
return new DefaultLoadBalancerRetryHandler(config);
}
@Bean
@ConditionalOnMissingBean
public ServerIntrospector serverIntrospector() {
return new DefaultServerIntrospector();
}这样RibbonLoadBalancerClient 客户端各项配置完成, 客户端启动成功。
负载均衡配置
使用到ribbon的负载均衡时需要我们自己注入个RestTemplate类, 并开启LoadBalanced注解
@Bean
@LoadBalanced
public RestTemplate getRestTemplate() {
return new RestTemplate();
}这里我们说下LoadBalancerAutoConfiguration和AsyncLoadBalancerAutoConfiguration 这两个负载均衡配置类
/*
* Copyright 2012-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.client.loadbalancer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingClass;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.client.ClientHttpRequestInterceptor;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.web.client.RestTemplate;
/**
* Auto-configuration for Ribbon (client-side load balancing).
*
* @author Spencer Gibb
* @author Dave Syer
* @author Will Tran
* @author Gang Li
*/
@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(RestTemplate.class)
@ConditionalOnBean(LoadBalancerClient.class)
@EnableConfigurationProperties(LoadBalancerRetryProperties.class)
public class LoadBalancerAutoConfiguration {
//获取容器内所有打上LoadBalanced
注解的RestTemplate实例
@LoadBalanced
@Autowired(required = false)
private List<RestTemplate> restTemplates = Collections.emptyList();
//允许应用程序在给定所选ServiceInstance的情况下转换负载均衡的HttpRequest
@Autowired(required = false)
private List<LoadBalancerRequestTransformer> transformers = Collections.emptyList();
@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);
}
}
});
}
// 设置requestFactory,依赖上面创建好的loadBalancerClient bean实例
@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);
}
// 将负载均衡拦截器加入到开启了负载均衡的restTemplate中
@Bean
@ConditionalOnMissingBean
public RestTemplateCustomizer restTemplateCustomizer(
final LoadBalancerInterceptor loadBalancerInterceptor) {
return restTemplate -> {
List<ClientHttpRequestInterceptor> list = new ArrayList<>(
restTemplate.getInterceptors());
list.add(loadBalancerInterceptor);
restTemplate.setInterceptors(list);
};
}
}
/**
* Auto configuration for retry mechanism.
*/
@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(RetryTemplate.class)
public static class RetryAutoConfiguration {
// 设置loadBalancedRetryFactory
@Bean
@ConditionalOnMissingBean
public LoadBalancedRetryFactory loadBalancedRetryFactory() {
return new LoadBalancedRetryFactory() {
};
}
}
/**
* Auto configuration for retry intercepting mechanism.
*/
@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(RetryTemplate.class)
public static class RetryInterceptorAutoConfiguration {
// RetryLoadBalancerInterceptor 拦截器
@Bean
@ConditionalOnMissingBean
public RetryLoadBalancerInterceptor ribbonInterceptor(
LoadBalancerClient loadBalancerClient,
LoadBalancerRetryProperties properties,
LoadBalancerRequestFactory requestFactory,
LoadBalancedRetryFactory loadBalancedRetryFactory) {
return new RetryLoadBalancerInterceptor(loadBalancerClient, properties,
requestFactory, loadBalancedRetryFactory);
}
@Bean
@ConditionalOnMissingBean
public RestTemplateCustomizer restTemplateCustomizer(
final RetryLoadBalancerInterceptor loadBalancerInterceptor) {
return restTemplate -> {
List<ClientHttpRequestInterceptor> list = new ArrayList<>(
restTemplate.getInterceptors());
list.add(loadBalancerInterceptor);
restTemplate.setInterceptors(list);
};
}
}
}
上面客户端启动, 负载均衡配置, 拦截器设置等一系列配置完成后, 在实际请求就进行负载均衡了。
RestTemplate负载均衡
这里先简单讲讲RestTemplate配合Ribbon的负载均衡, 至于OpenFeign配合Ribbon的负载均衡则在后续的OpenFeign中解析。
我们直接看RestTemplate的doExecute
@Nullable
protected T doExecute(URI url, @Nullable HttpMethod method, @Nullable RequestCallback requestCallback,
@Nullable ResponseExtractor responseExtractor) throws RestClientException {
Assert.notNull(url, "URI is required");
Assert.notNull(method, "HttpMethod is required");
ClientHttpResponse response = null;
try {
ClientHttpRequest request = createRequest(url, method);
if (requestCallback != null) {
requestCallback.doWithRequest(request);
}
response = request.execute();
handleResponse(url, method, response);
return (responseExtractor != null ? responseExtractor.extractData(response) : null);
}
catch (IOException ex) {
String resource = url.toString();
String query = url.getRawQuery();
resource = (query != null ? resource.substring(0, resource.indexOf('?')) : resource);
throw new ResourceAccessException("I/O error on " + method.name() +
" request for \"" + resource + "\": " + ex.getMessage(), ex);
}
finally {
if (response != null) {
response.close();
}
}
}
- 根据URL 和 请求类型创建ClientHttpRequest 对象 返回 InterceptingClientHttpRequest
- 执行InterceptingClientHttpRequest 的execute 方法, 由于设置了LoadBalancerInterceptor 拦截器, 调用intercept
public ClientHttpResponse intercept(final HttpRequest request, final byte[] body,
final ClientHttpRequestExecution execution) throws IOException {
final URI originalUri = request.getURI();
String serviceName = originalUri.getHost();
Assert.state(serviceName != null,
"Request URI does not contain a valid hostname: " + originalUri);
return this.loadBalancer.execute(serviceName,
this.requestFactory.createRequest(request, body, execution));
}
- RibbonLoadBalancerClient execute
public T execute(String serviceId, LoadBalancerRequest request, Object hint)
throws IOException {
ILoadBalancer loadBalancer = getLoadBalancer(serviceId);
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));
return execute(serviceId, ribbonServer, request);
}
在getServer中选择不同的Server, 由于我使用的注册中心是eureka, 并且是Zone只有一个, 所以会调用父类的chooseServer方法。
public Server chooseServer(Object key) {
if (!ENABLED.get() || getLoadBalancerStats().getAvailableZones().size() <= 1) {
logger.debug("Zone aware logic disabled or there is only one zone");
return super.chooseServer(key);
}
Server server = null;
try {
LoadBalancerStats lbStats = getLoadBalancerStats();
Map zoneSnapshot = ZoneAvoidanceRule.createSnapshot(lbStats);
logger.debug("Zone snapshots: {}", zoneSnapshot);
if (triggeringLoad == null) {
triggeringLoad = DynamicPropertyFactory.getInstance().getDoubleProperty(
"ZoneAwareNIWSDiscoveryLoadBalancer." + this.getName() + ".triggeringLoadPerServerThreshold", 0.2d);
}
if (triggeringBlackoutPercentage == null) {
triggeringBlackoutPercentage = DynamicPropertyFactory.getInstance().getDoubleProperty(
"ZoneAwareNIWSDiscoveryLoadBalancer." + this.getName() + ".avoidZoneWithBlackoutPercetage", 0.99999d);
}
Set availableZones = ZoneAvoidanceRule.getAvailableZones(zoneSnapshot, triggeringLoad.get(), triggeringBlackoutPercentage.get());
logger.debug("Available zones: {}", availableZones);
if (availableZones != null && availableZones.size() < zoneSnapshot.keySet().size()) {
String zone = ZoneAvoidanceRule.randomChooseZone(zoneSnapshot, availableZones);
logger.debug("Zone chosen: {}", zone);
if (zone != null) {
BaseLoadBalancer zoneLoadBalancer = getLoadBalancer(zone);
server = zoneLoadBalancer.chooseServer(key);
}
}
} catch (Exception e) {
logger.error("Error choosing server using zone aware logic for load balancer={}", name, e);
}
if (server != null) {
return server;
} else {
logger.debug("Zone avoidance logic is not invoked.");
return super.chooseServer(key);
}
}
根据不同的负载均衡规则来进行Server的选择, 所有的实现都来自于接口IRule
- RoundRobinRule 轮询
- RandomRule 随机
- RetryRule 先按照RoundRobinRule的策略获取服务，如果获取服务失败则在指定时间内会进行重试
- WeightedResponseTimeRule 对RoundRobinRule的扩展，响应速度越快的实例选择权重越大，越容易被选择
- BestAvailableRule 会先过滤掉由于多次访问故障而处于断路器跳闸状态的服务，然后选择一个并发量最小的服务
- AvailabilityFilteringRule 先过滤掉故障实例，再选择并发较小的实例
- ZoneAvoidanceRule 默认规则，复合判断server所在区域的性能和server的可用性选择服务器
ZoneAvoidanceRule 使用的是父类的规则是一种过滤轮询算法, 至于默认规则为什么是这个, 请看上面的内容 点我跳转。所以一般说负载均衡的默认算法是轮询算法, 但不是说使用的RoundRobinRule 轮询规则。
会根据zone来进行服务过滤, 过滤的代码如下ZoneAvoidancePredicate:
@Override
public boolean apply(@Nullable PredicateKey input) {
if (!ENABLED.get()) {
return true;
}
String serverZone = input.getServer().getZone();
if (serverZone == null) {
// there is no zone information from the server, we do not want to filter
// out this server
return true;
}
LoadBalancerStats lbStats = getLBStats();
if (lbStats == null) {
// no stats available, do not filter
return true;
}
if (lbStats.getAvailableZones().size() <= 1) {
// only one zone is available, do not filter
return true;
}
Map zoneSnapshot = ZoneAvoidanceRule.createSnapshot(lbStats);
if (!zoneSnapshot.keySet().contains(serverZone)) {
// The server zone is unknown to the load balancer, do not filter it out
return true;
}
logger.debug("Zone snapshots: {}", zoneSnapshot);
Set availableZones = ZoneAvoidanceRule.getAvailableZones(zoneSnapshot, triggeringLoad.get(), triggeringBlackoutPercentage.get());
logger.debug("Available zones: {}", availableZones);
if (availableZones != null) {
return availableZones.contains(input.getServer().getZone());
} else {
return false;
}
}
服务过滤完成后, 进行轮询选择, 定义一个原子整形类来记录请求的次数private final AtomicInteger nextIndex = new AtomicInteger() 。请求的次数和可用服务进行取余操作, 然后用cas 自旋锁进行更新, 获得的这个角标, 从服务列表中进行取出Server, 进行请求发送。
private int incrementAndGetModulo(int modulo) {
for (;;) {
int current = nextIndex.get();
int next = (current + 1) % modulo;
if (nextIndex.compareAndSet(current, next) && current < modulo)
return current;
}
}
以上就是RestTemplate 配合 Ribbon使用, 由于本人知识有限, 有错误和疏漏之处, 烦请指正! 关于Server的拉取 和 更新后面有机会再说。
代码中使用了大量的设计模式, 比如工厂模式, 建造者模式, 策略模式等。但本人对于策略模式处于这种状态:
眼 : 我会了( • ̀ω•́ )✧
手、大脑 : 不, 你不会 (;￢＿￢)
还是有许多学的, 任重而道远
Comments | 2 条评论
I was pretty pleased to discover this great site. I need to to thank you for your time for this particularly fantastic read!! I definitely appreciated every part of it and I have you book marked to see new stuff on your blog.
@נערות ליווי בתל אביב thanks