客户端负载均衡
学习spring cloud ribbon的时候不得不提到客户端负载均衡。在客户端负载均衡中，所有客户端节点都维护着自己要访问的服务端清单，而这些服务端的清单来自于服务注册中心。客户端发送一个请求经过客户端负载均衡器后发向不同的服务端。
RestTemplate
Spring Cloud Ribbon实现客户端负载均衡的方法是带有@LoadBalanced注解的RestTemplate对象来实现的，只要带上这个注解，发出的rest请求就会经过客户端负载均衡。
Ribbon
Spring Cloud Ribbon 基于Netflix 的 Ribbon实现，Spring Cloud对其封装了，提供了统一的接口，而最终实现是netfilx的ribbon实现。
源码分析
首先我们看下这个@LoadBalanced注解的源码：
/**
* Annotation to mark a RestTemplate bean to be configured to use a LoadBalancerClient
* @author Spencer Gibb
*/
@Target({ ElementType.FIELD, ElementType.PARAMETER, ElementType.METHOD })
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
@Qualifier
public @interface LoadBalanced {
}
从注释可以看出使用这个注解是为了能使用LoadBalancerClient，即负载均衡客户端，这个接口的源码如下：
public interface LoadBalancerClient extends ServiceInstanceChooser {
<T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException;
<T> T execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest<T> request) throws IOException;
URI reconstructURI(ServiceInstance instance, URI original);
}
execute方法：使用从负载均衡器中挑选出的服务实例来执行请求内容。
reconstructURI方法：为系统构建一个合适的host:port形式的uri。
接下来来看下为实现客户端负载均衡器的自动化配置类LoadBalancerAutoConfiguration
/**
* Auto configuration for Ribbon (client side load balancing).
*
* @author Spencer Gibb
* @author Dave Syer
* @author Will Tran
* @author Gang Li
*/
@Configuration
@ConditionalOnClass(RestTemplate.class)
@ConditionalOnBean(LoadBalancerClient.class)
@EnableConfigurationProperties(LoadBalancerRetryProperties.class)
public class LoadBalancerAutoConfiguration {
@LoadBalanced
@Autowired(required = false)
private List<RestTemplate> restTemplates = Collections.emptyList();
@Bean
public SmartInitializingSingleton loadBalancedRestTemplateInitializer(
final List<RestTemplateCustomizer> customizers) {
return new SmartInitializingSingleton() {
@Override
public void afterSingletonsInstantiated() {
for (RestTemplate restTemplate : LoadBalancerAutoConfiguration.this.restTemplates) {
for (RestTemplateCustomizer customizer : customizers) {
customizer.customize(restTemplate);
}
}
}
};
}
@Autowired(required = false)
private List<LoadBalancerRequestTransformer> transformers = Collections.emptyList();
@Bean
@ConditionalOnMissingBean
public LoadBalancerRequestFactory loadBalancerRequestFactory(
LoadBalancerClient loadBalancerClient) {
return new LoadBalancerRequestFactory(loadBalancerClient, transformers);
}
@Configuration
@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 new RestTemplateCustomizer() {
@Override
public void customize(RestTemplate restTemplate) {
List<ClientHttpRequestInterceptor> list = new ArrayList<>(
restTemplate.getInterceptors());
list.add(loadBalancerInterceptor);
restTemplate.setInterceptors(list);
}
};
}
}
@Configuration
@ConditionalOnClass(RetryTemplate.class)
public static class RetryAutoConfiguration {
@Bean
@ConditionalOnMissingBean
public RetryTemplate retryTemplate() {
RetryTemplate template = new RetryTemplate();
template.setThrowLastExceptionOnExhausted(true);
return template;
}
@Bean
@ConditionalOnMissingBean
public LoadBalancedRetryPolicyFactory loadBalancedRetryPolicyFactory() {
return new LoadBalancedRetryPolicyFactory.NeverRetryFactory();
}
@Bean
@ConditionalOnMissingBean
public LoadBalancedBackOffPolicyFactory loadBalancedBackOffPolicyFactory() {
return new LoadBalancedBackOffPolicyFactory.NoBackOffPolicyFactory();
}
@Bean
@ConditionalOnMissingBean
public LoadBalancedRetryListenerFactory loadBalancedRetryListenerFactory() {
return new LoadBalancedRetryListenerFactory.DefaultRetryListenerFactory();
}
}
@Configuration
@ConditionalOnClass(RetryTemplate.class)
public static class RetryInterceptorAutoConfiguration {
@Bean
@ConditionalOnMissingBean
public RetryLoadBalancerInterceptor ribbonInterceptor(
LoadBalancerClient loadBalancerClient, LoadBalancerRetryProperties properties,
LoadBalancedRetryPolicyFactory lbRetryPolicyFactory,
LoadBalancerRequestFactory requestFactory,
LoadBalancedBackOffPolicyFactory backOffPolicyFactory,
LoadBalancedRetryListenerFactory retryListenerFactory) {
return new RetryLoadBalancerInterceptor(loadBalancerClient, properties,
lbRetryPolicyFactory, requestFactory, backOffPolicyFactory, retryListenerFactory);
}
@Bean
@ConditionalOnMissingBean
public RestTemplateCustomizer restTemplateCustomizer(
final RetryLoadBalancerInterceptor loadBalancerInterceptor) {
return new RestTemplateCustomizer() {
@Override
public void customize(RestTemplate restTemplate) {
List<ClientHttpRequestInterceptor> list = new ArrayList<>(
restTemplate.getInterceptors());
list.add(loadBalancerInterceptor);
restTemplate.setInterceptors(list);
}
};
}
}
}
这个类的注释@ConditionalOnClass(RestTemplate.class)表示RestTemplate类必须存在于当前工程的环境中；@ConditionalOnBean(LoadBalancerClient.class)表示当前工程中必须要有LoadBalancerClient的实现bean。
我们先从这个类内部的一个静态类开始分析：
@Configuration
@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 new RestTemplateCustomizer() {
@Override
public void customize(RestTemplate restTemplate) {
List<ClientHttpRequestInterceptor> list = new ArrayList<>(
restTemplate.getInterceptors());
list.add(loadBalancerInterceptor);
restTemplate.setInterceptors(list);
}
};
}
}
这个类主要做了两件事：
- 创建一个LoadBalancerInterceptor的bean。
- 创建一个RestTemplateCustomizer的bean，并将1中创建的拦截器设置到传入的restTemplate对象中。
回到主类LoadBalancerAutoConfiguration的开头，可以看到LoadBalancerAutoConfiguration这个类维护了一个被@LoadBalanced注解修饰的RestTemplate对象列表，在loadBalancedRestTemplateInitializer方法中，使用我们上面提到的RestTemplateCustomizer实例来给RestTemplate对象添加LoadBalancerInterceptor拦截器。
那这个拦截器的具体是怎么实现负载均衡的呢，我们来看这个拦截器的源码：
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));
}
@Override
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, requestFactory.createRequest(request, body, execution));
}
}
当RestTemplate对象发送一个请求时，首先会执行intercept方法。从上面的负载均衡自动化配置类中可以知道，拦截器类LoadBalancerInterceptor中的LoadBalancerClient对象是从外界传入的，而LoadBalancerInterceptor是一个接口，所以在拦截器的intercept方法中执行的LoadBalancerClient对象的execute方法是某个具体的LoadBalancerClient的实现类完成的，针对ribbon的负载均衡器这个接口对应的实现类为RibbonLoadBalancerClient,这个类在netfilx.ribbon包下。
接下来我们就来看下RibbonLoadBalancerClient客户端的execute方法实现：
@Override
public <T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException {
ILoadBalancer loadBalancer = getLoadBalancer(serviceId);
Server server = getServer(loadBalancer);
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);
}
@Override
public <T> T execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest<T> request) throws IOException {
Server server = null;
if(serviceInstance instanceof RibbonServer) {
server = ((RibbonServer)serviceInstance).getServer();
}
if (server == null) {
throw new IllegalStateException("No instances available for " + serviceId);
}
RibbonLoadBalancerContext context = this.clientFactory
.getLoadBalancerContext(serviceId);
RibbonStatsRecorder statsRecorder = new RibbonStatsRecorder(context, server);
try {
T returnVal = request.apply(serviceInstance);
statsRecorder.recordStats(returnVal);
return returnVal;
}
// catch IOException and rethrow so RestTemplate behaves correctly
catch (IOException ex) {
statsRecorder.recordStats(ex);
throw ex;
}
catch (Exception ex) {
statsRecorder.recordStats(ex);
ReflectionUtils.rethrowRuntimeException(ex);
}
return null;
}
- 从源码可知getLoadBalancer完成了通过传入的服务名来选择具体的负载均衡器，代码中的ILoadBalancer负载均衡器是一个接口，ribbon的所有负载均衡器实现类都是这个接口的实现类，不同的负载均衡器对应着不同的负载均衡策略。ribbon在整合spring cloud的时候ILoadBalancer接口的默认实现类为ZoneAwareLoadBalancer类。
- 另外使用getServer方法从负载均衡器中获取的服务对象类Server类定义了一个传统的服务端节点，在该类中存储了服务端节点的一些元数据信息，包括host,port以及一些部署信息。
- 此外代码中提及的另一个类RibbonServer是对Server类进行的包装，它还保存了服务名serviceid,是否需要使用https等其他信息。
- execute方法最终执行的是
request.apply(serviceInstance);这句代码通过调用LoadBalancerRequest请求的apply方法来实现请求的访问。apply方法传入的ServiceInstance接口对象是对服务实例的抽象定义。从代码上可知RibbonServer类就是ServiceInstance接口的实现。另外LoadBalancerRequest是一个接口，它的具体实现要回溯到LoadBalancerInterceptor拦截器中的传参：requestFactory.createRequest(request, body, execution)。而requestFactory是从LoadBalancerAutoConfiguration配置类传过来的，回溯到这个自动化配置类中发现实际传入的对象是LoadBalancerRequestFactory对象。下面我们看下LoadBalancerRequestFactory类的实现：
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 new LoadBalancerRequest<ClientHttpResponse>() {
@Override
public ClientHttpResponse apply(final ServiceInstance instance)
throws Exception {
HttpRequest serviceRequest = new ServiceRequestWrapper(request, instance, loadBalancer);
if (transformers != null) {
for (LoadBalancerRequestTransformer transformer : transformers) {
serviceRequest = transformer.transformRequest(serviceRequest, instance);
}
}
return execution.execute(serviceRequest, body);
}
};
}
}
