一、负载均衡1、RestTemplate在研究 eureka 源码上篇中，我们在 demo-consumer 消费者服务中定义了用 @LoadBalanced 标记的 RestTemplate，然后使用 RestTemplate 通过服务名的形式来调用远程服务 demo-producer，然后请求会轮询到两个 demo-producer 实例上。 RestTemplate 是 Spring Resources 中一个访问第三方 RESTful API 接口的网络请求框架。RestTemplate 是用来消费 REST 服务的，所以 RestTemplate 的主要方法都与 REST 的 Http协议的一些方法紧密相连，例如 HEAD、GET、POST、PUT、DELETE 和 OPTIONS 等方法，这些方法在 RestTemplate 类对应的方法为 headForHeaders()、getForObject()、postForObject()、put() 和 delete() 等。 RestTemplate 本身是不具备负载均衡的能力的，如果 RestTemplate 未使用 @LoadBalanced 标记，就通过服务名的形式来调用，必然会报错。用 @LoadBalanced 标记后，调用 RestTemplate 的 REST 方法就会通过负载均衡的方式通过一定的策略路由到某个服务实例上，底层负责负载均衡的组件就是 Ribbon。后面我们再来看 @LoadBalanced 是如何让 RestTemplate 具备负载均衡的能力的。 1 @SpringBootApplication 2 public class ConsumerApplication { 3 4 @Bean 5 @LoadBalanced 6 public RestTemplate restTemplate() { 7 return new RestTemplate(); 8 } 9 10 public static void main(String[] args) {11 SpringApplication.run(ConsumerApplication.class, args);12 }13 }14 15 @RestController16 public class DemoController {17 private final Logger logger = LoggerFactory.getLogger(getClass());18 19 @Autowired20 private RestTemplate restTemplate;21 22 @GetMapping("/v1/id")23 public ResponseEntity<String> getId() {24 ResponseEntity<String> result = restTemplate.getForEntity("http://demo-producer/v1/uuid", String.class);25 String uuid = result.getBody();26 logger.info("request id: {}", uuid);27 return ResponseEntity.ok(uuid);28 }29 }2、Ribbon 与负载均衡① 负载均衡 负载均衡是指将负载分摊到多个执行单元上，负载均衡主要可以分为集中式负载均衡与进程内负载均衡：
Ribbon 是一个客户端负载均衡器，可以很好地控制 HTTP 和 TCP 客户端的负载均衡行为。Ribbon 是 Netflix 公司开源的一个负载均衡组件，已经整合到 SpringCloud 生态中，它在 Spring Cloud 生态内是一个不可缺少的组件，少了它，服务便不能横向扩展。 ② Ribbon 模块 Ribbon 有很多子模块，官方文档中说明，目前 Netflix 公司主要用于生产环境的 Ribbon 子模块如下：
③ springcloud 与 ribbon 整合 与 eureka 整合到 springcloud 类似，springcloud 提供了对应的 spring-cloud-starter-netflix-eureka-client(server) 依赖包，ribbon 则整合到了 spring-cloud-starter-netflix-ribbon 中。一般也不需要单独引入 ribbon 的依赖包，spring-cloud-starter-netflix-eureka-client 中已经依赖了 spring-cloud-starter-netflix-ribbon。因此我们引入了 spring-cloud-starter-netflix-eureka-client 就可以使用 Ribbon 的功能了。
④ Ribbon 与 RestTemplate 整合使用 在 Spring Cloud 构建的微服务系统中，Ribbon 作为服务消费者的负载均衡器，有两种使用方式，一种是和 RestTemplate 相结合，另一种是和 Feign 相结合。前面已经演示了带有负载均衡的 RestTemplate 的使用，下面用一张图来看看 RestTemplate 基于 Ribbon 的远程调用。
二、RestTemplate 负载均衡1、@LoadBalanced 注解以 RestTemplate 为切入点，来看 Ribbon 的负载均衡核心原理。那么首先就要先看看 @LoadBalanced 注解如何让 RestTemplate 具备负载均衡的能力了。 首先看 @LoadBalanced 这个注解的定义，可以得到如下信息：
1 /** 2 * Annotation to mark a RestTemplate or WebClient bean to be configured to use a LoadBalancerClient. 3 * @author Spencer Gibb 4 */ 5 @Target({ ElementType.FIELD, ElementType.PARAMETER, ElementType.METHOD }) 6 @Retention(RetentionPolicy.RUNTIME) 7 @Documented 8 @Inherited 9 @Qualifier10 public @interface LoadBalanced {11 12 }注意 @LoadBalanced 是 spring-cloud-commons 模块下 loadbalancer 包下的。
2、RestTemplate 负载均衡自动化配置在 @LoadBalanced 同包下，有一个 LoadBalancerAutoConfiguration 自动化配置类，从注释也可以看出，这是客户端负载均衡 Ribbon 的自动化配置类。 从这个自动化配置类可以得到如下信息：
1 /** 2 * Auto-configuration for Ribbon (client-side load balancing). 3 */ 4 @Configuration(proxyBeanMethods = false) 5 @ConditionalOnClass(RestTemplate.class) // 有 RestTemplate 的依赖 6 @ConditionalOnBean(LoadBalancerClient.class) // 定义了 LoadBalancerClient 的 bean 对象 7 @EnableConfigurationProperties(LoadBalancerRetryProperties.class) 8 public class LoadBalancerAutoConfiguration { 9 10 // 注入 @LoadBalanced 标记的 RestTemplate 对象11 @LoadBalanced12 @Autowired(required = false)13 private List<RestTemplate> restTemplates = Collections.emptyList();14 15 @Autowired(required = false)16 private List<LoadBalancerRequestTransformer> transformers = Collections.emptyList();17 18 @Bean19 public SmartInitializingSingleton loadBalancedRestTemplateInitializerDeprecated(20 final ObjectProvider<List<RestTemplateCustomizer>> restTemplateCustomizers) {21 return () -> restTemplateCustomizers.ifAvailable(customizers -> {22 for (RestTemplate restTemplate : LoadBalancerAutoConfiguration.this.restTemplates) {23 for (RestTemplateCustomizer customizer : customizers) {24 // 利用 RestTemplateCustomizer 定制化 restTemplate25 customizer.customize(restTemplate);26 }27 }28 });29 }30 31 @Bean32 @ConditionalOnMissingBean33 public LoadBalancerRequestFactory loadBalancerRequestFactory(34 LoadBalancerClient loadBalancerClient) {35 return new LoadBalancerRequestFactory(loadBalancerClient, this.transformers);36 }37 38 @Configuration(proxyBeanMethods = false)39 @ConditionalOnMissingClass("org.springframework.retry.support.RetryTemplate")40 static class LoadBalancerInterceptorConfig {41 42 // 创建 LoadBalancerInterceptor 需要 LoadBalancerClient 和 LoadBalancerRequestFactory43 @Bean44 public LoadBalancerInterceptor ribbonInterceptor(45 LoadBalancerClient loadBalancerClient,46 LoadBalancerRequestFactory requestFactory) {47 return new LoadBalancerInterceptor(loadBalancerClient, requestFactory);48 }49 50 @Bean51 @ConditionalOnMissingBean52 public RestTemplateCustomizer restTemplateCustomizer(53 final LoadBalancerInterceptor loadBalancerInterceptor) {54 return restTemplate -> {55 List<ClientHttpRequestInterceptor> list = new ArrayList<>(56 restTemplate.getInterceptors());57 // 向 restTemplate 添加 LoadBalancerInterceptor 拦截器58 list.add(loadBalancerInterceptor);59 restTemplate.setInterceptors(list);60 };61 }62 63 }64 }3、RestTemplate 拦截器 LoadBalancerInterceptorLoadBalancerAutoConfiguration 自动化配置主要就是给 RestTemplate 添加了一个负载均衡拦截器 LoadBalancerInterceptor。从 setInterceptors 的参数可以看出，拦截器的类型是 ClientHttpRequestInterceptor，如果我们想定制化 RestTemplate，就可以实现这个接口来定制化，然后还可以用 @Order 标记拦截器的先后顺序。 1 public void setInterceptors(List<ClientHttpRequestInterceptor> interceptors) {2 if (this.interceptors != interceptors) {3 this.interceptors.clear();4 this.interceptors.addAll(interceptors);5 // 根据 @Order 注解的顺序排序6 AnnotationAwareOrderComparator.sort(this.interceptors);7 }8 }interceptors 拦截器是在 RestTemplate 的父类 InterceptingHttpAccessor 中的， RestTemplate 的类结构如下图所示。
从 restTemplate.getForEntity("http://demo-producer/v1/uuid", String.class) 这个GET请求进去看看，是如何使用 LoadBalancerInterceptor 的。一步步进去，可以看到最终是进入到 doExecute 这个方法了。 在 doExecute 方法中，首先根据 url、method 创建一个 ClientHttpRequest，然后利用 ClientHttpRequest 来发起请求。 1 protected <T> T doExecute(URI url, @Nullable HttpMethod method, @Nullable RequestCallback requestCallback, 2 @Nullable ResponseExtractor<T> responseExtractor) throws RestClientException { 3 ClientHttpResponse response = null; 4 try { 5 // 创建一个 ClientHttpRequest 6 ClientHttpRequest request = createRequest(url, method); 7 if (requestCallback != null) { 8 requestCallback.doWithRequest(request); 9 }10 // 调用 ClientHttpRequest 的 execute() 方法11 response = request.execute();12 // 处理返回结果13 handleResponse(url, method, response);14 return (responseExtractor != null ? responseExtractor.extractData(response) : null);15 }16 catch (IOException ex) {17 // ...18 }19 finally {20 if (response != null) {21 response.close();22 }23 }24 }25 26 //////////////////////////////////////27 28 protected ClientHttpRequest createRequest(URI url, HttpMethod method) throws IOException {29 ClientHttpRequest request = getRequestFactory().createRequest(url, method);30 initialize(request);31 if (logger.isDebugEnabled()) {32 logger.debug("HTTP " + method.name() + " " + url);33 }34 return request;35 }InterceptingHttpAccessor 中重写了父类 HttpAccessor 的 getRequestFactory 方法，父类默认的 requestFactory 是 SimpleClientHttpRequestFactory。 重写后的 getRequestFactory 方法中，如果拦截器不为空，则基于父类默认的 SimpleClientHttpRequestFactory 和拦截器创建了 InterceptingClientHttpRequestFactory。 1 public ClientHttpRequestFactory getRequestFactory() { 2 List<ClientHttpRequestInterceptor> interceptors = getInterceptors(); 3 if (!CollectionUtils.isEmpty(interceptors)) { 4 ClientHttpRequestFactory factory = this.interceptingRequestFactory; 5 if (factory == null) { 6 // 传入 SimpleClientHttpRequestFactory 和 ClientHttpRequestInterceptor 拦截器 7 factory = new InterceptingClientHttpRequestFactory(super.getRequestFactory(), interceptors); 8 this.interceptingRequestFactory = factory; 9 }10 return factory;11 }12 else {13 return super.getRequestFactory();14 }15 }也就是说调用了 InterceptingClientHttpRequestFactory 的 createRequest 方法来创建 ClientHttpRequest。进去可以看到，ClientHttpRequest 的实际类型就是 InterceptingClientHttpRequest。 1 protected ClientHttpRequest createRequest(URI uri, HttpMethod httpMethod, ClientHttpRequestFactory requestFactory) {2 return new InterceptingClientHttpRequest(requestFactory, this.interceptors, uri, httpMethod);3 }InterceptingClientHttpRequest 的类结构如下：
RestTemplate 的 doExecute 中调用 request.execute() 其实是调用了 InterceptingClientHttpRequest 父类 AbstractClientHttpRequest 中的 execute 方法。一步步进去可以发现最终其实是调用了 InterceptingClientHttpRequest 的 executeInternal 方法。 在 InterceptingClientHttpRequest 的 executeInternal 方法中，创建了 InterceptingRequestExecution 来执行请求。在 InterceptingRequestExecution 的 execute 方法中，会先遍历执行所有拦截器，然后通过 ClientHttpRequest 发起真正的 http 请求。 1 protected final ClientHttpResponse executeInternal(HttpHeaders headers, byte[] bufferedOutput) throws IOException { 2 // 创建 InterceptingRequestExecution 3 InterceptingRequestExecution requestExecution = new InterceptingRequestExecution(); 4 // 请求调用 5 return requestExecution.execute(this, bufferedOutput); 6 } 7 8 private class InterceptingRequestExecution implements ClientHttpRequestExecution { 9 10 private final Iterator<ClientHttpRequestInterceptor> iterator;11 12 public InterceptingRequestExecution() {13 // 拦截器迭代器14 this.iterator = interceptors.iterator();15 }16 17 @Override18 public ClientHttpResponse execute(HttpRequest request, byte[] body) throws IOException {19 if (this.iterator.hasNext()) {20 ClientHttpRequestInterceptor nextInterceptor = this.iterator.next();21 // 利用拦截器拦截处理，并传入 InterceptingRequestExecution22 return nextInterceptor.intercept(request, body, this);23 }24 else {25 // 拦截器遍历完后开始发起真正的 http 请求26 HttpMethod method = request.getMethod();27 Assert.state(method != null, "No standard HTTP method");28 ClientHttpRequest delegate = requestFactory.createRequest(request.getURI(), method);29 request.getHeaders().forEach((key, value) -> delegate.getHeaders().addAll(key, value));30 if (body.length > 0) {31 if (delegate instanceof StreamingHttpOutputMessage) {32 StreamingHttpOutputMessage streamingOutputMessage = (StreamingHttpOutputMessage) delegate;33 streamingOutputMessage.setBody(outputStream -> StreamUtils.copy(body, outputStream));34 }35 else {36 StreamUtils.copy(body, delegate.getBody());37 }38 }39 return delegate.execute();40 }41 }42 }进入到 LoadBalancerInterceptor 的 intercept 拦截方法内，可以看到从请求的原始地址中获取了服务名称，然后调用了 loadBalancer 的 execute 方法，也就是 LoadBalancerClient。 到这里，其实已经可以想象，loadBalancer.execute 这行代码就是根据服务名称去获取一个具体的实例，然后将原始地址替换为实例的IP地址。那这个 loadBalancer 又是什么呢？ 1 public ClientHttpResponse intercept(final HttpRequest request, final byte[] body, final ClientHttpRequestExecution execution) throws IOException { 2 // 原始地址：http://demo-producer/v1/uuid 3 final URI originalUri = request.getURI(); 4 // host 就是服务名：demo-producer 5 String serviceName = originalUri.getHost(); 6 Assert.state(serviceName != null, 7 "Request URI does not contain a valid hostname: " + originalUri); 8 return this.loadBalancer.execute(serviceName, 9 this.requestFactory.createRequest(request, body, execution));10 }4、负载均衡客户端 LoadBalancerClient在配置 LoadBalancerInterceptor 时，需要两个参数，LoadBalancerClient 和 LoadBalancerRequestFactory，LoadBalancerRequestFactory前面已经知道是如何创建的了。LoadBalancerClient 又是在哪创建的呢？通过 IDEA 搜索，可以发现是在 spring-cloud-netflix-ribbon 模块下的 RibbonAutoConfiguration 中配置的，可以看到 LoadBalancerClient 的实际类型是 RibbonLoadBalancerClient。 配置类的顺序是 EurekaClientAutoConfiguration、RibbonAutoConfiguration、LoadBalancerAutoConfiguration，因为使 RestTemplate 具备负载均衡的能力需要 LoadBalancerInterceptor 拦截器，创建 LoadBalancerInterceptor 又需要 LoadBalancerClient，而 LoadBalancerClient 底层要根据服务名获取某个实例，肯定又需要一个实例库，比如从配置文件、注册中心获取。从这里就可以看出来，RibbonLoadBalancerClient 默认会从 Eureka 注册中心获取实例。 1 @Configuration 2 @Conditional(RibbonAutoConfiguration.RibbonClassesConditions.class) 3 @RibbonClients 4 // 后于 EurekaClientAutoConfiguration 配置 5 @AutoConfigureAfter(name = "org.springframework.cloud.netflix.eureka.EurekaClientAutoConfiguration") 6 // 先于 LoadBalancerAutoConfiguration 配置 7 @AutoConfigureBefore({ LoadBalancerAutoConfiguration.class, AsyncLoadBalancerAutoConfiguration.class }) 8 @EnableConfigurationProperties({ RibbonEagerLoadProperties.class, ServerIntrospectorProperties.class }) 9 public class RibbonAutoConfiguration {10 11 @Autowired(required = false)12 private List<RibbonClientSpecification> configurations = new ArrayList<>();13 14 @Bean15 @ConditionalOnMissingBean16 public SpringClientFactory springClientFactory() {17 SpringClientFactory factory = new SpringClientFactory();18 factory.setConfigurations(this.configurations);19 return factory;20 }21 22 @Bean23 @ConditionalOnMissingBean(LoadBalancerClient.class)24 public LoadBalancerClient loadBalancerClient() {25 return new RibbonLoadBalancerClient(springClientFactory());26 }27 }LoadBalancerClient 主要提供了三个接口： 1 public interface LoadBalancerClient extends ServiceInstanceChooser { 2 3 // 从 LoadBalancer 找一个 Server 来发送请求 4 <T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException; 5 6 // 从传入的 ServiceInstance 取 Server 来发送请求 7 <T> T execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest<T> request) throws IOException; 8 9 // 对原始 URI 重构10 URI reconstructURI(ServiceInstance instance, URI original);11 }进入到 RibbonLoadBalancerClient 的 execute 方法中可以看到：
1 public <T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException { 2 return execute(serviceId, request, null); 3 } 4 5 public <T> T execute(String serviceId, LoadBalancerRequest<T> request, Object hint) 6 throws IOException { 7 // 根据服务名获取一个负载均衡器 ILoadBalancer 8 ILoadBalancer loadBalancer = getLoadBalancer(serviceId); 9 // 利用负载均衡器获取实例 Server10 Server server = getServer(loadBalancer, hint);11 if (server == null) {12 throw new IllegalStateException("No instances available for " + serviceId);13 }14 // 封装实例信息：RibbonServer 的父类是 ServiceInstance15 RibbonServer ribbonServer = new RibbonServer(serviceId, server,16 isSecure(server, serviceId),17 serverIntrospector(serviceId).getMetadata(server));18 return execute(serviceId, ribbonServer, request);19 }20 21 @Override22 public <T> T execute(String serviceId, ServiceInstance serviceInstance,23 LoadBalancerRequest<T> request) throws IOException {24 Server server = null;25 if (serviceInstance instanceof RibbonServer) {26 server = ((RibbonServer) serviceInstance).getServer();27 }28 if (server == null) {29 throw new IllegalStateException("No instances available for " + serviceId);30 }31 32 try {33 // 处理地址，将服务名替换为真实的IP地址34 T returnVal = request.apply(serviceInstance);35 return returnVal;36 } catch (Exception ex) {37 // ...38 }39 return null;40 }这个 LoadBalancerRequest 其实就是 LoadBalancerInterceptor 的 intercept 中创建的一个匿名类，在它的函数式接口内，主要是用装饰器 ServiceRequestWrapper 将 request 包了一层。 1 public LoadBalancerRequest<ClientHttpResponse> createRequest(final HttpRequest request, final byte[] body, final ClientHttpRequestExecution execution) { 2 return instance -> { 3 // 封装 HttpRequest，ServiceRequestWrapper 重载了 getURI 方法。 4 HttpRequest serviceRequest = new ServiceRequestWrapper(request, instance, this.loadBalancer); 5 if (this.transformers != null) { 6 for (LoadBalancerRequestTransformer transformer : this.transformers) { 7 serviceRequest = transformer.transformRequest(serviceRequest, instance); 8 } 9 }10 // 继续执行拦截器11 return execution.execute(serviceRequest, body);12 };13 }ServiceRequestWrapper 主要就是重写了 getURI 方法，在重写的 getURI 方法内，它用 loadBalancer 对 URI 进行了重构，进去可以发现，就是将原始地址中的服务名替换为 Server 的真实IP、端口地址。 1 @Override2 public URI getURI() {3 // 重构 URI4 URI uri = this.loadBalancer.reconstructURI(this.instance, getRequest().getURI());5 return uri;6 } 1 public URI reconstructURI(ServiceInstance instance, URI original) { 2 Assert.notNull(instance, "instance can not be null"); 3 // 服务名 4 String serviceId = instance.getServiceId(); 5 RibbonLoadBalancerContext context = this.clientFactory.getLoadBalancerContext(serviceId); 6 7 URI uri; 8 Server server; 9 if (instance instanceof RibbonServer) {10 RibbonServer ribbonServer = (RibbonServer) instance;11 server = ribbonServer.getServer();12 uri = updateToSecureConnectionIfNeeded(original, ribbonServer);13 }14 else {15 server = new Server(instance.getScheme(), instance.getHost(), instance.getPort());16 IClientConfig clientConfig = clientFactory.getClientConfig(serviceId);17 ServerIntrospector serverIntrospector = serverIntrospector(serviceId);18 uri = updateToSecureConnectionIfNeeded(original, clientConfig, serverIntrospector, server);19 }20 // 重构地址21 return context.reconstructURIWithServer(server, uri);22 }reconstructURIWithServer：  1 public URI reconstructURIWithServer(Server server, URI original) { 2 String host = server.getHost(); 3 int port = server.getPort(); 4 String scheme = server.getScheme(); 5 6 if (host.equals(original.getHost())
7 && port == original.getPort() 8 && scheme == original.getScheme()) { 9 return original;10 }11 if (scheme == null) {12 scheme = original.getScheme();13 }14 if (scheme == null) {15 scheme = deriveSchemeAndPortFromPartialUri(original).first();16 }17 18 try {19 StringBuilder sb = new StringBuilder();20 sb.append(scheme).append("://");21 if (!Strings.isNullOrEmpty(original.getRawUserInfo())) {22 sb.append(original.getRawUserInfo()).append("@");23 }24 sb.append(host);25 if (port >= 0) {26 sb.append(":").append(port);27 }28 sb.append(original.getRawPath());29 if (!Strings.isNullOrEmpty(original.getRawQuery())) {30 sb.append("?").append(original.getRawQuery());31 }32 if (!Strings.isNullOrEmpty(original.getRawFragment())) {33 sb.append("#").append(original.getRawFragment());34 }35 URI newURI = new URI(sb.toString());36 return newURI;
37 } catch (URISyntaxException e) {38 throw new RuntimeException(e);39 }40 }5、RestTemplate 负载均衡总结到这里，我们基本就弄清楚了一个简单的 @LoadBalanced 注解如何让 RestTemplate 具备了负载均衡的能力了，这一节来做个小结。 ① RestTemplate 如何获得负载均衡的能力
② 核心组件 这里会涉及多个模块，下面是核心组件的所属模块： spring-web：
spring-cloud-commons：
spring-cloud-netflix-ribbon：
ribbon-loadbalancer：
③ 最后再用一张图把 RestTemplate 这块的关系捋一下
三、ILoadBalancer 获取 Server从前面 RestTemplate 那张图可以看出，使 RestTemplate 具备负载均衡的能力，最重要的一个组件之一就是 ILoadBalancer，因为要用它来获取能调用的 Server，有了 Server 才能对原始带有服务名的 URI 进行重构。这节就来看下 Ribbon 的负载均衡器 ILoadBalancer 是如何创建的以及如何通过它获取 Server。 1、创建负载均衡器 ILoadBalancer① SpringClientFactory与上下文 ILoadBalancer 是用 SpringClientFactory 的 getLoadBalancer 方法根据服务名获取的，从 getInstance 一步步进去可以发现，每个服务都会创建一个 AnnotationConfigApplicationContext，也就是一个应用上下文 ApplicationContext。相当于就是一个服务绑定一个 ILoadBalancer。 1 public <C> C getInstance(String name, Class<C> type) {2 C instance = super.getInstance(name, type);3 if (instance != null) {4 return instance;5 }6 IClientConfig config = getInstance(name, IClientConfig.class);7 return instantiateWithConfig(getContext(name), type, config);8 }1 public <T> T getInstance(String name, Class<T> type) {2 // 根据名称获取3 AnnotationConfigApplicationContext context = getContext(name);4 if (BeanFactoryUtils.beanNamesForTypeIncludingAncestors(context, type).length > 0) {5 return context.getBean(type);6 }7 return null;8 } 1 protected AnnotationConfigApplicationContext getContext(String name) { 2 // contexts => Map<String, AnnotationConfigApplicationContext> 3 if (!this.contexts.containsKey(name)) { 4 synchronized (this.contexts) { 5 if (!this.contexts.containsKey(name)) { 6 this.contexts.put(name, createContext(name)); 7 } 8 } 9 }10 return this.contexts.get(name);11 }调试看下 AnnotationConfigApplicationContext 上下文，可以看到放入了与这个服务绑定的 ILoadBalancer、IClientConfig、RibbonLoadBalancerContext 等。 它这里为什么要每个服务都绑定一个 ApplicationContext 呢？我猜想应该是因为服务实例列表可以有多个来源，比如可以从 eureka 注册中心获取、可以通过代码配置、可以通过配置文件配置，另外每个服务还可以有很多个性化的配置，有默认的配置、定制的全局配置、个别服务的特定配置等，它这样做就便于用户定制每个服务的负载均衡策略。
② Ribbon的饥饿加载 而且这个Ribbon客户端的应用上下文默认是懒加载的，并不是在启动的时候就加载上下文，而是在第一次调用的时候才会去初始化。 如果想服务启动时就初始化，可以指定Ribbon客户端的具体名称，在启动的时候就加载配置项的上下文： 1 ribbon:2 eager-load:3 enabled: true4 clients: demo-producer,demo-xxx 在 RibbonAutoConfiguration 配置类中可以找到这个饥饿配置，如果开启了饥饿加载，就会创建 RibbonApplicationContextInitializer 来在启动时初始化上下文。 1 @Bean 2 @ConditionalOnProperty("ribbon.eager-load.enabled") 3 public RibbonApplicationContextInitializer ribbonApplicationContextInitializer() { 4 return new RibbonApplicationContextInitializer(springClientFactory(), ribbonEagerLoadProperties.getClients()); 5 } 6 7 public class RibbonApplicationContextInitializer implements ApplicationListener<ApplicationReadyEvent> { 8 private final SpringClientFactory springClientFactory; 9 10 // List of Ribbon client names11 private final List<String> clientNames;12 13 public RibbonApplicationContextInitializer(SpringClientFactory springClientFactory, List<String> clientNames) {14 this.springClientFactory = springClientFactory;15 this.clientNames = clientNames;16 }17 18 protected void initialize() {19 if (clientNames != null) {20 for (String clientName : clientNames) {21 // 提前初始化上下文22 this.springClientFactory.getContext(clientName);23 }24 }25 }26 27 @Override28 public void onApplicationEvent(ApplicationReadyEvent event) {29 initialize();30 }31 }③ RibbonClientConfiguration ILoadBalancer 的创建在哪呢？看 RibbonClientConfiguration，这个配置类提供了 ILoadBalancer 的默认创建方法，ILoadBalancer 的默认实现类为 ZoneAwareLoadBalancer。 1 public class RibbonClientConfiguration { 2 3 public static final int DEFAULT_CONNECT_TIMEOUT = 1000; 4 5 public static final int DEFAULT_READ_TIMEOUT = 1000; 6 7 public static final boolean DEFAULT_GZIP_PAYLOAD = true; 8 9 @RibbonClientName10 private String name = "client";11 12 @Autowired13 private PropertiesFactory propertiesFactory;14 15 @Bean16 @ConditionalOnMissingBean17 public IClientConfig ribbonClientConfig() {18 DefaultClientConfigImpl config = new DefaultClientConfigImpl();19 config.loadProperties(this.name);20 config.set(CommonClientConfigKey.ConnectTimeout, DEFAULT_CONNECT_TIMEOUT);21 config.set(CommonClientConfigKey.ReadTimeout, DEFAULT_READ_TIMEOUT);22 config.set(CommonClientConfigKey.GZipPayload, DEFAULT_GZIP_PAYLOAD);23 return config;24 }25 26 @Bean27 @ConditionalOnMissingBean28 public IRule ribbonRule(IClientConfig config) {29 if (this.propertiesFactory.isSet(IRule.class, name)) {30 return this.propertiesFactory.get(IRule.class, config, name);31 }32 ZoneAvoidanceRule rule = new ZoneAvoidanceRule();33 rule.initWithNiwsConfig(config);34 return rule;35 }36 37 @Bean38 @ConditionalOnMissingBean39 public IPing ribbonPing(IClientConfig config) {40 if (this.propertiesFactory.isSet(IPing.class, name)) {41 return this.propertiesFactory.get(IPing.class, config, name);42 }43 return new DummyPing();44 }45 46 @Bean47 @ConditionalOnMissingBean48 @SuppressWarnings("unchecked")49 public ServerList<Server> ribbonServerList(IClientConfig config) {50 if (this.propertiesFactory.isSet(ServerList.class, name)) {51 return this.propertiesFactory.get(ServerList.class, config, name);52 }53 ConfigurationBasedServerList serverList = new ConfigurationBasedServerList();54 serverList.initWithNiwsConfig(config);55 return serverList;56 }57 58 @Bean59 @ConditionalOnMissingBean60 public ServerListUpdater ribbonServerListUpdater(IClientConfig config) {61 return new PollingServerListUpdater(config);62 }63 64 @Bean65 @ConditionalOnMissingBean66 @SuppressWarnings("unchecked")67 public ServerListFilter<Server> ribbonServerListFilter(IClientConfig config) {68 if (this.propertiesFactory.isSet(ServerListFilter.class, name)) {69 return this.propertiesFactory.get(ServerListFilter.class, config, name);70 }71 ZonePreferenceServerListFilter filter = new ZonePreferenceServerListFilter();72 filter.initWithNiwsConfig(config);73 return filter;74 }75 76 @Bean77 @ConditionalOnMissingBean78 public ILoadBalancer ribbonLoadBalancer(IClientConfig config,79 ServerList<Server> serverList, ServerListFilter<Server> serverListFilter,80 IRule rule, IPing ping, ServerListUpdater serverListUpdater) {81 // 先判断配置文件中是否配置了负载均衡器82 if (this.propertiesFactory.isSet(ILoadBalancer.class, name)) {83 // 通过反射创建84 return this.propertiesFactory.get(ILoadBalancer.class, config, name);85 }86 return new ZoneAwareLoadBalancer<>(config, rule, ping, serverList,87 serverListFilter, serverListUpdater);88 }89 }可以看到创建 ILoadBalancer 需要 IClientConfig、ServerList<Server>、ServerListFilter<Server>、IRule、IPing、ServerListUpdater，其实这6个接口加上 ILoadBalancer 就是 Ribbon 的核心接口，它们共同定义了 Ribbon 的行为特性。 这7个核心接口和默认实现类如下：
2、客户端 Ribbon 定制可以看到在 RibbonClientConfiguration 中创建 IRule、IPing、ServerList<Server>、ServerListFilter<Server>、ILoadBalancer 时，都先通过 propertiesFactory.isSet 判断是否已配置了对应类型的实现类，没有才使用默认的实现类。 也就是说针对特定的服务，这几个类可以自行定制化，也可以通过配置指定其它的实现类。
① 全局策略配置 如果想要全局更改配置，需要加一个配置类，比如像下面这样： 1 @Configuration 2 public class GlobalRibbonConfiguration { 3 4 @Bean 5 public IRule ribbonRule() { 6 return new RandomRule(); 7 } 8 9 @Bean10 public IPing ribbonPing() {11 return new NoOpPing();12 }13 }② 基于注解的配置 如果想针对某一个服务定制配置，可以通过 @RibbonClients 来配置特定服务的配置类。 需要先定义一个服务配置类： 1 @Configuration 2 public class ProducerRibbonConfiguration { 3 4 @Bean 5 public IRule ribbonRule() { 6 return new RandomRule(); 7 } 8 9 @Bean10 public IPing ribbonPing() {11 return new NoOpPing();12 }13 }用 @RibbonClients 注解为服务指定特定的配置类，并排除掉，不让 Spring 扫描，否则就变成了全局配置了。 1 @RibbonClients({2 @RibbonClient(name = "demo-producer", configuration = ProducerRibbonConfiguration.class)3 })4 @ComponentScan(excludeFilters = {5 @ComponentScan.Filter(type = FilterType.ASSIGNABLE_TYPE, classes = ProducerRibbonConfiguration.class)6 })③ 配置文件配置 通过配置文件的方式来配置，配置的格式就是 <服务名称>.ribbon.<属性>： 1 demo-producer: 2 ribbon: 3 # ILoadBalancer 4 NFLoadBalancerClassName: com.netflix.loadbalancer.NoOpLoadBalancer 5 # IRule 6 NFLoadBalancerRuleClassName: com.netflix.loadbalancer.RandomRule 7 # IPing 8 NFLoadBalancerPingClassName: 9 # ServerList<Server>10 NIWSServerListClassName:11 # ServerListFilter<Server>12 NIWSServerListFilterClassName: ④ 优先级顺序 这几种配置方式的优先级顺序是 配置文件配置 > @RibbonClients 配置 > 全局配置 > 默认配置。 3、ZoneAwareLoadBalancer 选择 Server获取到 ILoadBalancer 后，就要去获取 Server 了，可以看到，就是用 ILoadBalancer 来获取 Server。 1 protected Server getServer(ILoadBalancer loadBalancer, Object hint) {2 if (loadBalancer == null) {3 return null;4 }5 // Use 'default' on a null hint, or just pass it on?6 return loadBalancer.chooseServer(hint != null ? hint : "default");7 }ILoadBalancer 的默认实现类是 ZoneAwareLoadBalancer，进入它的 chooseServer 方法内，如果只配置了一个 zone，就走父类的 chooseServer，否则从多个 zone 中去选择实例。 1 public Server chooseServer(Object key) { 2 // ENABLED => ZoneAwareNIWSDiscoveryLoadBalancer.enabled 默认 true 3 // AvailableZones 配置的只有一个 defaultZone 4 if (!ENABLED.get() || getLoadBalancerStats().getAvailableZones().size() <= 1) { 5 logger.debug("Zone aware logic disabled or there is only one zone"); 6 // 走父类获取 Server 的逻辑 7 return super.chooseServer(key); 8 } 9 10 // 多 zone 逻辑....11 }先看下 ZoneAwareLoadBalancer 的类继承结构，ZoneAwareLoadBalancer 的直接父类是 DynamicServerListLoadBalancer，DynamicServerListLoadBalancer 的父类又是 BaseLoadBalancer。
ZoneAwareLoadBalancer 调用父类的 chooseServer 方法是在 BaseLoadBalancer 中的，进去可以看到，它主要是用 IRule 来选择实例，最终选择实例的策略就交给了 IRule 接口。 1 public Server chooseServer(Object key) { 2 if (counter == null) { 3 counter = createCounter(); 4 } 5 counter.increment(); 6 if (rule == null) { 7 return null; 8 } else { 9 try {10 // IRule11 return rule.choose(key);12 } catch (Exception e) {13 logger.warn("LoadBalancer [{}]: Error choosing server for key {}", name, key, e);14 return null;15 }16 }17 }4、ZoneAvoidanceRule 断言筛选、轮询选择 ServerIRule 的默认实现类是 ZoneAvoidanceRule，先看下 ZoneAvoidanceRule 的继承结构，ZoneAvoidanceRule 的直接父类是 PredicateBasedRule。
rule.choose 的逻辑在 PredicateBasedRule 中，getPredicate() 返回的是 ZoneAvoidanceRule 创建的一个组合断言 CompositePredicate，就是用这个断言来过滤出可用的 Server，并通过轮询的策略返回一个 Server。 1 public Server choose(Object key) { 2 ILoadBalancer lb = getLoadBalancer(); 3 // getPredicate() Server断言 => CompositePredicate 4 // RoundRobin 轮询方式获取实例 5 Optional<Server> server = getPredicate().chooseRoundRobinAfterFiltering(lb.getAllServers(), key); 6 if (server.isPresent()) { 7 return server.get(); 8 } else { 9 return null;10 }
11 }在初始化 ZoneAvoidanceRule 配置时，创建了 CompositePredicate，可以看到这个组合断言主要有两个断言，一个是断言 Server 的 zone 是否可用，一个断言 Server 本身是否可用，例如 Server 无法 ping 通。 1 public void initWithNiwsConfig(IClientConfig clientConfig) { 2 // 断言 Server 的 zone 是否可用，只有一个 defaultZone 的情况下都是可用的 3 ZoneAvoidancePredicate zonePredicate = new ZoneAvoidancePredicate(this, clientConfig); 4 // 断言 Server 是否可用 5 AvailabilityPredicate availabilityPredicate = new AvailabilityPredicate(this, clientConfig); 6 // 封装组合断言 7 compositePredicate = createCompositePredicate(zonePredicate, availabilityPredicate); 8 } 9 10 private CompositePredicate createCompositePredicate(ZoneAvoidancePredicate p1, AvailabilityPredicate p2) {11 // 建造者模式创建断言12 return CompositePredicate.withPredicates(p1, p2)13 .addFallbackPredicate(p2)14 .addFallbackPredicate(AbstractServerPredicate.alwaysTrue())15 .build();16 17 }接着看选择Server的 chooseRoundRobinAfterFiltering，参数 servers 是通过 ILoadBalancer 获取的所有实例，可以看到它其实就是返回了 ILoadBalancer 在内存中缓存的服务所有 Server。这个 Server 从哪来的我们后面再来看。 1 public List<Server> getAllServers() {2 // allServerList => List<Server>3 return Collections.unmodifiableList(allServerList);4 }先对所有实例通过断言过滤掉不可用的 Server，然后是通过轮询的方式获取一个 Server 返回。这就是默认配置下 ILoadBalancer（ZoneAwareLoadBalancer） 通过 IRule（ZoneAvoidanceRule） 选择 Server 的流程了。 1 public Optional<Server> chooseRoundRobinAfterFiltering(List<Server> servers, Object loadBalancerKey) { 2 // 断言获取可用的 Server 3 List<Server> eligible = getEligibleServers(servers, loadBalancerKey); 4 if (eligible.size() == 0) { 5 return Optional.absent(); 6 } 7 // 通过取模的方式轮询 Server 8 return Optional.of(eligible.get(incrementAndGetModulo(eligible.size()))); 9 }10 11 public List<Server> getEligibleServers(List<Server> servers, Object loadBalancerKey) {12 if (loadBalancerKey == null) {13 return ImmutableList.copyOf(Iterables.filter(servers, this.getServerOnlyPredicate()));
14 } else {15 List<Server> results = Lists.newArrayList();16 // 对每个 Server 断言17 for (Server server: servers) {18 if (this.apply(new PredicateKey(loadBalancerKey, server))) {19 results.add(server);20 }21 }22 return results;
23 }24 }25 26 private int incrementAndGetModulo(int modulo) {27 for (;;) {28 int current = nextIndex.get();29 // 模运算取余数30 int next = (current + 1) % modulo;31 // CAS 更新 nextIndex32 if (nextIndex.compareAndSet(current, next) && current < modulo)33 return current;34 }35 }四、Ribbon 整合 Eureka Client 拉取Server列表前面在通过 IRule 选择 Server 的时候，首先通过 lb.getAllServers() 获取了所有的 Server，那这些 Server 从哪里来的呢，这节就来看下。 1、ILoadBalancer 初始化ILoadBalancer 的默认实现类是 ZoneAwareLoadBalancer，先从 ZoneAwareLoadBalancer 的构造方法进去看看都做了些什么事情。 1 @Bean 2 @ConditionalOnMissingBean 3 public ILoadBalancer ribbonLoadBalancer(IClientConfig config, 4 ServerList<Server> serverList, ServerListFilter<Server> serverListFilter, 5 IRule rule, IPing ping, ServerListUpdater serverListUpdater) { 6 if (this.propertiesFactory.isSet(ILoadBalancer.class, name)) { 7 return this.propertiesFactory.get(ILoadBalancer.class, config, name); 8 } 9 return new ZoneAwareLoadBalancer<>(config, rule, ping, serverList,10 serverListFilter, serverListUpdater);11 }可以看到，ZoneAwareLoadBalancer 直接调用了父类 DynamicServerListLoadBalancer 的构造方法，DynamicServerListLoadBalancer 先调用父类 BaseLoadBalancer 初始化，然后又做了一些剩余的初始化工作。 1 public ZoneAwareLoadBalancer(IClientConfig clientConfig, IRule rule, 2 IPing ping, ServerList<T> serverList, ServerListFilter<T> filter, 3 ServerListUpdater serverListUpdater) { 4 // DynamicServerListLoadBalancer 5 super(clientConfig, rule, ping, serverList, filter, serverListUpdater); 6 } 7 8 public DynamicServerListLoadBalancer(IClientConfig clientConfig, IRule rule, IPing ping, 9 ServerList<T> serverList, ServerListFilter<T> filter,10 ServerListUpdater serverListUpdater) {11 // BaseLoadBalancer12 super(clientConfig, rule, ping);13 this.serverListImpl = serverList;14 this.filter = filter;15 this.serverListUpdater = serverListUpdater;16 if (filter instanceof AbstractServerListFilter) {17 ((AbstractServerListFilter) filter).setLoadBalancerStats(getLoadBalancerStats());18 }19 // 剩余的一些初始化20 restOfInit(clientConfig);21 }22 23 public BaseLoadBalancer(IClientConfig config, IRule rule, IPing ping) {24 // createLoadBalancerStatsFromConfig => LoadBalancerStats 统计25 initWithConfig(config, rule, ping, createLoadBalancerStatsFromConfig(config));26看 BaseLoadBalancer 的 initWithConfig，主要做了如下的初始化：
1 void initWithConfig(IClientConfig clientConfig, IRule rule, IPing ping, LoadBalancerStats stats) { 2 this.config = clientConfig; 3 String clientName = clientConfig.getClientName(); 4 this.name = clientName; 5 // ping 间隔时间，默认30秒 6 int pingIntervalTime = Integer.parseInt("" 7 + clientConfig.getProperty( 8 CommonClientConfigKey.NFLoadBalancerPingInterval, 9 Integer.parseInt("30")));10 // 没看到用的地方 11 int maxTotalPingTime = Integer.parseInt(""12 + clientConfig.getProperty(13 CommonClientConfigKey.NFLoadBalancerMaxTotalPingTime,14 Integer.parseInt("2")));15 // 设置 ping 间隔时间，并重新设置了 ping 任务16 setPingInterval(pingIntervalTime);17 setMaxTotalPingTime(maxTotalPingTime);18 19 // 设置 IRule、IPing20 setRule(rule);21 setPing(ping);22 23 setLoadBalancerStats(stats);24 rule.setLoadBalancer(this);25 if (ping instanceof AbstractLoadBalancerPing) {26 ((AbstractLoadBalancerPing) ping).setLoadBalancer(this);27 }28 logger.info("Client: {} instantiated a LoadBalancer: {}", name, this);29 30 // PrimeConnections，请求预热，默认关闭31 // 作用主要用于解决那些部署环境(如读EC2)在实际使用实时请求之前，从防火墙连接/路径进行预热（比如先加白名单、初始化等等动作比较耗时，可以用它先去打通）。32 boolean enablePrimeConnections = clientConfig.get(33 CommonClientConfigKey.EnablePrimeConnections, DefaultClientConfigImpl.DEFAULT_ENABLE_PRIME_CONNECTIONS);34 if (enablePrimeConnections) {35 this.setEnablePrimingConnections(true);36 PrimeConnections primeConnections = new PrimeConnections(37 this.getName(), clientConfig);38 this.setPrimeConnections(primeConnections);39 }40 // 注册一些监控41 init();42 }43 44 protected void init() {45 Monitors.registerObject("LoadBalancer_" + name, this);46 // register the rule as it contains metric for available servers count47 Monitors.registerObject("Rule_" + name, this.getRule());48 // 默认关闭49 if (enablePrimingConnections && primeConnections != null) {50 primeConnections.primeConnections(getReachableServers());51 }52 }再看下 DynamicServerListLoadBalancer 的初始化，核心的初始化逻辑在 restOfInit 中，主要就是做了两件事情：
1 void restOfInit(IClientConfig clientConfig) { 2 boolean primeConnection = this.isEnablePrimingConnections(); 3 // turn this off to avoid duplicated asynchronous priming done in BaseLoadBalancer.setServerList() 4 this.setEnablePrimingConnections(false); 5 6 // 开启动态更新 Server 的特性 7 enableAndInitLearnNewServersFeature(); 8 9 // 更新 Server 列表10 updateListOfServers();11 12 // 开启请求预热的情况下，对可用的 Server 进行预热13 if (primeConnection && this.getPrimeConnections() != null) {14 this.getPrimeConnections()15 .primeConnections(getReachableServers());16 }17 this.setEnablePrimingConnections(primeConnection);18 LOGGER.info("DynamicServerListLoadBalancer for client {} initialized: {}", clientConfig.getClientName(), this.toString());19 }2、全量更新Server列表先看下 updateListOfServers() 是如何更新 Server 列表的，进而看下 ILoadBalancer 是如何存储 Server 的。
1 public void updateListOfServers() { 2 List<T> servers = new ArrayList<T>(); 3 if (serverListImpl != null) { 4 // 从 ServerList 获取所有 Server 列表 5 servers = serverListImpl.getUpdatedListOfServers(); 6 LOGGER.debug("List of Servers for {} obtained from Discovery client: {}", getIdentifier(), servers); 7 8 if (filter != null) { 9 // 用 ServerListFilter 过滤 Server10 servers = filter.getFilteredListOfServers(servers);11 LOGGER.debug("Filtered List of Servers for {} obtained from Discovery client: {}", getIdentifier(), servers);12 }13 }14 // 更新所有 Server 到本地缓存15 updateAllServerList(servers);16 }17 18 protected void updateAllServerList(List<T> ls) {19 if (serverListUpdateInProgress.compareAndSet(false, true)) {20 try {21 for (T s : ls) {22 s.setAlive(true); // 设置 Server alive23 }24 setServersList(ls);25 // 强制初始化 Ping26 super.forceQuickPing();27 } finally {28 serverListUpdateInProgress.set(false);29 }30 }31 }32 33 public void setServersList(List lsrv) {34 // BaseLoadBalancer35 super.setServersList(lsrv);36 37 // 将 Server 更新到 LoadBalancerStats 统计中 ....38 }接着看父类的 setServersList，可以看出，存储所有 Server 的数据结构 allServerList 是一个加了 synchronized 的线程安全的容器，setServersList 就是直接将得到的 Server 列表替换 allServerList。 1 public void setServersList(List lsrv) { 2 Lock writeLock = allServerLock.writeLock(); 3 ArrayList<Server> newServers = new ArrayList<Server>(); 4 // 加写锁 5 writeLock.lock(); 6 try { 7 // for 循环将 lsrv 中的 Server 转移到 allServers 8 ArrayList<Server> allServers = new ArrayList<Server>(); 9 for (Object server : lsrv) {10 if (server == null) {11 continue;12 }13 if (server instanceof String) {14 server = new Server((String) server);15 }16 if (server instanceof Server) {17 logger.debug("LoadBalancer [{}]: addServer [{}]", name, ((Server) server).getId());18 allServers.add((Server) server);19 } else {20 throw new IllegalArgumentException("Type String or Server expected, instead found:" + server.getClass());21 }22 }23 24 boolean listChanged = false;25 // allServerList => volatile List<Server> allServerList = Collections.synchronizedList(new ArrayList<Server>())26 if (!allServerList.equals(allServers)) {27 listChanged = true;28 // 服务列表变更监听器 ServerListChangeListener， 发出服务变更通知...29 }30 31 // 启用了服务预热，开始 Server 预热...32 33 // 直接替换34 allServerList = allServers;35 if (canSkipPing()) {36 for (Server s : allServerList) {37 s.setAlive(true);38 }39 upServerList = allServerList;40 } else if (listChanged) {41 forceQuickPing();42 }43 } finally {44 // 释放写锁45 writeLock.unlock();46 }47 }前面 chooseRoundRobinAfterFiltering 获取所有 Server 时就是返回的这个 allServerList列表。 1 public List<Server> getAllServers() {2 return Collections.unmodifiableList(allServerList);3 }3、Eureka Ribbon 客户端配置获取 Server 的组件是 ServerList，RibbonClientConfiguration 中配置的默认实现类是 ConfigurationBasedServerList。ConfigurationBasedServerList 默认是从配置文件中获取，可以像下面这样配置服务实例地址，多个 Server 地址用逗号隔开。 1 demo-producer:2 ribbon:3 listOfServers: http://10.215.0.92:8010,http://10.215.0.92:8011 但是和 eureka-client 结合后，也就是引入 spring-cloud-starter-netflix-eureka-client 的客户端依赖，它会帮我们引入 spring-cloud-netflix-eureka-client 依赖，这个包中有一个 RibbonEurekaAutoConfiguration 自动化配置类，它通过 @RibbonClients 注解定义了全局的 Ribbon 客户端配置类 为 EurekaRibbonClientConfiguration 1 @Configuration(proxyBeanMethods = false)2 @EnableConfigurationProperties3 @ConditionalOnRibbonAndEurekaEnabled4 @AutoConfigureAfter(RibbonAutoConfiguration.class)5 @RibbonClients(defaultConfiguration = EurekaRibbonClientConfiguration.class)6 public class RibbonEurekaAutoConfiguration {7 8 }进入 EurekaRibbonClientConfiguration 可以看到：
1 @Configuration(proxyBeanMethods = false) 2 public class EurekaRibbonClientConfiguration { 3 @Value("${ribbon.eureka.approximateZoneFromHostname:false}") 4 private boolean approximateZoneFromHostname = false; 5 6 @RibbonClientName 7 private String serviceId = "client"; 8 @Autowired 9 private PropertiesFactory propertiesFactory;10 11 @Bean12 @ConditionalOnMissingBean13 public IPing ribbonPing(IClientConfig config) {14 if (this.propertiesFactory.isSet(IPing.class, serviceId)) {15 return this.propertiesFactory.get(IPing.class, config, serviceId);16 }17 NIWSDiscoveryPing ping = new NIWSDiscoveryPing();18 ping.initWithNiwsConfig(config);19 return ping;20 }21 22 @Bean23 @ConditionalOnMissingBean24 public ServerList<?> ribbonServerList(IClientConfig config,25 Provider<EurekaClient> eurekaClientProvider) {26 if (this.propertiesFactory.isSet(ServerList.class, serviceId)) {27 return this.propertiesFactory.get(ServerList.class, config, serviceId);28 }29 DiscoveryEnabledNIWSServerList discoveryServerList = new DiscoveryEnabledNIWSServerList(config, eurekaClientProvider);30 DomainExtractingServerList serverList = new DomainExtractingServerList(discoveryServerList, config, this.approximateZoneFromHostname);31 return serverList;32 }33 }4、从 DiscoveryClient 获取Server列表DynamicServerListLoadBalancer 中通过 ServerList 的 getUpdatedListOfServers 方法全量获取服务列表，在 eureka-client 环境下，ServerList 默认实现类为 DomainExtractingServerList，那就先看下它的 getUpdatedListOfServers 方法。 可以看出，DomainExtractingServerList 先用 DomainExtractingServerList 获取服务列表，然后根据 Ribbon 客户端配置重新构造 Server 对象返回。获取服务列表的核心在 DiscoveryEnabledNIWSServerList 中。 1 @Override 2 public List<DiscoveryEnabledServer> getUpdatedListOfServers() { 3 // list => DiscoveryEnabledNIWSServerList 4 List<DiscoveryEnabledServer> servers = setZones(this.list.getUpdatedListOfServers()); 5 return servers; 6 } 7 8 private List<DiscoveryEnabledServer> setZones(List<DiscoveryEnabledServer> servers) { 9 List<DiscoveryEnabledServer> result = new ArrayList<>();10 boolean isSecure = this.ribbon.isSecure(true);11 boolean shouldUseIpAddr = this.ribbon.isUseIPAddrForServer();12 // 根据客户端配置重新构造 DomainExtractingServer 返回13 for (DiscoveryEnabledServer server : servers) {14 result.add(new DomainExtractingServer(server, isSecure, shouldUseIpAddr, this.approximateZoneFromHostname));15 }16 return result;17 }先看下 DiscoveryEnabledNIWSServerList 的构造初始化：
1 public DiscoveryEnabledNIWSServerList(IClientConfig clientConfig, Provider<EurekaClient> eurekaClientProvider) { 2 this.eurekaClientProvider = eurekaClientProvider; 3 initWithNiwsConfig(clientConfig); 4 } 5 6 @Override 7 public void initWithNiwsConfig(IClientConfig clientConfig) { 8 // 客户端名称，就是服务名称 9 clientName = clientConfig.getClientName();10 // vipAddresses 得到的也是客户端名称11 vipAddresses = clientConfig.resolveDeploymentContextbasedVipAddresses();12 13 // 其它的一些配置....14 }接着看获取实例的 getUpdatedListOfServers，可以看到它的核心逻辑就是根据服务名从 EurekaClient 获取 InstanceInfo 实例列表，然后封装 Server 信息返回。 1 public List<DiscoveryEnabledServer> getUpdatedListOfServers(){ 2 return obtainServersViaDiscovery(); 3 } 4 5 private List<DiscoveryEnabledServer> obtainServersViaDiscovery() { 6 List<DiscoveryEnabledServer> serverList = new ArrayList<DiscoveryEnabledServer>(); 7 if (eurekaClientProvider == null || eurekaClientProvider.get() == null) { 8 return new ArrayList<DiscoveryEnabledServer>(); 9 }10 // 得到 EurekaClient，实际类型是 CloudEurekaClient，其父类是 DiscoveryClient11 EurekaClient eurekaClient = eurekaClientProvider.get();12 if (vipAddresses!=null){13 // 分割 vipAddresses，默认就是服务名称14 for (String vipAddress : vipAddresses.split(",")) {15 // 根据服务名称从 EurekaClient 获取实例信息16 List<InstanceInfo> listOfInstanceInfo = eurekaClient.getInstancesByVipAddress(vipAddress, isSecure, targetRegion);17 for (InstanceInfo ii : listOfInstanceInfo) {18 if (ii.getStatus().equals(InstanceStatus.UP)) {19 // ...20 // 根据实例信息 InstanceInfo 创建 Server21 DiscoveryEnabledServer des = createServer(ii, isSecure, shouldUseIpAddr);22 serverList.add(des);23 }24 }25 if (serverList.size()>0 && prioritizeVipAddressBasedServers){26 break; // if the current vipAddress has servers, we dont use subsequent vipAddress based servers27 }28 }29 }30 return serverList;31 }注意这里的 vipAddress 其实就是服务名：
最后看 EurekaClient 的 getInstancesByVipAddress，到这里就很清楚了，其实就是从 DiscoveryClient 的本地应用 Applications 中根据服务名取出所有的实例列表。 这里就和 Eureka 源码那块衔接上了，eureka-client 全量抓取注册表以及每隔30秒增量抓取注册表，都是合并到本地的 Applications 中。Ribbon 与 Eureka 结合后，Ribbon 获取 Server 就从 DiscoveryClient 的 Applications 中获取 Server 列表了。 1 public List<InstanceInfo> getInstancesByVipAddress(String vipAddress, boolean secure, String region) { 2 // ... 3 Applications applications; 4 if (instanceRegionChecker.isLocalRegion(region)) { 5 // 取本地应用 Applications 6 applications = this.localRegionApps.get(); 7 } else { 8 applications = remoteRegionVsApps.get(region); 9 if (null == applications) {10 return Collections.emptyList();11 }12 }13 14 if (!secure) {15 // 返回服务名对应的实例16 return applications.getInstancesByVirtualHostName(vipAddress);17 } else {18 return applications.getInstancesBySecureVirtualHostName(vipAddress);19 }20 }5、定时更新Server列表DynamicServerListLoadBalancer 初始化时，有个方法还没说，就是 enableAndInitLearnNewServersFeature()。这个方法只是调用 ServerListUpdater 启动了一个 UpdateAction，这个 UpdateAction 又只是调用了一下 updateListOfServers 方法，就是前面讲解过的全量更新 Server 的逻辑。 1 public void enableAndInitLearnNewServersFeature() { 2 serverListUpdater.start(updateAction); 3 } 4 5 protected final ServerListUpdater.UpdateAction updateAction = new ServerListUpdater.UpdateAction() { 6 @Override 7 public void doUpdate() { 8 // 调用 updateListOfServers 9 updateListOfServers();10 }11 };ServerListUpdater 的默认实现类是 PollingServerListUpdater，看下它的 start 方法： 其实就是以固定的频率，每隔30秒调用一下 updateListOfServers 方法，将 DiscoveryClient 中 Applications 中缓存的实例同步到 ILoadBalancer 中的 allServerList 列表中。 1 public synchronized void start(final UpdateAction updateAction) { 2 if (isActive.compareAndSet(false, true)) { 3 final Runnable wrapperRunnable = new Runnable() { 4 @Override 5 public void run() { 6 // ... 7 try { 8 // 执行一次 updateListOfServers 9 updateAction.doUpdate();10 // 设置最后更新时间11 lastUpdated = System.currentTimeMillis();12 } catch (Exception e) {13 logger.warn("Failed one update cycle", e);14 }15 }16 };17 18 // 固定频率调度19 scheduledFuture = getRefreshExecutor().scheduleWithFixedDelay(20 wrapperRunnable,21 initialDelayMs, // 默认 100022 refreshIntervalMs, // 默认 30 * 100023 TimeUnit.MILLISECONDS24 );25 } else {26 logger.info("Already active, no-op");27 }28 }6、判断Server是否存活在创建 ILoadBalancer 时，IPing 还没有看过是如何工作的。在初始化的时候，可以看到，主要就是设置了当前的 ping，然后重新设置了一个调度任务，默认每隔30秒调度一次 PingTask 任务。 1 public void setPing(IPing ping) { 2 if (ping != null) { 3 if (!ping.equals(this.ping)) { 4 this.ping = ping; 5 // 设置 Ping 任务 6 setupPingTask(); 7 } 8 } else { 9 this.ping = null;10 // cancel the timer task11 lbTimer.cancel();12 }13 }14 15 void setupPingTask() {16 // ...17 // 创建一个定时调度器18 lbTimer = new ShutdownEnabledTimer("NFLoadBalancer-PingTimer-" + name, true);19 // pingIntervalTime 默认为 30 秒，每隔30秒调度一次 PingTask20 lbTimer.schedule(new PingTask(), 0, pingIntervalSeconds * 1000);21 // 立即发起以 Ping22 forceQuickPing();23 }ShutdownEnabledTimer 可以简单了解下，它是继承自 Timer 的，它在创建的时候向 Runtime 注册了一个回调，在 jvm 关闭的时候来取消 Timer 的执行，进而释放资源。  1 public class ShutdownEnabledTimer extends Timer { 2 private static final Logger LOGGER = LoggerFactory.getLogger(ShutdownEnabledTimer.class); 3 4 private Thread cancelThread; 5 private String name; 6 7 public ShutdownEnabledTimer(String name, boolean daemon) { 8 super(name, daemon); 9 this.name = name;10 // 取消定时器的线程11 this.cancelThread = new Thread(new Runnable() {12 public void run() {13 ShutdownEnabledTimer.super.cancel();14 }15 });16 17 LOGGER.info("Shutdown hook installed for: {}", this.name);18 // 向 Runtime 注册一个钩子，在 jvm 关闭时，调用 cancelThread 取消定时任务19 Runtime.getRuntime().addShutdownHook(this.cancelThread);20 }21 22 @Override23 public void cancel() {24 super.cancel();25 LOGGER.info("Shutdown hook removed for: {}", this.name);26 try {27 Runtime.getRuntime().removeShutdownHook(this.cancelThread);28 } catch (IllegalStateException ise) {29 LOGGER.info("Exception caught (might be ok if at shutdown)", ise);30 }31 32 }33 }再来看下 PingTask，PingTask 核心逻辑就是遍历 allServers 列表，使用 IPingStrategy 和 IPing 来判断 Server 是否存活，并更新 Server 的状态，以及将所有存活的 Server 更新到 upServerList 中，upServerList 缓存了所有存活的 Server。 1 class PingTask extends TimerTask { 2 public void run() { 3 try { 4 // pingStrategy => SerialPingStrategy 5 new Pinger(pingStrategy).runPinger(); 6 } catch (Exception e) { 7 logger.error("LoadBalancer [{}]: Error pinging", name, e); 8 } 9 }10 }11 12 class Pinger {13 private final IPingStrategy pingerStrategy;14 15 public Pinger(IPingStrategy pingerStrategy) {16 this.pingerStrategy = pingerStrategy;17 }18 19 public void runPinger() throws Exception {20 if (!pingInProgress.compareAndSet(false, true)) {
21 return; // Ping in progress - nothing to do22 }23 24 Server[] allServers = null;25 boolean[] results = null;26 27 Lock allLock = null;28 Lock upLock = null;29 30 try {31 allLock = allServerLock.readLock();32 allLock.lock();33 // 加读锁，取出 allServerList 中的 Server34 allServers = allServerList.toArray(new Server[allServerList.size()]);35 allLock.unlock();36 37 int numCandidates = allServers.length;38 // 使用 IPingStrategy 和 IPing 对所有 Server 发起 ping 请求39 results = pingerStrategy.pingServers(ping, allServers);40 41 final List<Server> newUpList = new ArrayList<Server>();42 final List<Server> changedServers = new ArrayList<Server>();43 44 for (int i = 0; i < numCandidates; i++) {45 boolean isAlive = results[i];46 Server svr = allServers[i];47 boolean oldIsAlive = svr.isAlive();48 // 设置 alive 是否存活49 svr.setAlive(isAlive);50 51 // 实例变更52 if (oldIsAlive != isAlive) {53 changedServers.add(svr);54 logger.debug("LoadBalancer [{}]: Server [{}] status changed to {}", name, svr.getId(), (isAlive ? "ALIVE" : "DEAD"));55 }56 57 // 添加存活的 Server58 if (isAlive) {59 newUpList.add(svr);60 }61 }62 upLock = upServerLock.writeLock();63 upLock.lock();64 // 更新 upServerList，upServerList 只保存了存活的 Server65 upServerList = newUpList;66 upLock.unlock();67 // 通知变更68 notifyServerStatusChangeListener(changedServers);69 } finally {70 pingInProgress.set(false);71 }72 }73 }IPingStrategy 的默认实现类是 SerialPingStrategy，进入可以发现它只是遍历所有 Server，然后用 IPing 判断 Server 是否存活。 1 private static class SerialPingStrategy implements IPingStrategy { 2 @Override 3 public boolean[] pingServers(IPing ping, Server[] servers) { 4 int numCandidates = servers.length; 5 boolean[] results = new boolean[numCandidates]; 6 7 for (int i = 0; i < numCandidates; i++) { 8 results[i] = false; 9 try {10 if (ping != null) {11 // 使用 IPing 判断 Server 是否存活12 results[i] = ping.isAlive(servers[i]);13 }14 } catch (Exception e) {15 logger.error("Exception while pinging Server: '{}'", servers[i], e);16 }17 }18 return results;19 }20 }在集成 eureka-client 后，IPing默认实现类是 NIWSDiscoveryPing，看它的 isAlive 方法，其实就是判断对应 Server 的实例 InstanceInfo 的状态是否是 UP 状态，UP状态就表示 Server 存活。 1 public boolean isAlive(Server server) { 2 boolean isAlive = true; 3 if (server!=null && server instanceof DiscoveryEnabledServer){ 4 DiscoveryEnabledServer dServer = (DiscoveryEnabledServer)server;
5 InstanceInfo instanceInfo = dServer.getInstanceInfo(); 6 if (instanceInfo!=null){
7 InstanceStatus status = instanceInfo.getStatus(); 8 if (status!=null){ 9 // 判断Server对应的实例状态是否是 UP10 isAlive = status.equals(InstanceStatus.UP);11 }12 }13 }14 return isAlive;15 }7、一张图总结 Ribbon 核心原理① Ribbon 核心工作原理总结
② 下面用一张图来总结下 Ribbon 这块获取Server的核心流程以及对应的核心接口间的关系。
8、Ribbon 脱离 Eureka 使用在默认情况下，Ribbon 客户端会从 EurekaClient 获取服务列表，其实就是间接从注册中心读取服务注册信息列表，来达到动态负载均衡的功能。但如果不想从 EurekaClient 读取，可以禁用 Ribbon 的 Eureka 功能。 1 ribbon:2 eureka:3 enabled: false 那 ribbon.eureka.enabled 是如何控制禁用 Eureka 的呢？看 RibbonEurekaAutoConfiguration： 1 @Configuration(proxyBeanMethods = false)2 @EnableConfigurationProperties3 @ConditionalOnRibbonAndEurekaEnabled4 @AutoConfigureAfter(RibbonAutoConfiguration.class)5 @RibbonClients(defaultConfiguration = EurekaRibbonClientConfiguration.class)6 public class RibbonEurekaAutoConfiguration {7 8 }这个配置类通过 @RibbonClients 指定了默认的客户端配置类为 EurekaRibbonClientConfiguration，但生效的前提是 @ConditionalOnRibbonAndEurekaEnabled，进去可以看到这个条件注解就是判断 Ribbon Eureka 是否启用了，就可以设置 ribbon.eureka.enabled=false 来禁用 RIbbon Eureka。 1 @Target({ ElementType.TYPE, ElementType.METHOD }) 2 @Retention(RetentionPolicy.RUNTIME) 3 @Documented 4 // 条件类 OnRibbonAndEurekaEnabledCondition 5 @Conditional(ConditionalOnRibbonAndEurekaEnabled.OnRibbonAndEurekaEnabledCondition.class) 6 public @interface ConditionalOnRibbonAndEurekaEnabled { 7 8 class OnRibbonAndEurekaEnabledCondition extends AllNestedConditions { 9 10 OnRibbonAndEurekaEnabledCondition() {11 super(ConfigurationPhase.REGISTER_BEAN);12 }13 14 // 引入了类：DiscoveryEnabledNIWSServerList 15 @ConditionalOnClass(DiscoveryEnabledNIWSServerList.class)16 // 存在bean对象：SpringClientFactory17 @ConditionalOnBean(SpringClientFactory.class)18 // ribbon.eureka.enabled=true19 @ConditionalOnProperty(value = "ribbon.eureka.enabled", matchIfMissing = true)20 static class Defaults {21 22 }23 24 // 存在bean对象：EurekaClient25 @ConditionalOnBean(EurekaClient.class)26 static class EurekaBeans {27 28 }29 30 // eureka.client.enabled=true31 @ConditionalOnProperty(value = "eureka.client.enabled", matchIfMissing = true)32 @ConditionalOnDiscoveryEnabled33 static class OnEurekaClientEnabled {34 35 }36 }37 }如果想从其它地方获取服务列表，可以自定义接口实现 ServerList<Server> 来获取，也可以在配置文件中设置地址列表： 1 <client-name>:2 ribbon:3 listOfServers: http://10.215.0.92:8010,http://10.215.0.92:8011 |
|
|
