SpringCloud 源码系列（4）—— 负载均衡 Ribbon（上）

SpringCloud 源码系列（4）—— 负载均衡 Ribbon（上）

SpringCloud 源码系列（5）—— 负载均衡 Ribbon（下）

一、负载均衡

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 具备负载均衡的能力的。

@SpringBootApplication
public class ConsumerApplication {

    @Bean
    @LoadBalanced
    public RestTemplate restTemplate() {
        return new RestTemplate();
    }

    public static void main(String[] args) {
        SpringApplication.run(ConsumerApplication.class, args);
    }
}

@RestController
public class DemoController {
    private final Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private RestTemplate restTemplate;

    @GetMapping("/v1/id")
    public ResponseEntity<String> getId() {
        ResponseEntity<String> result = restTemplate.getForEntity("http://demo-producer/v1/uuid", String.class);
        String uuid = result.getBody();
        logger.info("request id: {}", uuid);
        return ResponseEntity.ok(uuid);
    }
}

2、Ribbon 与负载均衡

① 负载均衡

负载均衡是指将负载分摊到多个执行单元上，负载均衡主要可以分为集中式负载均衡与进程内负载均衡：

• 集中式负载均衡指位于因特网与执行单元之间，并负责把网络请求转发到各个执行单元上，比如 Nginx、F5。集中式负载均衡也可以称为服务端负载均衡。
• 进程内负载均衡是将负载均衡逻辑集成到客户端上，客户端维护了一份服务提供者的实例列表，实例列表一般会从注册中心比如 Eureka 中获取。有了实例列表，就可以通过负载均衡策略将请求分摊给多个服务提供者，从而达到负载均衡的目的。进程内负载均衡一般也称为客户端负载均衡。

Ribbon 是一个客户端负载均衡器，可以很好地控制 HTTP 和 TCP 客户端的负载均衡行为。Ribbon 是 Netflix 公司开源的一个负载均衡组件，已经整合到 SpringCloud 生态中，它在 Spring Cloud 生态内是一个不可缺少的组件，少了它，服务便不能横向扩展。

② Ribbon 模块

Ribbon 有很多子模块，官方文档中说明，目前 Netflix 公司主要用于生产环境的 Ribbon 子模块如下：

• ribbon-loadbalancer：可以独立使用或与其他模块一起使用的负载均衡器 API。
• ribbon-eureka：Ribbon 结合 Eureka 客户端的 API，为负载均衡器提供动态服务注册列表信息。
• ribbon-core：Ribbon 的核心API。

③ 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 这个注解的定义，可以得到如下信息：

• 这个注解使用 @Qualifier 标记，其它地方就可以注入 LoadBalanced 注解的 bean 对象。
• 从注释中可以了解到，@LoadBalanced 标记的 RestTemplate 或 WebClient 将使用 LoadBalancerClient 来配置 bean 对象。

/**
 * Annotation to mark a RestTemplate or WebClient 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 {

}

注意 @LoadBalanced 是 spring-cloud-commons 模块下 loadbalancer 包下的。

2、RestTemplate  负载均衡自动化配置

在 @LoadBalanced 同包下，有一个 LoadBalancerAutoConfiguration 自动化配置类，从注释也可以看出，这是客户端负载均衡 Ribbon 的自动化配置类。

从这个自动化配置类可以得到如下信息：

• 首先要有 RestTemplate 的依赖和定义了 LoadBalancerClient 对象的前提下才会触发这个自动化配置类，这也对应了前面，RestTemplate 要用 LoadBalancerClient  来配置。
• 接着可以看到这个类注入了带有 @LoadBalanced 标识的 RestTemplate 对象，就是要对这部分对象增加负载均衡的能力。
• 从 SmartInitializingSingleton 的构造中可以看到，就是在 bean 初始化完成后，用 RestTemplateCustomizer 定制化 RestTemplate。
• 再往下可以看到，RestTemplateCustomizer 其实就是向 RestTemplate 中添加了 LoadBalancerInterceptor 这个拦截器。
• 而 LoadBalancerInterceptor 的构建又需要 LoadBalancerClient 和 LoadBalancerRequestFactory，LoadBalancerRequestFactory 则通过 LoadBalancerClient 和 LoadBalancerRequestTransformer 构造完成。

/**
 * Auto-configuration for Ribbon (client-side load balancing).
 */
@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(RestTemplate.class) // 有 RestTemplate 的依赖
@ConditionalOnBean(LoadBalancerClient.class) // 定义了 LoadBalancerClient 的 bean 对象
@EnableConfigurationProperties(LoadBalancerRetryProperties.class)
public class LoadBalancerAutoConfiguration {

    // 注入 @LoadBalanced 标记的 RestTemplate 对象
    @LoadBalanced
    @Autowired(required = false)
    private List<RestTemplate> restTemplates = Collections.emptyList();

    @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) {
                    // 利用 RestTemplateCustomizer 定制化 restTemplate
                    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 {

        // 创建 LoadBalancerInterceptor 需要 LoadBalancerClient 和 LoadBalancerRequestFactory
        @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());
                // 向 restTemplate 添加 LoadBalancerInterceptor 拦截器
                list.add(loadBalancerInterceptor);
                restTemplate.setInterceptors(list);
            };
        }

    }
}

3、RestTemplate 拦截器 LoadBalancerInterceptor

LoadBalancerAutoConfiguration 自动化配置主要就是给 RestTemplate 添加了一个负载均衡拦截器 LoadBalancerInterceptor。从 setInterceptors 的参数可以看出，拦截器的类型是 ClientHttpRequestInterceptor，如果我们想定制化 RestTemplate，就可以实现这个接口来定制化，然后还可以用 @Order 标记拦截器的先后顺序。

public void setInterceptors(List<ClientHttpRequestInterceptor> interceptors) {
    if (this.interceptors != interceptors) {
        this.interceptors.clear();
        this.interceptors.addAll(interceptors);
        // 根据 @Order 注解的顺序排序
        AnnotationAwareOrderComparator.sort(this.interceptors);
    }
}

interceptors 拦截器是在 RestTemplate 的父类 InterceptingHttpAccessor 中的， RestTemplate 的类结构如下图所示。

从 restTemplate.getForEntity("http://demo-producer/v1/uuid", String.class) 这个GET请求进去看看，是如何使用 LoadBalancerInterceptor 的。一步步进去，可以看到最终是进入到 doExecute 这个方法了。

在 doExecute 方法中，首先根据 url、method 创建一个 ClientHttpRequest，然后利用 ClientHttpRequest 来发起请求。

protected <T> T doExecute(URI url, @Nullable HttpMethod method, @Nullable RequestCallback requestCallback,
        @Nullable ResponseExtractor<T> responseExtractor) throws RestClientException {
    ClientHttpResponse response = null;
    try {
        // 创建一个 ClientHttpRequest
        ClientHttpRequest request = createRequest(url, method);
        if (requestCallback != null) {
            requestCallback.doWithRequest(request);
        }
        // 调用 ClientHttpRequest 的 execute() 方法
        response = request.execute();
        // 处理返回结果
        handleResponse(url, method, response);
        return (responseExtractor != null ? responseExtractor.extractData(response) : null);
    }
    catch (IOException ex) {
        // ...
    }
    finally {
        if (response != null) {
            response.close();
        }
    }
}

//////////////////////////////////////

protected ClientHttpRequest createRequest(URI url, HttpMethod method) throws IOException {
    ClientHttpRequest request = getRequestFactory().createRequest(url, method);
    initialize(request);
    if (logger.isDebugEnabled()) {
        logger.debug("HTTP " + method.name() + " " + url);
    }
    return request;
}

InterceptingHttpAccessor 中重写了父类 HttpAccessor 的 getRequestFactory 方法，父类默认的 requestFactory 是 SimpleClientHttpRequestFactory。

重写后的 getRequestFactory 方法中，如果拦截器不为空，则基于父类默认的 SimpleClientHttpRequestFactory 和拦截器创建了 InterceptingClientHttpRequestFactory。

public ClientHttpRequestFactory getRequestFactory() {
    List<ClientHttpRequestInterceptor> interceptors = getInterceptors();
    if (!CollectionUtils.isEmpty(interceptors)) {
        ClientHttpRequestFactory factory = this.interceptingRequestFactory;
        if (factory == null) {
            // 传入 SimpleClientHttpRequestFactory 和 ClientHttpRequestInterceptor 拦截器
            factory = new InterceptingClientHttpRequestFactory(super.getRequestFactory(), interceptors);
            this.interceptingRequestFactory = factory;
        }
        return factory;
    }
    else {
        return super.getRequestFactory();
    }
}

也就是说调用了 InterceptingClientHttpRequestFactory 的 createRequest 方法来创建 ClientHttpRequest。进去可以看到，ClientHttpRequest 的实际类型就是 InterceptingClientHttpRequest。

protected ClientHttpRequest createRequest(URI uri, HttpMethod httpMethod, ClientHttpRequestFactory requestFactory) {
    return new InterceptingClientHttpRequest(requestFactory, this.interceptors, uri, httpMethod);
}

InterceptingClientHttpRequest 的类结构如下：

RestTemplate 的 doExecute 中调用 request.execute() 其实是调用了 InterceptingClientHttpRequest 父类 AbstractClientHttpRequest 中的 execute 方法。一步步进去可以发现最终其实是调用了 InterceptingClientHttpRequest 的 executeInternal 方法。

在 InterceptingClientHttpRequest  的 executeInternal 方法中，创建了 InterceptingRequestExecution 来执行请求。在 InterceptingRequestExecution 的 execute 方法中，会先遍历执行所有拦截器，然后通过 ClientHttpRequest 发起真正的 http 请求。

protected final ClientHttpResponse executeInternal(HttpHeaders headers, byte[] bufferedOutput) throws IOException {
    // 创建 InterceptingRequestExecution
    InterceptingRequestExecution requestExecution = new InterceptingRequestExecution();
    // 请求调用
    return requestExecution.execute(this, bufferedOutput);
}

private class InterceptingRequestExecution implements ClientHttpRequestExecution {

    private final Iterator<ClientHttpRequestInterceptor> iterator;

    public InterceptingRequestExecution() {
        // 拦截器迭代器
        this.iterator = interceptors.iterator();
    }

    @Override
    public ClientHttpResponse execute(HttpRequest request, byte[] body) throws IOException {
        if (this.iterator.hasNext()) {
            ClientHttpRequestInterceptor nextInterceptor = this.iterator.next();
            // 利用拦截器拦截处理，并传入 InterceptingRequestExecution
            return nextInterceptor.intercept(request, body, this);
        }
        else {
            // 拦截器遍历完后开始发起真正的 http 请求
            HttpMethod method = request.getMethod();
            Assert.state(method != null, "No standard HTTP method");
            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();
        }
    }
}

进入到 LoadBalancerInterceptor 的 intercept 拦截方法内，可以看到从请求的原始地址中获取了服务名称，然后调用了 loadBalancer 的 execute 方法，也就是 LoadBalancerClient。

到这里，其实已经可以想象，loadBalancer.execute 这行代码就是根据服务名称去获取一个具体的实例，然后将原始地址替换为实例的IP地址。那这个 loadBalancer 又是什么呢？

public ClientHttpResponse intercept(final HttpRequest request, final byte[] body, final ClientHttpRequestExecution execution) throws IOException {
    // 原始地址：http://demo-producer/v1/uuid
    final URI originalUri = request.getURI();
    // host 就是服务名：demo-producer
    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));
}

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 注册中心获取实例。

@Configuration
@Conditional(RibbonAutoConfiguration.RibbonClassesConditions.class)
@RibbonClients
// 后于 EurekaClientAutoConfiguration 配置
@AutoConfigureAfter(name = "org.springframework.cloud.netflix.eureka.EurekaClientAutoConfiguration")
// 先于 LoadBalancerAutoConfiguration 配置
@AutoConfigureBefore({ LoadBalancerAutoConfiguration.class,    AsyncLoadBalancerAutoConfiguration.class })
@EnableConfigurationProperties({ RibbonEagerLoadProperties.class, ServerIntrospectorProperties.class })
public class RibbonAutoConfiguration {

    @Autowired(required = false)
    private List<RibbonClientSpecification> configurations = new ArrayList<>();

    @Bean
    @ConditionalOnMissingBean
    public SpringClientFactory springClientFactory() {
        SpringClientFactory factory = new SpringClientFactory();
        factory.setConfigurations(this.configurations);
        return factory;
    }

    @Bean
    @ConditionalOnMissingBean(LoadBalancerClient.class)
    public LoadBalancerClient loadBalancerClient() {
        return new RibbonLoadBalancerClient(springClientFactory());
    }
}

LoadBalancerClient 主要提供了三个接口：

public interface LoadBalancerClient extends ServiceInstanceChooser {

    // 从 LoadBalancer 找一个 Server 来发送请求
    <T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException;

    // 从传入的 ServiceInstance 取 Server 来发送请求  
    <T> T execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest<T> request) throws IOException;

    // 对原始 URI 重构
    URI reconstructURI(ServiceInstance instance, URI original);
}

进入到 RibbonLoadBalancerClient 的 execute 方法中可以看到：

• 首先根据服务名获取服务对应的负载均衡器 ILoadBalancer。
• 然后从 ILoadBalancer 中根据一定策略选出一个实例 Server。
• 然后将 server、serviceId 等信息封装到 RibbonServer 中，也就是一个服务实例 ServiceInstance。
• 最后调用了 LoadBalancerRequest 的 apply，并传入 ServiceInstance，将地址中的服务名替换为真实的IP地址。

public <T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException {
    return execute(serviceId, request, null);
}

public <T> T execute(String serviceId, LoadBalancerRequest<T> request, Object hint)
        throws IOException {
    // 根据服务名获取一个负载均衡器 ILoadBalancer
    ILoadBalancer loadBalancer = getLoadBalancer(serviceId);
    // 利用负载均衡器获取实例 Server
    Server server = getServer(loadBalancer, hint);
    if (server == null) {
        throw new IllegalStateException("No instances available for " + serviceId);
    }
    // 封装实例信息：RibbonServer 的父类是 ServiceInstance
    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);
    }

    try {
        // 处理地址，将服务名替换为真实的IP地址
        T returnVal = request.apply(serviceInstance);
        return returnVal;
    } catch (Exception ex) {
        // ...
    }
    return null;
}

这个 LoadBalancerRequest 其实就是 LoadBalancerInterceptor 的 intercept 中创建的一个匿名类，在它的函数式接口内，主要是用装饰器 ServiceRequestWrapper 将 request 包了一层。

public LoadBalancerRequest<ClientHttpResponse> createRequest(final HttpRequest request, final byte[] body, final ClientHttpRequestExecution execution) {
    return instance -> {
        // 封装 HttpRequest，ServiceRequestWrapper 重载了 getURI 方法。
        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);
    };
}

ServiceRequestWrapper 主要就是重写了 getURI 方法，在重写的 getURI 方法内，它用 loadBalancer 对 URI 进行了重构，进去可以发现，就是将原始地址中的服务名替换为 Server 的真实IP、端口地址。

@Override
public URI getURI() {
    // 重构 URI
    URI uri = this.loadBalancer.reconstructURI(this.instance, getRequest().getURI());
    return uri;
}

public URI reconstructURI(ServiceInstance instance, URI original) {
    Assert.notNull(instance, "instance can not be null");
    // 服务名
    String serviceId = instance.getServiceId();
    RibbonLoadBalancerContext context = this.clientFactory.getLoadBalancerContext(serviceId);

    URI uri;
    Server server;
    if (instance instanceof RibbonServer) {
        RibbonServer ribbonServer = (RibbonServer) instance;
        server = ribbonServer.getServer();
        uri = updateToSecureConnectionIfNeeded(original, ribbonServer);
    }
    else {
        server = new Server(instance.getScheme(), instance.getHost(), instance.getPort());
        IClientConfig clientConfig = clientFactory.getClientConfig(serviceId);
        ServerIntrospector serverIntrospector = serverIntrospector(serviceId);
        uri = updateToSecureConnectionIfNeeded(original, clientConfig, serverIntrospector, server);
    }
    // 重构地址
    return context.reconstructURIWithServer(server, uri);
}

reconstructURIWithServer：

public URI reconstructURIWithServer(Server server, URI original) {
    String host = server.getHost();
    int port = server.getPort();
    String scheme = server.getScheme();
    
    if (host.equals(original.getHost()) 
            && port == original.getPort()
            && scheme == original.getScheme()) {
        return original;
    }
    if (scheme == null) {
        scheme = original.getScheme();
    }
    if (scheme == null) {
        scheme = deriveSchemeAndPortFromPartialUri(original).first();
    }

    try {
        StringBuilder sb = new StringBuilder();
        sb.append(scheme).append("://");
        if (!Strings.isNullOrEmpty(original.getRawUserInfo())) {
            sb.append(original.getRawUserInfo()).append("@");
        }
        sb.append(host);
        if (port >= 0) {
            sb.append(":").append(port);
        }
        sb.append(original.getRawPath());
        if (!Strings.isNullOrEmpty(original.getRawQuery())) {
            sb.append("?").append(original.getRawQuery());
        }
        if (!Strings.isNullOrEmpty(original.getRawFragment())) {
            sb.append("#").append(original.getRawFragment());
        }
        URI newURI = new URI(sb.toString());
        return newURI;            
    } catch (URISyntaxException e) {
        throw new RuntimeException(e);
    }
}

5、RestTemplate 负载均衡总结

到这里，我们基本就弄清楚了一个简单的 @LoadBalanced 注解如何让 RestTemplate 具备了负载均衡的能力了，这一节来做个小结。

① RestTemplate 如何获得负载均衡的能力

• 1）首先 RestTemplate 是 spring-web 模块下一个访问第三方 RESTful API 接口的网络请求框架
• 2）在 spring cloud 微服务架构中，用 @LoadBalanced 对 RestTemplate 做个标记，就可以使 RestTemplate 具备负载均衡的能力
• 3）使 RestTemplate 具备负载均衡的核心组件就是 LoadBalancerAutoConfiguration 配置类中向其添加的 LoadBalancerInterceptor 负载均衡拦截器
• 4）RestTemplate 在发起 http 调用前，会遍历所有拦截器来对 RestTemplate 定制化，LoadBalancerInterceptor 就是在这时将URI中的服务名替换为实例的真实IP地址。定制完成后，就会发起真正的 http 请求。
• 5）LoadBalancerInterceptor 又主要是使用负载均衡客户端 LoadBalancerClient 来完成URI的重构的，LoadBalancerClient 就可以根据服务名查找一个可用的实例，然后重构URI。

② 核心组件

这里会涉及多个模块，下面是核心组件的所属模块：

spring-web：

• RestTemplate
• InterceptingClientHttpRequest：执行拦截器，并发起最终http调用

spring-cloud-commons：

• @LoadBalanced
• LoadBalancerAutoConfiguration
• LoadBalancerRequestFactory：创建装饰类 ServiceRequestWrapper 替换原来的 HttpRequest，重载 getURI 方法。
• LoadBalancerInterceptor：负载均衡拦截器
• LoadBalancerClient：负载均衡客户端接口

spring-cloud-netflix-ribbon：

• RibbonLoadBalancerClient：LoadBalancerClient 的实现类，Ribbon 的负载均衡客户端
• RibbonAutoConfiguration

ribbon-loadbalancer：

• ILoadBalancer：负载均衡器
• Server：实例

③ 最后再用一张图把 RestTemplate 这块的关系捋一下

三、ILoadBalancer 获取 Server

从前面 RestTemplate 那张图可以看出，使 RestTemplate 具备负载均衡的能力，最重要的一个组件之一就是 ILoadBalancer，因为要用它来获取能调用的 Server，有了 Server 才能对原始带有服务名的 URI 进行重构。这节就来看下 Ribbon 的负载均衡器 ILoadBalancer 是如何创建的以及如何通过它获取 Server。

1、创建负载均衡器 ILoadBalancer

① SpringClientFactory与上下文

ILoadBalancer 是用 SpringClientFactory 的 getLoadBalancer 方法根据服务名获取的，从 getInstance 一步步进去可以发现，每个服务都会创建一个 AnnotationConfigApplicationContext，也就是一个应用上下文 ApplicationContext。相当于就是一个服务绑定一个 ILoadBalancer。

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

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) {
    // contexts => Map<String, AnnotationConfigApplicationContext>
    if (!this.contexts.containsKey(name)) {
        synchronized (this.contexts) {
            if (!this.contexts.containsKey(name)) {
                this.contexts.put(name, createContext(name));
            }
        }
    }
    return this.contexts.get(name);
}

调试看下 AnnotationConfigApplicationContext 上下文，可以看到放入了与这个服务绑定的 ILoadBalancer、IClientConfig、RibbonLoadBalancerContext 等。

它这里为什么要每个服务都绑定一个 ApplicationContext 呢？我猜想应该是因为服务实例列表可以有多个来源，比如可以从 eureka 注册中心获取、可以通过代码配置、可以通过配置文件配置，另外每个服务还可以有很多个性化的配置，有默认的配置、定制的全局配置、个别服务的特定配置等，它这样做就便于用户定制每个服务的负载均衡策略。

② Ribbon的饥饿加载

而且这个Ribbon客户端的应用上下文默认是懒加载的，并不是在启动的时候就加载上下文，而是在第一次调用的时候才会去初始化。

如果想服务启动时就初始化，可以指定Ribbon客户端的具体名称，在启动的时候就加载配置项的上下文：

ribbon:
  eager-load:
    enabled: true
    clients: demo-producer,demo-xxx

在 RibbonAutoConfiguration 配置类中可以找到这个饥饿配置，如果开启了饥饿加载，就会创建 RibbonApplicationContextInitializer 来在启动时初始化上下文。

@Bean
@ConditionalOnProperty("ribbon.eager-load.enabled")
public RibbonApplicationContextInitializer ribbonApplicationContextInitializer() {
    return new RibbonApplicationContextInitializer(springClientFactory(), ribbonEagerLoadProperties.getClients());
}

public class RibbonApplicationContextInitializer implements ApplicationListener<ApplicationReadyEvent> {
    private final SpringClientFactory springClientFactory;

    // List of Ribbon client names
    private final List<String> clientNames;

    public RibbonApplicationContextInitializer(SpringClientFactory springClientFactory, List<String> clientNames) {
        this.springClientFactory = springClientFactory;
        this.clientNames = clientNames;
    }

    protected void initialize() {
        if (clientNames != null) {
            for (String clientName : clientNames) {
                // 提前初始化上下文
                this.springClientFactory.getContext(clientName);
            }
        }
    }

    @Override
    public void onApplicationEvent(ApplicationReadyEvent event) {
        initialize();
    }
}

③ RibbonClientConfiguration

ILoadBalancer 的创建在哪呢？看 RibbonClientConfiguration，这个配置类提供了 ILoadBalancer 的默认创建方法，ILoadBalancer 的默认实现类为 ZoneAwareLoadBalancer。

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 ServerListUpdater ribbonServerListUpdater(IClientConfig config) {
        return new PollingServerListUpdater(config);
    }

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

可以看到创建 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 判断是否已配置了对应类型的实现类，没有才使用默认的实现类。

也就是说针对特定的服务，这几个类可以自行定制化，也可以通过配置指定其它的实现类。

① 全局策略配置

如果想要全局更改配置，需要加一个配置类，比如像下面这样：

@Configuration
public class GlobalRibbonConfiguration {

    @Bean
    public IRule ribbonRule() {
        return new RandomRule();
    }

    @Bean
    public IPing ribbonPing() {
        return new NoOpPing();
    }
}

② 基于注解的配置

如果想针对某一个服务定制配置，可以通过 @RibbonClients 来配置特定服务的配置类。

需要先定义一个服务配置类：

@Configuration
public class ProducerRibbonConfiguration {

    @Bean
    public IRule ribbonRule() {
        return new RandomRule();
    }

    @Bean
    public IPing ribbonPing() {
        return new NoOpPing();
    }
}

用 @RibbonClients 注解为服务指定特定的配置类，并排除掉，不让 Spring 扫描，否则就变成了全局配置了。

@RibbonClients({
    @RibbonClient(name = "demo-producer", configuration = ProducerRibbonConfiguration.class)
})
@ComponentScan(excludeFilters = {
    @ComponentScan.Filter(type = FilterType.ASSIGNABLE_TYPE, classes = ProducerRibbonConfiguration.class)
})

③ 配置文件配置

通过配置文件的方式来配置，配置的格式就是 <服务名称>.ribbon.<属性>：

demo-producer:
  ribbon:
    # ILoadBalancer
    NFLoadBalancerClassName: com.netflix.loadbalancer.NoOpLoadBalancer
    # IRule
    NFLoadBalancerRuleClassName: com.netflix.loadbalancer.RandomRule
    # IPing
    NFLoadBalancerPingClassName:
    # ServerList<Server>
    NIWSServerListClassName:
    # ServerListFilter<Server>
    NIWSServerListFilterClassName:

④ 优先级顺序

这几种配置方式的优先级顺序是 配置文件配置 > @RibbonClients 配置 > 全局配置 > 默认配置。

3、ZoneAwareLoadBalancer 选择 Server

获取到 ILoadBalancer 后，就要去获取 Server 了，可以看到，就是用 ILoadBalancer 来获取 Server。

protected Server getServer(ILoadBalancer loadBalancer, Object hint) {
    if (loadBalancer == null) {
        return null;
    }
    // Use 'default' on a null hint, or just pass it on?
    return loadBalancer.chooseServer(hint != null ? hint : "default");
}

ILoadBalancer  的默认实现类是 ZoneAwareLoadBalancer，进入它的 chooseServer 方法内，如果只配置了一个 zone，就走父类的 chooseServer，否则从多个 zone 中去选择实例。

public Server chooseServer(Object key) {
    // ENABLED => ZoneAwareNIWSDiscoveryLoadBalancer.enabled 默认 true
    // AvailableZones 配置的只有一个 defaultZone
    if (!ENABLED.get() || getLoadBalancerStats().getAvailableZones().size() <= 1) {
        logger.debug("Zone aware logic disabled or there is only one zone");
        // 走父类获取 Server 的逻辑
        return super.chooseServer(key);
    }
    
    // 多 zone 逻辑....
}

先看下 ZoneAwareLoadBalancer 的类继承结构，ZoneAwareLoadBalancer 的直接父类是 DynamicServerListLoadBalancer，DynamicServerListLoadBalancer 的父类又是 BaseLoadBalancer。

ZoneAwareLoadBalancer 调用父类的 chooseServer 方法是在 BaseLoadBalancer 中的，进去可以看到，它主要是用 IRule 来选择实例，最终选择实例的策略就交给了 IRule 接口。

public Server chooseServer(Object key) {
    if (counter == null) {
        counter = createCounter();
    }
    counter.increment();
    if (rule == null) {
        return null;
    } else {
        try {
            // IRule
            return rule.choose(key);
        } catch (Exception e) {
            logger.warn("LoadBalancer [{}]:  Error choosing server for key {}", name, key, e);
            return null;
        }
    }
}

4、ZoneAvoidanceRule 断言筛选、轮询选择 Server

IRule 的默认实现类是 ZoneAvoidanceRule，先看下 ZoneAvoidanceRule 的继承结构，ZoneAvoidanceRule 的直接父类是 PredicateBasedRule。

rule.choose 的逻辑在 PredicateBasedRule 中，getPredicate() 返回的是 ZoneAvoidanceRule 创建的一个组合断言 CompositePredicate，就是用这个断言来过滤出可用的 Server，并通过轮询的策略返回一个 Server。

public Server choose(Object key) {
    ILoadBalancer lb = getLoadBalancer();
    // getPredicate() Server断言 => CompositePredicate
    // RoundRobin 轮询方式获取实例
    Optional<Server> server = getPredicate().chooseRoundRobinAfterFiltering(lb.getAllServers(), key);
    if (server.isPresent()) {
        return server.get();
    } else {
        return null;
    }       
}

在初始化 ZoneAvoidanceRule 配置时，创建了 CompositePredicate，可以看到这个组合断言主要有两个断言，一个是断言 Server 的 zone 是否可用，一个断言 Server 本身是否可用，例如 Server 无法 ping 通。

public void initWithNiwsConfig(IClientConfig clientConfig) {
    // 断言 Server 的 zone 是否可用，只有一个 defaultZone 的情况下都是可用的
    ZoneAvoidancePredicate zonePredicate = new ZoneAvoidancePredicate(this, clientConfig);
    // 断言 Server 是否可用
    AvailabilityPredicate availabilityPredicate = new AvailabilityPredicate(this, clientConfig);
    // 封装组合断言
    compositePredicate = createCompositePredicate(zonePredicate, availabilityPredicate);
}

private CompositePredicate createCompositePredicate(ZoneAvoidancePredicate p1, AvailabilityPredicate p2) {
    // 建造者模式创建断言
    return CompositePredicate.withPredicates(p1, p2)
                         .addFallbackPredicate(p2)
                         .addFallbackPredicate(AbstractServerPredicate.alwaysTrue())
                         .build();
    
}

接着看选择Server的 chooseRoundRobinAfterFiltering，参数 servers 是通过 ILoadBalancer 获取的所有实例，可以看到它其实就是返回了 ILoadBalancer 在内存中缓存的服务所有 Server。这个 Server 从哪来的我们后面再来看。

public List<Server> getAllServers() {
    // allServerList => List<Server>
    return Collections.unmodifiableList(allServerList);
}

先对所有实例通过断言过滤掉不可用的 Server，然后是通过轮询的方式获取一个 Server 返回。这就是默认配置下 ILoadBalancer（ZoneAwareLoadBalancer） 通过 IRule（ZoneAvoidanceRule） 选择 Server 的流程了。

public Optional<Server> chooseRoundRobinAfterFiltering(List<Server> servers, Object loadBalancerKey) {
    // 断言获取可用的 Server
    List<Server> eligible = getEligibleServers(servers, loadBalancerKey);
    if (eligible.size() == 0) {
        return Optional.absent();
    }
    // 通过取模的方式轮询 Server
    return Optional.of(eligible.get(incrementAndGetModulo(eligible.size())));
}

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();
        // 对每个 Server 断言
        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;
        // CAS 更新 nextIndex
        if (nextIndex.compareAndSet(current, next) && current < modulo)
            return current;
    }
}

四、Ribbon 整合 Eureka Client 拉取Server列表

前面在通过 IRule 选择 Server 的时候，首先通过 lb.getAllServers() 获取了所有的 Server，那这些 Server 从哪里来的呢，这节就来看下。

1、ILoadBalancer 初始化

ILoadBalancer 的默认实现类是 ZoneAwareLoadBalancer，先从 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);
}

可以看到，ZoneAwareLoadBalancer 直接调用了父类 DynamicServerListLoadBalancer 的构造方法，DynamicServerListLoadBalancer 先调用父类 BaseLoadBalancer 初始化，然后又做了一些剩余的初始化工作。

public ZoneAwareLoadBalancer(IClientConfig clientConfig, IRule rule,
                             IPing ping, ServerList<T> serverList, ServerListFilter<T> filter,
                             ServerListUpdater serverListUpdater) {
    // DynamicServerListLoadBalancer
    super(clientConfig, rule, ping, serverList, filter, serverListUpdater);
}

public DynamicServerListLoadBalancer(IClientConfig clientConfig, IRule rule, IPing ping,
                                     ServerList<T> serverList, ServerListFilter<T> filter,
                                     ServerListUpdater serverListUpdater) {
    // BaseLoadBalancer
    super(clientConfig, rule, ping);
    this.serverListImpl = serverList;
    this.filter = filter;
    this.serverListUpdater = serverListUpdater;
    if (filter instanceof AbstractServerListFilter) {
        ((AbstractServerListFilter) filter).setLoadBalancerStats(getLoadBalancerStats());
    }
    // 剩余的一些初始化
    restOfInit(clientConfig);
}

public BaseLoadBalancer(IClientConfig config, IRule rule, IPing ping) {
    // createLoadBalancerStatsFromConfig => LoadBalancerStats 统计
    initWithConfig(config, rule, ping, createLoadBalancerStatsFromConfig(config));

看 BaseLoadBalancer 的 initWithConfig，主要做了如下的初始化：

• 设置 IPing 和 IRule，ping 的间隔时间是 30 秒，setPing 会启动一个后台定时任务，然后每隔30秒运行一次 PingTask 任务。
• 设置了 ILoadBalancer 的 统计器 LoadBalancerStats，对 ILoadBalancer 的 Server 状态进行统计，比如连接失败、成功、熔断等信息。
• 在启用 PrimeConnections 请求预热的情况下，创建 PrimeConnections 来预热客户端 与 Server 的链接。默认是关闭的。
• 最后是注册了一些监控、开启请求预热。

void initWithConfig(IClientConfig clientConfig, IRule rule, IPing ping, LoadBalancerStats stats) {
    this.config = clientConfig;
    String clientName = clientConfig.getClientName();
    this.name = clientName;
    // ping 间隔时间，默认30秒
    int pingIntervalTime = Integer.parseInt(""
            + clientConfig.getProperty(
                    CommonClientConfigKey.NFLoadBalancerPingInterval,
                    Integer.parseInt("30")));
    // 没看到用的地方                
    int maxTotalPingTime = Integer.parseInt(""
            + clientConfig.getProperty(
                    CommonClientConfigKey.NFLoadBalancerMaxTotalPingTime,
                    Integer.parseInt("2")));
    // 设置 ping 间隔时间，并重新设置了 ping 任务
    setPingInterval(pingIntervalTime);
    setMaxTotalPingTime(maxTotalPingTime);

    // 设置 IRule、IPing
    setRule(rule);
    setPing(ping);

    setLoadBalancerStats(stats);
    rule.setLoadBalancer(this);
    if (ping instanceof AbstractLoadBalancerPing) {
        ((AbstractLoadBalancerPing) ping).setLoadBalancer(this);
    }
    logger.info("Client: {} instantiated a LoadBalancer: {}", name, this);
    
    // PrimeConnections，请求预热，默认关闭
    // 作用主要用于解决那些部署环境(如读EC2)在实际使用实时请求之前，从防火墙连接/路径进行预热（比如先加白名单、初始化等等动作比较耗时，可以用它先去打通）。
    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();
}

protected void init() {
    Monitors.registerObject("LoadBalancer_" + name, this);
    // register the rule as it contains metric for available servers count
    Monitors.registerObject("Rule_" + name, this.getRule());
    // 默认关闭
    if (enablePrimingConnections && primeConnections != null) {
        primeConnections.primeConnections(getReachableServers());
    }
}

再看下 DynamicServerListLoadBalancer 的初始化，核心的初始化逻辑在 restOfInit 中，主要就是做了两件事情：

• 开启动态更新 Server 的特性，比如实例上线、下线、故障等，要能够更新 ILoadBalancer 的 Server 列表。
• 然后就全量更新一次本地的 Server 列表。

void restOfInit(IClientConfig clientConfig) {
    boolean primeConnection = this.isEnablePrimingConnections();
    // turn this off to avoid duplicated asynchronous priming done in BaseLoadBalancer.setServerList()
    this.setEnablePrimingConnections(false);
    
    // 开启动态更新 Server 的特性
    enableAndInitLearnNewServersFeature();

    // 更新 Server 列表
    updateListOfServers();
    
    // 开启请求预热的情况下，对可用的 Server 进行预热
    if (primeConnection && this.getPrimeConnections() != null) {
        this.getPrimeConnections()
                .primeConnections(getReachableServers());
    }
    this.setEnablePrimingConnections(primeConnection);
    LOGGER.info("DynamicServerListLoadBalancer for client {} initialized: {}", clientConfig.getClientName(), this.toString());
}

2、全量更新Server列表

先看下 updateListOfServers() 是如何更新 Server 列表的，进而看下 ILoadBalancer 是如何存储 Server 的。

• 首先使用 ServerList 获取所有的 Server 列表，在 RibbonClientConfiguration 中配置的是 ConfigurationBasedServerList，但和 eureka 集合和，就不是 ConfigurationBasedServerList 了，这块下一节再来看。
• 然后使用 ServerListFilter 对 Server 列表过滤，其默认实现类是 ZonePreferenceServerListFilter，它主要是过滤出当前 Zone（defaultZone）下的 Server。
• 最后就是更新所有 Server 列表，先是设置 Server alive，然后调用父类（BaseLoadBalancer）的 setServersList 来更新Server列表，这说明 Server 是存储在 BaseLoadBalancer 里的。

public void updateListOfServers() {
    List<T> servers = new ArrayList<T>();
    if (serverListImpl != null) {
        // 从 ServerList 获取所有 Server 列表
        servers = serverListImpl.getUpdatedListOfServers();
        LOGGER.debug("List of Servers for {} obtained from Discovery client: {}", getIdentifier(), servers);

        if (filter != null) {
            // 用 ServerListFilter 过滤 Server
            servers = filter.getFilteredListOfServers(servers);
            LOGGER.debug("Filtered List of Servers for {} obtained from Discovery client: {}", getIdentifier(), servers);
        }
    }
    // 更新所有 Server 到本地缓存
    updateAllServerList(servers);
}

protected void updateAllServerList(List<T> ls) {
    if (serverListUpdateInProgress.compareAndSet(false, true)) {
        try {
            for (T s : ls) {
                s.setAlive(true); // 设置 Server alive
            }
            setServersList(ls);
            // 强制初始化 Ping
            super.forceQuickPing();
        } finally {
            serverListUpdateInProgress.set(false);
        }
    }
}

public void setServersList(List lsrv) {
    // BaseLoadBalancer
    super.setServersList(lsrv);
    
    // 将 Server 更新到 LoadBalancerStats 统计中 ....
}

接着看父类的 setServersList，可以看出，存储所有 Server 的数据结构 allServerList 是一个加了 synchronized 的线程安全的容器，setServersList 就是直接将得到的 Server 列表替换  allServerList。

public void setServersList(List lsrv) {
    Lock writeLock = allServerLock.writeLock();
    ArrayList<Server> newServers = new ArrayList<Server>();
    // 加写锁
    writeLock.lock();
    try {
        // for 循环将 lsrv 中的 Server 转移到 allServers
        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) {
                logger.debug("LoadBalancer [{}]:  addServer [{}]", name, ((Server) server).getId());
                allServers.add((Server) server);
            } else {
                throw new IllegalArgumentException("Type String or Server expected, instead found:" + server.getClass());
            }
        }
        
        boolean listChanged = false;
        // allServerList => volatile List<Server> allServerList = Collections.synchronizedList(new ArrayList<Server>())
        if (!allServerList.equals(allServers)) {
            listChanged = true;
            // 服务列表变更监听器 ServerListChangeListener， 发出服务变更通知...
        }
        
        // 启用了服务预热，开始 Server 预热...
        
        // 直接替换
        allServerList = allServers;
        if (canSkipPing()) {
            for (Server s : allServerList) {
                s.setAlive(true);
            }
            upServerList = allServerList;
        } else if (listChanged) {
            forceQuickPing();
        }
    } finally {
        // 释放写锁
        writeLock.unlock();
    }
}

前面 chooseRoundRobinAfterFiltering 获取所有 Server 时就是返回的这个 allServerList列表。

public List<Server> getAllServers() {
    return Collections.unmodifiableList(allServerList);
}

3、Eureka Ribbon 客户端配置

获取 Server 的组件是 ServerList，RibbonClientConfiguration 中配置的默认实现类是 ConfigurationBasedServerList。ConfigurationBasedServerList 默认是从配置文件中获取，可以像下面这样配置服务实例地址，多个 Server 地址用逗号隔开。

demo-producer:
  ribbon:
    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

@Configuration(proxyBeanMethods = false)
@EnableConfigurationProperties
@ConditionalOnRibbonAndEurekaEnabled
@AutoConfigureAfter(RibbonAutoConfiguration.class)
@RibbonClients(defaultConfiguration = EurekaRibbonClientConfiguration.class)
public class RibbonEurekaAutoConfiguration {

}

进入 EurekaRibbonClientConfiguration  可以看到：

• IPing 的默认实现类为 NIWSDiscoveryPing。
• ServerList 的默认实现类为 DomainExtractingServerList，但是 DomainExtractingServerList 在构造时又传入了一个类型为 DiscoveryEnabledNIWSServerList 的 ServerList。看名字大概也可以看出，DiscoveryEnabledNIWSServerList 就是从 EurekaClient 获取 Server 的组件。

@Configuration(proxyBeanMethods = false)
public class EurekaRibbonClientConfiguration {
    @Value("${ribbon.eureka.approximateZoneFromHostname:false}")
    private boolean approximateZoneFromHostname = false;

    @RibbonClientName
    private String serviceId = "client";
    @Autowired
    private PropertiesFactory propertiesFactory;

    @Bean
    @ConditionalOnMissingBean
    public IPing ribbonPing(IClientConfig config) {
        if (this.propertiesFactory.isSet(IPing.class, serviceId)) {
            return this.propertiesFactory.get(IPing.class, config, serviceId);
        }
        NIWSDiscoveryPing ping = new NIWSDiscoveryPing();
        ping.initWithNiwsConfig(config);
        return ping;
    }

    @Bean
    @ConditionalOnMissingBean
    public ServerList<?> ribbonServerList(IClientConfig config,
            Provider<EurekaClient> eurekaClientProvider) {
        if (this.propertiesFactory.isSet(ServerList.class, serviceId)) {
            return this.propertiesFactory.get(ServerList.class, config, serviceId);
        }
        DiscoveryEnabledNIWSServerList discoveryServerList = new DiscoveryEnabledNIWSServerList(config, eurekaClientProvider);
        DomainExtractingServerList serverList = new DomainExtractingServerList(discoveryServerList, config, this.approximateZoneFromHostname);
        return serverList;
    }
}

4、从 DiscoveryClient 获取Server列表

DynamicServerListLoadBalancer 中通过 ServerList 的 getUpdatedListOfServers 方法全量获取服务列表，在 eureka-client 环境下，ServerList 默认实现类为 DomainExtractingServerList，那就先看下它的 getUpdatedListOfServers 方法。

可以看出，DomainExtractingServerList 先用 DomainExtractingServerList 获取服务列表，然后根据 Ribbon 客户端配置重新构造 Server 对象返回。获取服务列表的核心在 DiscoveryEnabledNIWSServerList 中。

@Override
public List<DiscoveryEnabledServer> getUpdatedListOfServers() {
    // list => DiscoveryEnabledNIWSServerList
    List<DiscoveryEnabledServer> servers = setZones(this.list.getUpdatedListOfServers());
    return servers;
}

private List<DiscoveryEnabledServer> setZones(List<DiscoveryEnabledServer> servers) {
    List<DiscoveryEnabledServer> result = new ArrayList<>();
    boolean isSecure = this.ribbon.isSecure(true);
    boolean shouldUseIpAddr = this.ribbon.isUseIPAddrForServer();
    // 根据客户端配置重新构造 DomainExtractingServer 返回
    for (DiscoveryEnabledServer server : servers) {
        result.add(new DomainExtractingServer(server, isSecure, shouldUseIpAddr, this.approximateZoneFromHostname));
    }
    return result;
}

先看下 DiscoveryEnabledNIWSServerList  的构造初始化：

• 主要是传入了 Provider<EurekaClient> 用来获取 EurekaClient
• 另外还设置了客户端名称 clientName ，以及 vipAddresses 也是客户端名称，这个后面会用得上。

public DiscoveryEnabledNIWSServerList(IClientConfig clientConfig, Provider<EurekaClient> eurekaClientProvider) {
    this.eurekaClientProvider = eurekaClientProvider;
    initWithNiwsConfig(clientConfig);
}

@Override
public void initWithNiwsConfig(IClientConfig clientConfig) {
    // 客户端名称，就是服务名称
    clientName = clientConfig.getClientName();
    // vipAddresses 得到的也是客户端名称
    vipAddresses = clientConfig.resolveDeploymentContextbasedVipAddresses();
    
    // 其它的一些配置....
}

接着看获取实例的 getUpdatedListOfServers，可以看到它的核心逻辑就是根据服务名从 EurekaClient 获取 InstanceInfo 实例列表，然后封装 Server 信息返回。

public List<DiscoveryEnabledServer> getUpdatedListOfServers(){
    return obtainServersViaDiscovery();
}

private List<DiscoveryEnabledServer> obtainServersViaDiscovery() {
    List<DiscoveryEnabledServer> serverList = new ArrayList<DiscoveryEnabledServer>();
    if (eurekaClientProvider == null || eurekaClientProvider.get() == null) {
        return new ArrayList<DiscoveryEnabledServer>();
    }
    // 得到 EurekaClient，实际类型是 CloudEurekaClient，其父类是 DiscoveryClient
    EurekaClient eurekaClient = eurekaClientProvider.get();
    if (vipAddresses!=null){
        // 分割 vipAddresses，默认就是服务名称
        for (String vipAddress : vipAddresses.split(",")) {
            // 根据服务名称从 EurekaClient 获取实例信息
            List<InstanceInfo> listOfInstanceInfo = eurekaClient.getInstancesByVipAddress(vipAddress, isSecure, targetRegion);
            for (InstanceInfo ii : listOfInstanceInfo) {
                if (ii.getStatus().equals(InstanceStatus.UP)) {
                    // ...
                    // 根据实例信息 InstanceInfo 创建 Server
                    DiscoveryEnabledServer des = createServer(ii, isSecure, shouldUseIpAddr);
                    serverList.add(des);
                }
            }
            if (serverList.size()>0 && prioritizeVipAddressBasedServers){
                break; // if the current vipAddress has servers, we dont use subsequent vipAddress based servers
            }
        }
    }
    return serverList;
}

注意这里的 vipAddress 其实就是服务名：

最后看 EurekaClient 的 getInstancesByVipAddress，到这里就很清楚了，其实就是从 DiscoveryClient 的本地应用 Applications 中根据服务名取出所有的实例列表。

这里就和 Eureka 源码那块衔接上了，eureka-client 全量抓取注册表以及每隔30秒增量抓取注册表，都是合并到本地的 Applications 中。Ribbon 与 Eureka 结合后，Ribbon 获取 Server 就从 DiscoveryClient 的 Applications 中获取 Server 列表了。

public List<InstanceInfo> getInstancesByVipAddress(String vipAddress, boolean secure, String region) {
    // ...
    Applications applications;
    if (instanceRegionChecker.isLocalRegion(region)) {
        // 取本地应用 Applications
        applications = this.localRegionApps.get();
    } else {
        applications = remoteRegionVsApps.get(region);
        if (null == applications) {
            return Collections.emptyList();
        }
    }

    if (!secure) {
        // 返回服务名对应的实例
        return applications.getInstancesByVirtualHostName(vipAddress);
    } else {
        return applications.getInstancesBySecureVirtualHostName(vipAddress);
    }
}

5、定时更新Server列表

DynamicServerListLoadBalancer 初始化时，有个方法还没说，就是 enableAndInitLearnNewServersFeature()。这个方法只是调用 ServerListUpdater 启动了一个 UpdateAction，这个 UpdateAction 又只是调用了一下 updateListOfServers 方法，就是前面讲解过的全量更新 Server 的逻辑。

public void enableAndInitLearnNewServersFeature() {
    serverListUpdater.start(updateAction);
}

protected final ServerListUpdater.UpdateAction updateAction = new ServerListUpdater.UpdateAction() {
    @Override
    public void doUpdate() {
        // 调用 updateListOfServers
        updateListOfServers();
    }
};

ServerListUpdater 的默认实现类是 PollingServerListUpdater，看下它的 start 方法：

其实就是以固定的频率，每隔30秒调用一下 updateListOfServers 方法，将 DiscoveryClient 中 Applications 中缓存的实例同步到 ILoadBalancer 中的 allServerList 列表中。

public synchronized void start(final UpdateAction updateAction) {
    if (isActive.compareAndSet(false, true)) {
        final Runnable wrapperRunnable = new Runnable() {
            @Override
            public void run() {
                // ...
                try {
                    // 执行一次 updateListOfServers
                    updateAction.doUpdate();
                    // 设置最后更新时间
                    lastUpdated = System.currentTimeMillis();
                } catch (Exception e) {
                    logger.warn("Failed one update cycle", e);
                }
            }
        };
        
        // 固定频率调度
        scheduledFuture = getRefreshExecutor().scheduleWithFixedDelay(
                wrapperRunnable,
                initialDelayMs, // 默认 1000
                refreshIntervalMs, // 默认 30 * 1000
                TimeUnit.MILLISECONDS
        );
    } else {
        logger.info("Already active, no-op");
    }
}

6、判断Server是否存活

在创建 ILoadBalancer 时，IPing 还没有看过是如何工作的。在初始化的时候，可以看到，主要就是设置了当前的 ping，然后重新设置了一个调度任务，默认每隔30秒调度一次 PingTask 任务。

public void setPing(IPing ping) {
    if (ping != null) {
        if (!ping.equals(this.ping)) {
            this.ping = ping;
            // 设置 Ping 任务
            setupPingTask();
        }
    } else {
        this.ping = null;
        // cancel the timer task
        lbTimer.cancel();
    }
}

void setupPingTask() {
    // ...
    // 创建一个定时调度器
    lbTimer = new ShutdownEnabledTimer("NFLoadBalancer-PingTimer-" + name, true);
    // pingIntervalTime 默认为 30 秒，每隔30秒调度一次 PingTask
    lbTimer.schedule(new PingTask(), 0, pingIntervalSeconds * 1000);
    // 立即发起以 Ping
    forceQuickPing();
}

ShutdownEnabledTimer 可以简单了解下，它是继承自 Timer 的，它在创建的时候向 Runtime 注册了一个回调，在 jvm 关闭的时候来取消 Timer 的执行，进而释放资源。

public class ShutdownEnabledTimer extends Timer {
    private static final Logger LOGGER = LoggerFactory.getLogger(ShutdownEnabledTimer.class);

    private Thread cancelThread;
    private String name;

    public ShutdownEnabledTimer(String name, boolean daemon) {
        super(name, daemon);
        this.name = name;
        // 取消定时器的线程
        this.cancelThread = new Thread(new Runnable() {
            public void run() {
                ShutdownEnabledTimer.super.cancel();
            }
        });

        LOGGER.info("Shutdown hook installed for: {}", this.name);
        // 向 Runtime 注册一个钩子，在 jvm 关闭时，调用 cancelThread 取消定时任务
        Runtime.getRuntime().addShutdownHook(this.cancelThread);
    }

    @Override
    public void cancel() {
        super.cancel();
        LOGGER.info("Shutdown hook removed for: {}", this.name);
        try {
            Runtime.getRuntime().removeShutdownHook(this.cancelThread);
        } catch (IllegalStateException ise) {
            LOGGER.info("Exception caught (might be ok if at shutdown)", ise);
        }

    }
}

再来看下 PingTask，PingTask 核心逻辑就是遍历 allServers 列表，使用 IPingStrategy 和 IPing 来判断 Server 是否存活，并更新 Server 的状态，以及将所有存活的 Server 更新到 upServerList 中，upServerList 缓存了所有存活的 Server。

class PingTask extends TimerTask {
    public void run() {
        try {
            // pingStrategy => SerialPingStrategy
            new Pinger(pingStrategy).runPinger();
        } catch (Exception e) {
            logger.error("LoadBalancer [{}]: Error pinging", name, 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; // Ping in progress - nothing to do
        }

        Server[] allServers = null;
        boolean[] results = null;

        Lock allLock = null;
        Lock upLock = null;
        
        try {
            allLock = allServerLock.readLock();
            allLock.lock();
            // 加读锁，取出 allServerList 中的 Server
            allServers = allServerList.toArray(new Server[allServerList.size()]);
            allLock.unlock();

            int numCandidates = allServers.length;
            // 使用 IPingStrategy 和 IPing 对所有 Server 发起 ping 请求
            results = pingerStrategy.pingServers(ping, allServers);

            final List<Server> newUpList = new ArrayList<Server>();
            final List<Server> changedServers = new ArrayList<Server>();

            for (int i = 0; i < numCandidates; i++) {
                boolean isAlive = results[i];
                Server svr = allServers[i];
                boolean oldIsAlive = svr.isAlive();
                // 设置 alive 是否存活
                svr.setAlive(isAlive);

                // 实例变更
                if (oldIsAlive != isAlive) {
                    changedServers.add(svr);
                    logger.debug("LoadBalancer [{}]:  Server [{}] status changed to {}",  name, svr.getId(), (isAlive ? "ALIVE" : "DEAD"));
                }

                // 添加存活的 Server
                if (isAlive) {
                    newUpList.add(svr);
                }
            }
            upLock = upServerLock.writeLock();
            upLock.lock();
            // 更新 upServerList，upServerList 只保存了存活的 Server
            upServerList = newUpList;
            upLock.unlock();
            // 通知变更
            notifyServerStatusChangeListener(changedServers);
        } finally {
            pingInProgress.set(false);
        }
    }
}

IPingStrategy 的默认实现类是 SerialPingStrategy，进入可以发现它只是遍历所有 Server，然后用 IPing 判断 Server 是否存活。

private static class SerialPingStrategy implements IPingStrategy {
    @Override
    public boolean[] pingServers(IPing ping, Server[] servers) {
        int numCandidates = servers.length;
        boolean[] results = new boolean[numCandidates];

        for (int i = 0; i < numCandidates; i++) {
            results[i] = false;
            try {
                if (ping != null) {
                    // 使用 IPing 判断 Server 是否存活
                    results[i] = ping.isAlive(servers[i]);
                }
            } catch (Exception e) {
                logger.error("Exception while pinging Server: '{}'", servers[i], e);
            }
        }
        return results;
    }
}

在集成 eureka-client 后，IPing默认实现类是 NIWSDiscoveryPing，看它的 isAlive 方法，其实就是判断对应 Server 的实例 InstanceInfo 的状态是否是 UP 状态，UP状态就表示 Server 存活。

public boolean isAlive(Server server) {
    boolean isAlive = true;
    if (server!=null && server instanceof DiscoveryEnabledServer){
        DiscoveryEnabledServer dServer = (DiscoveryEnabledServer)server;                
        InstanceInfo instanceInfo = dServer.getInstanceInfo();
        if (instanceInfo!=null){                    
            InstanceStatus status = instanceInfo.getStatus();
            if (status!=null){
                // 判断Server对应的实例状态是否是 UP
                isAlive = status.equals(InstanceStatus.UP);
            }
        }
    }
    return isAlive;
}

7、一张图总结 Ribbon 核心原理

① Ribbon 核心工作原理总结

• 首先，Ribbon 的7个核心接口共同定义了 Ribbon 的行为特性，它们就是 Ribbon 的核心骨架。
• 使用 Ribbon 来对客户端做负载均衡，基本的用法就是用 @LoadBalanced 注解标注一个 RestTemplate 的 bean 对象，之后在 LoadBalancerAutoConfiguration 配置类中会对带有 @LoadBalanced 注解的 RestTemplate 添加 LoadBalancerInterceptor 拦截器。
• LoadBalancerInterceptor 会拦截 RestTemplate 的 HTTP 请求，将请求绑定进 Ribbon 负载均衡的生命周期，然后使用 LoadBalancerClient 的 execute 方法来处理请求。
• LoadBalancerClient 首先会得到一个 ILoadBalancer，再使用它去得到一个 Server，这个 Server 就是具体某一个实例的信息封装。得到 Server 之后，就用 Server 的 IP 和端口重构原始 URI。
• ILoadBalancer 最终在选择实例的时候，会通过 IRule 均衡策略来选择一个 Server。
• ILoadBalancer 的父类 BaseLoadBalancer 中有一个 allServerList 列表缓存了所有 Server，Ribbon 中 Server 的来源就是 allServerList。
• 在加载Ribbon客户端上下文时，ILoadBalancer 会用 ServerList 从 DiscoveryClient 的 Applications 中获取客户端对应的实例列表，然后使用 ServerListFilter 过滤，最后更新到 allServerList 中。
• ILoadBalancer 还会开启一个后台任务 ServerListUpdater ，每隔30秒运行一次，用 ServerList 将 DiscoveryClient 的 Applications 中的实例列表同步到 allServerList 中。
• ILoadBalancer 还会开启一个后台任务 PingTask，每隔30秒运行一次，用 IPing 判断 Server 的存活状态，EurekaClient 环境下，就是判断 InstanceInfo 的状态是否为 UP。

② 下面用一张图来总结下 Ribbon 这块获取Server的核心流程以及对应的核心接口间的关系。

8、Ribbon 脱离 Eureka 使用

在默认情况下，Ribbon 客户端会从 EurekaClient 获取服务列表，其实就是间接从注册中心读取服务注册信息列表，来达到动态负载均衡的功能。但如果不想从 EurekaClient 读取，可以禁用 Ribbon 的 Eureka 功能。
在 Ribbon 中禁用Eureka功能，可以做如下配置：

ribbon:
  eureka:
    enabled: false

那 ribbon.eureka.enabled 是如何控制禁用 Eureka 的呢？看 RibbonEurekaAutoConfiguration：

@Configuration(proxyBeanMethods = false)
@EnableConfigurationProperties
@ConditionalOnRibbonAndEurekaEnabled
@AutoConfigureAfter(RibbonAutoConfiguration.class)
@RibbonClients(defaultConfiguration = EurekaRibbonClientConfiguration.class)
public class RibbonEurekaAutoConfiguration {

}

这个配置类通过 @RibbonClients 指定了默认的客户端配置类为 EurekaRibbonClientConfiguration，但生效的前提是 @ConditionalOnRibbonAndEurekaEnabled，进去可以看到这个条件注解就是判断 Ribbon Eureka 是否启用了，就可以设置 ribbon.eureka.enabled=false 来禁用 RIbbon Eureka。

@Target({ ElementType.TYPE, ElementType.METHOD })
@Retention(RetentionPolicy.RUNTIME)
@Documented
// 条件类 OnRibbonAndEurekaEnabledCondition
@Conditional(ConditionalOnRibbonAndEurekaEnabled.OnRibbonAndEurekaEnabledCondition.class)
public @interface ConditionalOnRibbonAndEurekaEnabled {

    class OnRibbonAndEurekaEnabledCondition extends AllNestedConditions {

        OnRibbonAndEurekaEnabledCondition() {
            super(ConfigurationPhase.REGISTER_BEAN);
        }
        
        // 引入了类：DiscoveryEnabledNIWSServerList 
        @ConditionalOnClass(DiscoveryEnabledNIWSServerList.class)
        // 存在bean对象：SpringClientFactory
        @ConditionalOnBean(SpringClientFactory.class)
        // ribbon.eureka.enabled=true
        @ConditionalOnProperty(value = "ribbon.eureka.enabled", matchIfMissing = true)
        static class Defaults {

        }

        // 存在bean对象：EurekaClient
        @ConditionalOnBean(EurekaClient.class)
        static class EurekaBeans {

        }
        
        // eureka.client.enabled=true
        @ConditionalOnProperty(value = "eureka.client.enabled", matchIfMissing = true)
        @ConditionalOnDiscoveryEnabled
        static class OnEurekaClientEnabled {

        }
    }
}

如果想从其它地方获取服务列表，可以自定义接口实现 ServerList<Server> 来获取，也可以在配置文件中设置地址列表：

<client-name>:
  ribbon:
    listOfServers: http://10.215.0.92:8010,http://10.215.0.92:8011