Ribbon 学习(三)：RestTemplate 请求负载流程解析

- 日理万妓 2022-12-04 07:44 63阅读 0赞

# 说明 #

在上篇博文[《Ribbon 学习(二)：Spring Cloud Ribbon 加载配置原理》][Ribbon _Spring Cloud Ribbon]中，我简单介绍了 Spring Cloud Ribbon 的加载配置原理及 RestTemplate 使用 @LoadBalanced 注解后实现请求负载的具体原因 ---- 通过 LoadBalancerInterceptor 拦截器进行了请求服务地址的负载。在本篇博文中，我将通过源码来解析 RestTemplate 在发出请求后到响应的整体流程，来探究 Ribbon 在其中的作用原理。

本篇博文仍基于之前博文[《Ribbon 学习(一)：脱离 Eureka 使用 Ribbon》][Ribbon _ Eureka _ Ribbon] 中介绍的简单使用示例进行阅读解析，因此会忽略一些内容。

# 正文 #

先根据图示了解 RestTemplate 的请求流程  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM2NTUzOTEz_size_16_color_FFFFFF_t_70_pic_center]

在图示中，我将请求过程分为了四部分，分别为 发出请求、构造请求、请求拦截、负载均衡。

## 发出请求 ##

@RequestMapping("/hello/{name}")
    public String hello(@PathVariable(name = "name") String name) {
        return this.restTemplate.getForObject("http://test-server/hello/" + name, String.class);
    }

调用 RestTemplate 的 getForObject 方法发出请求，在该方法中主要根据 ResponseType 及 HttpMessageConverter 创建响应消息提前器 HttpMessageConverterExtractor，之后调用 execute 进行下步请求。在 execute 方法中根据请求地址 url 及 请求参数构建 URI，之后调用 doExecute 方法进行下步请求。在 doExecute 方法中，则根据 URI 和 请求方法 HTTP Method 构建请求对象 ClientHttpRequest，接着请求执行，处理响应，最后返回结果。

protected <T> T doExecute(URI url, @Nullable HttpMethod method, @Nullable RequestCallback requestCallback, @Nullable ResponseExtractor<T> responseExtractor) throws RestClientException {
        Assert.notNull(url, "URI is required");
        Assert.notNull(method, "HttpMethod is required");
        ClientHttpResponse response = null;
    
        Object var14;
        try {
            // 创建请求
            ClientHttpRequest request = this.createRequest(url, method);
            // 设置请求头
            if (requestCallback != null) {
                requestCallback.doWithRequest(request);
            }
            // 请求执行
            response = request.execute();
            // 处理响应
            this.handleResponse(url, method, response);
            // 提取转换结果
            var14 = responseExtractor != null ? responseExtractor.extractData(response) : null;
        } catch (IOException var12) {
            String resource = url.toString();
            String query = url.getRawQuery();
            resource = query != null ? resource.substring(0, resource.indexOf(63)) : resource;
            throw new ResourceAccessException("I/O error on " + method.name() + " request for \"" + resource + "\": " + var12.getMessage(), var12);
        } finally {
            if (response != null) {
                response.close();
            }
    
        }
        // 返回结果
        return var14;
    }

## 构造请求 ##

ClientHttpRequest request = this.createRequest(url, method);

在 createRequest 方法中先通过 getRequestFactory 获取了 RequestFactory ，而该方法则是调用的 InterceptingHttpAccessor 抽象类的方法，在该方法中则根据是否存在拦截器而创建了 InterceptingClientHttpRequestFactory 对象 ，同时调用它的父类 HttpAccessor 抽象类的 getRequestFactory 方法，将返回的 SimpleClientHttpRequestFactory 对象作为构造参数传递。

public ClientHttpRequestFactory getRequestFactory() {
        List<ClientHttpRequestInterceptor> interceptors = this.getInterceptors();
        if (!CollectionUtils.isEmpty(interceptors)) {
            ClientHttpRequestFactory factory = this.interceptingRequestFactory;
            if (factory == null) {
                factory = new InterceptingClientHttpRequestFactory(super.getRequestFactory(), interceptors);
                this.interceptingRequestFactory = (ClientHttpRequestFactory)factory;
            }
    
            return (ClientHttpRequestFactory)factory;
        } else {
            return super.getRequestFactory();
        }
    }

接着调用 createRequest 方法进行请求创建

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

可以看到创建了 InterceptingClientHttpRequest 对象，之后进行请求执行。

response = request.execute();

## 请求拦截 ##

InterceptingClientHttpRequest 类结构图  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM2NTUzOTEz_size_16_color_FFFFFF_t_70_pic_center 1]

调用抽象类 AbstractClientHttpRequest 的模板方法 execute

public final ClientHttpResponse execute() throws IOException {
        this.assertNotExecuted();
        ClientHttpResponse result = this.executeInternal(this.headers);
        this.executed = true;
        return result;
    }

真正调用 InterceptingClientHttpRequest 的 executeInternal 方法

protected final ClientHttpResponse executeInternal(HttpHeaders headers, byte[] bufferedOutput) throws IOException {
        InterceptingClientHttpRequest.InterceptingRequestExecution requestExecution = new InterceptingClientHttpRequest.InterceptingRequestExecution();
        return requestExecution.execute(this, bufferedOutput);
    }

在该方法中创建了 InterceptingClientHttpRequest 内部类 InterceptingRequestExecution 对象，调用其 execute 方法进行请求真正执行。

public ClientHttpResponse execute(HttpRequest request, byte[] body) throws IOException {
        if (this.iterator.hasNext()) {
            ClientHttpRequestInterceptor nextInterceptor = (ClientHttpRequestInterceptor)this.iterator.next();
            return nextInterceptor.intercept(request, body, this);
        } else {
            HttpMethod method = request.getMethod();
            Assert.state(method != null, "No standard HTTP method");
            ClientHttpRequest delegate = InterceptingClientHttpRequest.this.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 方法

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

在该方法中，先根据 URI 来获取服务名称，接着通过 LoadBalancerRequestFactory 再构建请求，之后通过 LoadBalancerClient 进行请求执行。

(ClientHttpResponse)this.loadBalancer.execute(serviceName, this.requestFactory.createRequest(request, body, execution));

### LoadBalancerRequestFactory ###

this.requestFactory.createRequest(request, body, execution)

调用 LoadBalancerRequestFactory 的 createRequest 方法创建 LoadBalancerRequest 对象，根据以上代码的追踪结果，它的参数 request 为创建的 InterceptingClientHttpRequest 对象，参数 execution 为创建的 InterceptingRequestExecution 对象。

public LoadBalancerRequest<ClientHttpResponse> createRequest(final HttpRequest request, final byte[] body, final ClientHttpRequestExecution execution) {
        return (instance) -> {
            HttpRequest serviceRequest = new ServiceRequestWrapper(request, instance, this.loadBalancer);
            LoadBalancerRequestTransformer transformer;
            if (this.transformers != null) {
                for(Iterator var6 = this.transformers.iterator(); var6.hasNext(); serviceRequest = transformer.transformRequest((HttpRequest)serviceRequest, instance)) {
                    transformer = (LoadBalancerRequestTransformer)var6.next();
                }
            }
    
            return execution.execute((HttpRequest)serviceRequest, body);
        };
    }

根据代码可以看到创建了 LoadBalancerRequest 的实现主要作用是对请求示例进行了包装及通过之前 Riboon 配置的 LoadBalancerRequestTransformer 对请求进行处理。

HttpRequest serviceRequest = new ServiceRequestWrapper(request, instance, this.loadBalancer);

LoadBalancerRequest 包装了什么？我们继续往下追踪

## 负载均衡（请求执行） ##

完成创建 LoadBalancerRequest 对象后，接着通过 LoadBalancerClient 开始执行请求，我们已经知道之前 Ribbon 配置创建 LoadBalancerClient 的实现类为 RibbonLoadBalancerClient。

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

在这里根据之前获取的服务名称 serviceId 来获取对应的负载均衡器 ILoadBalancer。

public ILoadBalancer getLoadBalancer(String name) {
        return (ILoadBalancer)this.getInstance(name, ILoadBalancer.class);
    }

通过上篇博文我们已经知道，对于每个 Ribbon Client 定义都有一个 RibbonClientSpecification 与之对应，这里在获取负载均衡器时，则先根据 RibbonClientSpecification 通过 SpringClientFactory 创建了 Ribbon Client 的应用上下文，这里为 AnnotationConfigApplicationContext 对象。

关于负载均衡器的创建初始化，其中的细节如 PingTask 定时任务的设置，我将在接下来的文章进行介绍学习。

在获取到负载均衡器 ILoadBalancer 后，进行服务实例的选择创建。

Server server = this.getServer(loadBalancer, hint);

protected Server getServer(ILoadBalancer loadBalancer, Object hint) {
        return loadBalancer == null ? null : loadBalancer.chooseServer(hint != null ? hint : "default");
    }

通过上篇文章我们已经知道 Spring 默认提供的负载均衡器为 ZoneAwareLoadBalancer，接着再根据我们配置的 IRule 进行服务地址的选择。具体的细节就不在这里展开解析了。

在选择创建了 Server 实例后，创建 RibbonLoadBalancerClient 的内部类 RibbonServer 实例，进行请求执行。

public <T> T execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest<T> request) throws IOException {
        Server server = null;
        if (serviceInstance instanceof RibbonLoadBalancerClient.RibbonServer) {
            server = ((RibbonLoadBalancerClient.RibbonServer)serviceInstance).getServer();
        }
    
        if (server == null) {
            throw new IllegalStateException("No instances available for " + serviceId);
        } else {
            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 var8) {
                statsRecorder.recordStats(var8);
                throw var8;
            } catch (Exception var9) {
                statsRecorder.recordStats(var9);
                ReflectionUtils.rethrowRuntimeException(var9);
                return null;
            }
        }
    }

这里就调用了之前创建的 LoadBalancerRequest 实例的 apply 方法。

T returnVal = request.apply(serviceInstance);

这里就可以回答之前的问题，ServiceRequestWrapper 实例包装了什么，包装了原始请求 InterceptingClientHttpRequest，选择的服务实例 RibbonServer 及 Ribbon 的负载客户端 LoadBalancerClient。

接着调用 InterceptingRequestExecution 的 execute 方法，可以看到这里是一个递归调用的过程。在拦截器遍历执行完毕后，进入请求执行的最后阶段。

ClientHttpRequest delegate = InterceptingClientHttpRequest.this.requestFactory.createRequest(request.getURI(), method);

上面提到在创建 InterceptingClientHttpRequest 对象时，同时调用了父类的 getRequestFactory 方法，将返回的 SimpleClientHttpRequestFactory 对象作为构造参数传递。这里就是使用 SimpleClientHttpRequestFactory 创建 SimpleBufferingClientHttpRequest 对象，进而调用父类 AbstractClientHttpRequest 的 execute 模板方法，实现 executeInternal 方法的调用。

protected ClientHttpResponse executeInternal(HttpHeaders headers, byte[] bufferedOutput) throws IOException {
        addHeaders(this.connection, headers);
        if (this.getMethod() == HttpMethod.DELETE && bufferedOutput.length == 0) {
            this.connection.setDoOutput(false);
        }
    
        if (this.connection.getDoOutput() && this.outputStreaming) {
            this.connection.setFixedLengthStreamingMode(bufferedOutput.length);
        }
    
        this.connection.connect();
        if (this.connection.getDoOutput()) {
            FileCopyUtils.copy(bufferedOutput, this.connection.getOutputStream());
        } else {
            this.connection.getResponseCode();
        }
    
        return new SimpleClientHttpResponse(this.connection);
    }

最后返回 SimpleClientHttpResponse 实例最后请求响应。

至此，RestTemplate 的请求流程解析介绍完毕。

# 总结 #

RestTemplate 在请求过程中，通过不同的 RequetFactory 创建不同的 ClientHttpRequest 实例。其中通过 InterceptingClientHttpRequest 对普通请求进行拦截处理，通过 Ribbon 设置的 LoadBalancerInterceptor 拦截器，进入 Ribbon 对请求进行负载的处理流程，在这期间创建了 LoadBalancerRequest ，对 Ribbon 选择的 server 进行包装并根据配置的 LoadBalancerRequestTransformer 对请求进行处理，在拦截器处理完毕后，又通过 SimpleClientHttpRequestFactory 创建了 SimpleBufferingClientHttpRequest 实例，进行实际的请求执行，最后返回 SimpleClientHttpResponse 实例作为请求结果并进行处理。

并且在 Ribbon 进行选择 server 时创建了对应 Ribbon Client 的应用上下文，这也是为什么在第一次请求负载时速度较慢。

[Ribbon _Spring Cloud Ribbon]: https://blog.csdn.net/sinat_36553913/article/details/108298183
[Ribbon _ Eureka _ Ribbon]: https://blog.csdn.net/sinat_36553913/article/details/108028367
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM2NTUzOTEz_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/d0050693c01d47dc8f819e99791a3afe.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM2NTUzOTEz_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/106622ef3c6c4421bdd918fab451f163.png