SpringCloud--Ribbon--源码解析--Ribbon入口实现

Posted 朦胧的夜~

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了SpringCloud--Ribbon--源码解析--Ribbon入口实现相关的知识,希望对你有一定的参考价值。

  Ribbon总体的源码结构,如下图所示:

   上图是Ribbon源码的总览图,每一个颜色,代表源码中的一部分内容,总体来看,Ribbon源码的实现,总共分为五个部分,从上而下依次为,Ribbon入口实现、IloadBalancer实现、ServerListUpdater实现、ServerListFilter实现和IRule实现。

  针对这些实现,粗略描述如下:

实现内容 描述
Ribbon入口实现 自动配置加载及请求拦截器加载
IloadBalancer实现  负载均衡器
ServerListUpdater实现  动态更新本地存储的实例清单
ServerListFilter实现  过滤器
IRule实现  负载均衡策略

  本篇主要说Ribbon入口实现

  在SpringCloud--Eureka--搭建中,对于使用Ribbon的项目,主要是在主函数上增加了如下代码

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

  然后在后续的调用中,直接使用restTemplate进行调用,那么我们就可以先看一下@LoadBalanced注解,从@LoadBalanced注解源码可以看到,该注解用来给RestTemplate做标记,以使用负载均衡客户端LoadBalancerClient

  该客户端接口提供了三个方法:

方法 描述 源码
继承父类的choose方法 根据传入的服务名,从负载均衡器中挑选一个对应的服务实例 ServiceInstance choose(String serviceId);
execute方法 使用从负载均衡器中选择的服务实例进行请求

<T> T execute(String serviceId, LoadBalancerRequest<T> request)

<T> T execute(String serviceId, ServiceInstance serviceInstance,LoadBalancerRequest<T> request)

reconstructURI 为系统构建一个合适的host:port形式的URI,其中请求的ServiceInstance一般是 服务名:port,返回的URI则是ServiceInstance服务实例详情拼接的具体host:port形式的请求地址 URI reconstructURI(ServiceInstance instance, URI original);

  根据LoadBalancerClient查看,相关类的关系图如下所示:

   在上图中,通过类的命名可以看出LoadBalancerAutoConfiguration为客户端负载均衡器的自动化配置类

@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(RestTemplate.class)
@ConditionalOnBean(LoadBalancerClient.class)
@EnableConfigurationProperties(LoadBalancerRetryProperties.class)
public class LoadBalancerAutoConfiguration {

    @LoadBalanced
    @Autowired(required = false)
    private List<RestTemplate> restTemplates = Collections.emptyList();

    @Autowired(required = false)
    private List<LoadBalancerRequestTransformer> transformers = Collections.emptyList();

    @Bean
    public SmartInitializingSingleton loadBalancedRestTemplateInitializerDeprecated(
            final ObjectProvider<List<RestTemplateCustomizer>> restTemplateCustomizers) {
        return () -> restTemplateCustomizers.ifAvailable(customizers -> {
            for (RestTemplate restTemplate : LoadBalancerAutoConfiguration.this.restTemplates) {
                for (RestTemplateCustomizer customizer : customizers) {
                    customizer.customize(restTemplate);
                }
            }
        });
    }

    @Bean
    @ConditionalOnMissingBean
    public LoadBalancerRequestFactory loadBalancerRequestFactory(
            LoadBalancerClient loadBalancerClient) {
        return new LoadBalancerRequestFactory(loadBalancerClient, this.transformers);
    }

    @Configuration(proxyBeanMethods = false)
    @ConditionalOnMissingClass("org.springframework.retry.support.RetryTemplate")
    static class LoadBalancerInterceptorConfig {

        @Bean
        public LoadBalancerInterceptor ribbonInterceptor(
                LoadBalancerClient loadBalancerClient,
                LoadBalancerRequestFactory requestFactory) {
            return new LoadBalancerInterceptor(loadBalancerClient, requestFactory);
        }

        @Bean
        @ConditionalOnMissingBean
        public RestTemplateCustomizer restTemplateCustomizer(
                final LoadBalancerInterceptor loadBalancerInterceptor) {
            return restTemplate -> {
                List<ClientHttpRequestInterceptor> list = new ArrayList<>(
                        restTemplate.getInterceptors());
                list.add(loadBalancerInterceptor);
                restTemplate.setInterceptors(list);
            };
        }

    }

    /**
     * Auto configuration for retry mechanism.
     */
    @Configuration(proxyBeanMethods = false)
    @ConditionalOnClass(RetryTemplate.class)
    public static class RetryAutoConfiguration {

        @Bean
        @ConditionalOnMissingBean
        public LoadBalancedRetryFactory loadBalancedRetryFactory() {
            return new LoadBalancedRetryFactory() {
            };
        }

    }

    /**
     * Auto configuration for retry intercepting mechanism.
     */
    @Configuration(proxyBeanMethods = false)
    @ConditionalOnClass(RetryTemplate.class)
    public static class RetryInterceptorAutoConfiguration {

        @Bean
        @ConditionalOnMissingBean
        public RetryLoadBalancerInterceptor ribbonInterceptor(
                LoadBalancerClient loadBalancerClient,
                LoadBalancerRetryProperties properties,
                LoadBalancerRequestFactory requestFactory,
                LoadBalancedRetryFactory loadBalancedRetryFactory) {
            return new RetryLoadBalancerInterceptor(loadBalancerClient, properties,
                    requestFactory, loadBalancedRetryFactory);
        }

        @Bean
        @ConditionalOnMissingBean
        public RestTemplateCustomizer restTemplateCustomizer(
                final RetryLoadBalancerInterceptor loadBalancerInterceptor) {
            return restTemplate -> {
                List<ClientHttpRequestInterceptor> list = new ArrayList<>(
                        restTemplate.getInterceptors());
                list.add(loadBalancerInterceptor);
                restTemplate.setInterceptors(list);
            };
        }

    }

}

从源码类的注解上可以知道Ribbon负载均衡必须满足如下两个条件:

  1、@ConditionalOnClass(RestTemplate.class):RestTemplate类必须存在该项目中

  2、@ConditionalOnBean(LoadBalancerClient.class):必须要有LoadBalancerClient的先险类Bean

同时,该自动化配置类主要做了一下内容:

  1、创建了一个LoadBalancerInterceptor的Bean,用以实现对客户端请求的拦截,以实现客户端负载均衡

  2、创建了一个RestTemplateCustomizer的Bean,用以给restTemplate增加LoadBalancerInterceptor拦截器

  3、维护了一个@LoadBalanced注解修饰的RestTemplate集合,并且进行初始化,通过调用RestTemplateCustomizer的实例来给客户端的RestTemplate添加LoadBalancerInterceptor拦截器

那么接下来看下LoadBalancerInterceptor是如何将RestTemplate进行负载均衡的

public class LoadBalancerInterceptor implements ClientHttpRequestInterceptor {

    private LoadBalancerClient loadBalancer;

    private LoadBalancerRequestFactory requestFactory;

    public LoadBalancerInterceptor(LoadBalancerClient loadBalancer,
            LoadBalancerRequestFactory requestFactory) {
        this.loadBalancer = loadBalancer;
        this.requestFactory = requestFactory;
    }

    public LoadBalancerInterceptor(LoadBalancerClient loadBalancer) {
        // for backwards compatibility
        this(loadBalancer, new LoadBalancerRequestFactory(loadBalancer));
    }

    @Override
    public ClientHttpResponse intercept(final HttpRequest request, final byte[] body,
            final ClientHttpRequestExecution execution) throws IOException {
        final URI originalUri = request.getURI();
        String serviceName = originalUri.getHost();
        Assert.state(serviceName != null,
                "Request URI does not contain a valid hostname: " + originalUri);
        return this.loadBalancer.execute(serviceName,
                this.requestFactory.createRequest(request, body, execution));
    }

}

  通过源码可以看到,LoadBalancerInterceptor的构造函数传入了LoadBalancerClient,因此当一个被@LoadBalanced注解修饰的RestTemplate对外发起http请求时,就会被LoadBalancerInterceptor拦截,并调用LoadBalancerClient中的execute方法。

  RibbonLoadBalancerClient是LoadBalancerClient的实现类,其中的execute方法如下所示:

    public <T> T execute(String serviceId, LoadBalancerRequest<T> request, Object hint)
            throws IOException {
        ILoadBalancer loadBalancer = getLoadBalancer(serviceId);
        Server server = getServer(loadBalancer, hint);
        if (server == null) {
            throw new IllegalStateException("No instances available for " + serviceId);
        }
        RibbonServer ribbonServer = new RibbonServer(serviceId, server,
                isSecure(server, serviceId),
                serverIntrospector(serviceId).getMetadata(server));

        return execute(serviceId, ribbonServer, request);
    }

  由以上代码可见,先是使用ILoadBalancer获取了一个Server对象,然后将Server对象封装成RibbonServer对象(该对象除了存储服务实例的信息之外,还增加了服务名ServiceId、是否需要使用HTTPS等其他信息),最后使用该对象在回调LoadBalancerInterceptor请求拦截器中LoadBalancerRequest的apply方法,从而向一个具体的实例发起请求,从而实现一开始以服务名为host的URI请求到host:port形式实际访问地址的转换。

    public <T> T execute(String serviceId, ServiceInstance serviceInstance,
            LoadBalancerRequest<T> request) throws IOException {
        Server server = null;
        if (serviceInstance instanceof RibbonServer) {
            server = ((RibbonServer) serviceInstance).getServer();
        }
        if (server == null) {
            throw new IllegalStateException("No instances available for " + serviceId);
        }

        RibbonLoadBalancerContext context = this.clientFactory
                .getLoadBalancerContext(serviceId);
        RibbonStatsRecorder statsRecorder = new RibbonStatsRecorder(context, server);

        try {
            T returnVal = request.apply(serviceInstance);
            statsRecorder.recordStats(returnVal);
            return returnVal;
        }
        // catch IOException and rethrow so RestTemplate behaves correctly
        catch (IOException ex) {
            statsRecorder.recordStats(ex);
            throw ex;
        }
        catch (Exception ex) {
            statsRecorder.recordStats(ex);
            ReflectionUtils.rethrowRuntimeException(ex);
        }
        return null;
    }

  由以上代码可见,最终会回调拦截器org.springframework.cloud.client.loadbalancer.AsyncLoadBalancerInterceptor#intercept,在拦截器中,调用了org.springframework.http.client.InterceptingAsyncClientHttpRequest.AsyncRequestExecution#executeAsync方法,该方法传入了一个参数ServiceRequestWrapper

        public ListenableFuture<ClientHttpResponse> executeAsync(HttpRequest request, byte[] body)
                throws IOException {

            if (this.iterator.hasNext()) {
                AsyncClientHttpRequestInterceptor interceptor = this.iterator.next();
                return interceptor.intercept(request, body, this);
            }
            else {
                URI uri = request.getURI();
                HttpMethod method = request.getMethod();
                HttpHeaders headers = request.getHeaders();

                Assert.state(method != null, "No standard HTTP method");
                AsyncClientHttpRequest delegate = requestFactory.createAsyncRequest(uri, method);
                delegate.getHeaders().putAll(headers);
                if (body.length > 0) {
                    StreamUtils.copy(body, delegate.getBody());
                }

                return delegate.executeAsync();
            }
        }
    }

可以看到,该方法中,首先获取要请求的URI,因为入参的HttpRequest为ServiceRequestWrapper,因此使用的是ServiceRequestWrapper中的个月URI,然后再调用requestFactory.createAsyncRequest(uri, method)来创建一个http请求。

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

顺着该方法向下看,最终访问到org.springframework.cloud.netflix.ribbon.RibbonLoadBalancerClient#reconstructURI方法

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

  从上述代码可以看到,首先从ServiceInstance对象中获取了serviceId,然后根据serviceId从SpringClientFactory对象中获取RibbonLoadBalancerContext对象,最后,调用RibbonLoadBalancerContext对象的reconstructURIWithServer方法,获取具体的URI地址。

  在上述代码中,SpringClientFactory类是用来创建客户端负载均衡的工厂类,该工厂类会为每一个不同名的Ribbon客户端生成不同的Spring上下文。

  RibbonLoadBalancerContext类是LoadBalancerContext类的子类,该类用于存储一些被负载均衡器使用的上下文内容和API操作。

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

  通过上述代码可以看到,最终是使用host和port最终拼接成了最终的请求地址。

以上是关于SpringCloud--Ribbon--源码解析--Ribbon入口实现的主要内容,如果未能解决你的问题,请参考以下文章

SpringCloud---Ribbon服务调用,源码分析

SpringCloud----Ribbon服务调用,源码分析

springCloud:Ribbon实现客户端侧负载均衡-自定义Ribbon配置

springCloud:Ribbon实现客户端侧负载均衡-消费者整合Ribbon

SpringCloud:ribbon项目搭建(Gradle项目)

SpringCloud Ribbon组成和负载均衡规则