💖 Spring家族及微服务系列文章

✨【微服务】SpringCloud中Ribbon的WeightedResponseTimeRule策略

✨【微服务】SpringCloud中Ribbon的轮询(RoundRobinRule)与重试(RetryRule)策略

✨【微服务】SpringCloud中Ribbon集成Eureka实现负载均衡

✨【微服务】SpringCloud轮询拉取注册表及服务发现源码解析

✨【微服务】SpringCloud微服务续约源码解析

✨【微服务】SpringCloud微服务注册源码解析

✨【微服务】Nacos2.x服务发现？RPC调用？重试机制？

✨【微服务】Nacos通知客户端服务变更以及重试机制

✨【微服务】Nacos服务发现源码分析

✨【微服务】SpringBoot监听器机制以及在Nacos中的应用

✨【微服务】Nacos服务端完成微服务注册以及健康检查流程

✨【微服务】Nacos客户端微服务注册原理流程

✨【微服务】SpringCloud中使用Ribbon实现负载均衡的原理

✨【微服务】SpringBoot启动流程注册FeignClient

✨【微服务】SpringBoot启动流程初始化OpenFeign的入口

✨Spring Bean的生命周期

✨Spring事务原理

✨SpringBoot自动装配原理机制及过程

✨SpringBoot获取处理器流程

✨SpringBoot中处理器映射关系注册流程

✨Spring5.x中Bean初始化流程

✨Spring中Bean定义的注册流程

✨Spring的处理器映射器与适配器的架构设计

✨SpringMVC执行流程图解及源码

💖 Spring家族及微服务系列文章

一、前言

二、OpenFeign请求处理流程

1、入口

1.1、异常后决定是否重试

2、创建rest请求模板

2.1、SpringMvcContract转换处理

2.2、使用所提供的变量值替换解析所有表达式

3、执行请求并将返回结果解码

4、处理拦截器的apply()方法

5、LoadBalancerFeignClient负载均衡处理feign请求

5.1、根据服务名clientName获取FeignLoadBalancer

5.2、feign如何实现负载均衡

一、前言

前面的文章已经分析了Ribbon实现负载均衡的原理以及一些负载均衡策略，里面涉及到一些算法以及设计思想还是值得我们去汲取其中的精华的。在面试的情况下，面试官也喜欢面试这些问题，所以有必要对其中的原理有所深入理解。本篇文章我们来探究一下，OpenFeign是如何实现负载均衡的？重试机制是怎么实现的、线程睡眠？如何创建Rest请求模板，使用所提供的变量值替换解析所有表达式？拦截器？LoadBalancerFeignClient负载均衡处理feign请求，底层实现是什么、如何负载均衡选取服务、最终的请求URI是怎样的、负载平衡器的作用、记录响应统计信息(以便计算权重)？带着这些问题，我们往下读，找出我们的答案吧：

二、OpenFeign请求处理流程

在讲解下面的流程前先说明下，笔者在SpringCloud Alibaba项目中整合了sentinel了，所以debugger进来时会首先进到的是Sentinel的类进行处理，然后才进到下面流程的，后面时间允许得话就补充一篇文章详细梳理其中的细节。

1、入口

  @Overridepublic Object invoke(Object[] argv) throws Throwable {// 创建请求模板RequestTemplate template = buildTemplateFromArgs.create(argv);// 将参数封装为OptionsOptions options = findOptions(argv);Retryer retryer = this.retryer.clone();// 如果有必要循环重试while (true) {try {// 执行并将返回结果解码return executeAndDecode(template, options);} catch (RetryableException e) {try {// 重试retryer.continueOrPropagate(e);} catch (RetryableException th) {// 重试过程中抛异常，则抛出异常结束请求Throwable cause = th.getCause();if (propagationPolicy == UNWRAP && cause != null) {throw cause;} else {throw th;}}if (logLevel != Logger.Level.NONE) {logger.logRetry(metadata.configKey(), logLevel);}// 继续请求continue;}}}

主要逻辑：

创建请求模板，下面2大点分析
将参数封装为Options，克隆Retryer
如果有必要循环以重试：1）执行并将返回结果解码；2）如果执行过程中抛出重试异常，则走第一个catch逻辑；3）决定是否重试，下面分析；决定是否重试过程中，抛出了异常即不再重试，异常外抛结束该调用；否则continue继续

1.1、异常后决定是否重试

    public void continueOrPropagate(RetryableException e) {// 重试次数大于maxAttempts(默认5次)抛异常结束请求if (attempt++ >= maxAttempts) {throw e;}// 睡眠时长long interval;if (e.retryAfter() != null) {interval = e.retryAfter().getTime() - currentTimeMillis();if (interval > maxPeriod) {// 不能大于最大期限interval = maxPeriod;}if (interval < 0) {return;}} else {interval = nextMaxInterval();}try {// 让当前线程睡眠一下Thread.sleep(interval);} catch (InterruptedException ignored) {// 发生中断异常，中断当前线程Thread.currentThread().interrupt();// 抛异常结束请求throw e;}sleptForMillis += interval;}

主要逻辑：

如果重试次数大于maxAttempts(默认5次)抛异常结束请求
计算睡眠时长，interval < 0立即请求
否则让当前线程睡眠一下，发生中断异常，中断当前线程，异常结束请求

2、创建rest请求模板

    @Overridepublic RequestTemplate create(Object[] argv) {// 获取rest请求模板RequestTemplate mutable = RequestTemplate.from(metadata.template());mutable.feignTarget(target);if (metadata.urlIndex() != null) {int urlIndex = metadata.urlIndex();checkArgument(argv[urlIndex] != null, "URI parameter %s was null", urlIndex);mutable.target(String.valueOf(argv[urlIndex]));}Map<String, Object> varBuilder = new LinkedHashMap<>();// indexToName为Map<Integer, Collection<String>>类型// 遍历indexToNamefor (Entry<Integer, Collection<String>> entry : metadata.indexToName().entrySet()) {// 如：0int i = entry.getKey();// 参数值，如：CeaMObject value = argv[entry.getKey()];if (value != null) { // Null values are skipped.跳过空值。// 如果indexToExpander包含该keyif (indexToExpander.containsKey(i)) {// SpringMvcContract会对value进行转换处理value = expandElements(indexToExpander.get(i), value);}// 参数名-参数值for (String name : entry.getValue()) {varBuilder.put(name, value);}}}// 将参数填充到rest模板RequestTemplate template = resolve(argv, mutable, varBuilder);if (metadata.queryMapIndex() != null) {// add query map parameters after initial resolve so that they take// precedence over any predefined values// 在初始解析之后添加查询映射参数，使它们优先于任何预定义的值Object value = argv[metadata.queryMapIndex()];Map<String, Object> queryMap = toQueryMap(value);template = addQueryMapQueryParameters(queryMap, template);}if (metadata.headerMapIndex() != null) {// add header map parameters for a resolution of the user pojo object// 为用户 pojo 对象的解析添加头映射参数Object value = argv[metadata.headerMapIndex()];Map<String, Object> headerMap = toQueryMap(value);template = addHeaderMapHeaders(headerMap, template);}// 经过上面处理，返回完整的请求模板return template;}

图2-1

图2-2

主要逻辑：

初始化rest请求模板，初始化数据如图2-1
遍历indexToName，为Map<Integer, Collection<String>>类型：获取key以及参数值，如果参数值不为空、indexToExpander包含该key，则SpringMvcContract会对value进行转换处理；遍历参数名，put到varBuilder以便进一步解析成请求模板
将参数填充到rest模板，解析成相当完整的请求模板，下面分析

图2-3 ndexToName数据样例

2.1、SpringMvcContract转换处理

	private static class ConvertingExpanderFactory {private final ConversionService conversionService;ConvertingExpanderFactory(ConversionService conversionService) {this.conversionService = conversionService;}Param.Expander getExpander(TypeDescriptor typeDescriptor) {// 转化return value -> {Object converted = conversionService.convert(value, typeDescriptor,STRING_TYPE_DESCRIPTOR);return (String) converted;};}}

Feign不具备转换SpringMVC功能，于是OpenFeign提供了SpringMvcContract，底层委托Spring去做转换处理(提供各种类型转换器)。

2.2、使用所提供的变量值替换解析所有表达式

  public RequestTemplate resolve(Map<String, ?> variables) {StringBuilder uri = new StringBuilder();/* create a new template form this one, but explicitly* 创建一个新的模板这一个，但显式*/RequestTemplate resolved = RequestTemplate.from(this);// uriTemplate模板空则新建if (this.uriTemplate == null) {/* create a new uri template using the default root */this.uriTemplate = UriTemplate.create("", !this.decodeSlash, this.charset);}// 解析，不为空追加到uriString expanded = this.uriTemplate.expand(variables);if (expanded != null) {uri.append(expanded);}/** for simplicity, combine the queries into the uri and use the resulting uri to seed the* resolved template.* 为了简单起见，将查询组合到 uri 中，并使用生成的 uri 为解析的模板播种。*/if (!this.queries.isEmpty()) {/** since we only want to keep resolved query values, reset any queries on the resolved copy* 因为我们只想保留已解析的查询值，所以请重置已解析副本上的任何查询*/resolved.queries(Collections.emptyMap());StringBuilder query = new StringBuilder();Iterator<QueryTemplate> queryTemplates = this.queries.values().iterator();while (queryTemplates.hasNext()) {QueryTemplate queryTemplate = queryTemplates.next();String queryExpanded = queryTemplate.expand(variables);if (Util.isNotBlank(queryExpanded)) {query.append(queryExpanded);if (queryTemplates.hasNext()) {query.append("&");}}}String queryString = query.toString();if (!queryString.isEmpty()) {Matcher queryMatcher = QUERY_STRING_PATTERN.matcher(uri);if (queryMatcher.find()) {/* the uri already has a query, so any additional queries should be appended*  Uri 已经有了一个查询，所以任何附加的查询都应该被追加*/uri.append("&");} else {uri.append("?");}uri.append(queryString);}}/* add the uri to result将 uri 添加到 result 中 */resolved.uri(uri.toString());/* headers */if (!this.headers.isEmpty()) {/** same as the query string, we only want to keep resolved values, so clear the header map on* the resolved instance*/resolved.headers(Collections.emptyMap());for (HeaderTemplate headerTemplate : this.headers.values()) {/* resolve the header */String header = headerTemplate.expand(variables);if (!header.isEmpty()) {/* append the header as a new literal as the value has already been expanded.* 作为新的文字追加标题，因为值已经展开。 */resolved.appendHeader(headerTemplate.getName(), Collections.singletonList(header), true);}}}if (this.bodyTemplate != null) {resolved.body(this.bodyTemplate.expand(variables));}/* mark the new template resolved标记已解析的新模板 */resolved.resolved = true;return resolved;}

图2-5

该方法主要逻辑就是使用所提供的变量值替换解析所有表达式：如图2-5中初始uriTemplate为“/admin/feign/feign”，变量值variables：{"str": "CeaM"}。经过解析追加后，得到的uri为“/admin/feign/feign/?str=CeaM”，如果有多个查询参数会通过&符号拼接，平时用postman或者谷歌浏览器按F12发送请求得话我们都比较熟悉了吧。

图2-6

3、执行请求并将返回结果解码

    Object executeAndDecode(RequestTemplate template, Options options) throws Throwable {Request request = this.targetRequest(template);if (this.logLevel != Level.NONE) {this.logger.logRequest(this.metadata.configKey(), this.logLevel, request);}long start = System.nanoTime();Response response;try {response = this.client.execute(request, options);response = response.toBuilder().request(request).requestTemplate(template).build();} catch (IOException var13) {if (this.logLevel != Level.NONE) {this.logger.logIOException(this.metadata.configKey(), this.logLevel, var13, this.elapsedTime(start));}throw FeignException.errorExecuting(request, var13);}long elapsedTime = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);if (this.decoder != null) {return this.decoder.decode(response, this.metadata.returnType());} else {CompletableFuture<Object> resultFuture = new CompletableFuture();this.asyncResponseHandler.handleResponse(resultFuture, this.metadata.configKey(), response, this.metadata.returnType(), elapsedTime);try {if (!resultFuture.isDone()) {throw new IllegalStateException("Response handling not done");} else {return resultFuture.join();}} catch (CompletionException var12) {Throwable cause = var12.getCause();if (cause != null) {throw cause;} else {throw var12;}}}}

主要逻辑：

处理拦截器的apply()方法，下面分析
通过LoadBalancerFeignClient负载均衡处理feign请求，下面详细分析
将返回结果处理，如果指定解码器则调用解码器解码处理；否则异步处理返回结果

4、处理拦截器的apply()方法

  Request targetRequest(RequestTemplate template) {for (RequestInterceptor interceptor : requestInterceptors) {interceptor.apply(template);}return target.apply(template);}

遍历拦截器，然后调用拦截器的apply方法，主要是对RequestTemplate进行处理。注意拦截器是Feign的，我们也可以自己定义，在日常开发中也是会用到的。然后根据模板template调用target.apply()生成请求对象request。

5、LoadBalancerFeignClient负载均衡处理feign请求

顾名思义它使FeignClient具备负载均衡功能，它是OpenFeign项目里面的一个重要组件。

	@Overridepublic Response execute(Request request, Request.Options options) throws IOException {try {// 将url封装为URI以便获取服务名，http://ceam-admin/admin/feign/feign?str=CeaMURI asUri = URI.create(request.url());// 从URI获取服务名，如：ceam-adminString clientName = asUri.getHost();// 移除服务名，如：http:///admin/feign/feign?str=CeaMURI uriWithoutHost = cleanUrl(request.url(), clientName);// 将请求数据封装为ribbonRequestFeignLoadBalancer.RibbonRequest ribbonRequest = new FeignLoadBalancer.RibbonRequest(this.delegate, request, uriWithoutHost);// 根据options以及服务名获取请求配置，没有指定则使用默认的IClientConfig requestConfig = getClientConfig(options, clientName);// 根据服务名从缓存获取负载均衡器，没有则新建// 负载均衡执行return lbClient(clientName).executeWithLoadBalancer(ribbonRequest, requestConfig).toResponse();}catch (ClientException e) {IOException io = findIOException(e);if (io != null) {throw io;}throw new RuntimeException(e);}}

主要逻辑：

将url封装为URI以便获取服务名，http://ceam-admin/admin/feign/feign?str=CeaM
从URI获取服务名，如：ceam-admin
移除服务名，如：http:///admin/feign/feign?str=CeaM
将请求数据封装为ribbonRequest，RibbonRequest是FeignLoadBalancer的静态内部类，实现了ClientRequest接口。
根据options以及服务名，如果是DEFAULT_OPTIONS则从SpringClientFactory(SpringCloud上下文)获取请求配置IClientConfig；否则封装为FeignOptionsClientConfig来得到IClientConfig。
根据服务名，从CachingSpringLoadBalancerFactory获取FeignLoadBalancer

5.1、根据服务名clientName获取FeignLoadBalancer

	public FeignLoadBalancer create(String clientName) {// 首先根据服务名从缓存获取，如果有直接返回；否则新建缓存起来FeignLoadBalancer client = this.cache.get(clientName);if (client != null) {// 缓存有直接返回return client;}// 根据服务名从SpringClientFactory获取配置IClientConfig config = this.factory.getClientConfig(clientName);// 根据服务名从SpringClientFactory获取负载均衡器ILoadBalancer lb = this.factory.getLoadBalancer(clientName);// 根据服务名从SpringClientFactory服务器内部检查器ServerIntrospector serverIntrospector = this.factory.getInstance(clientName,ServerIntrospector.class);// 封装为FeignLoadBalancerclient = this.loadBalancedRetryFactory != null? new RetryableFeignLoadBalancer(lb, config, serverIntrospector,this.loadBalancedRetryFactory): new FeignLoadBalancer(lb, config, serverIntrospector);// 缓存起来提高系统性能，不需要多次创建造成资源浪费this.cache.put(clientName, client);return client;}

图5-1

主要逻辑：

首先根据服务名从缓存获取，如果有直接返回；否则新建缓存起来，第一次的话新建
根据服务名从SpringClientFactory获取配置IClientConfig、获取负载均衡器ILoadBalancer、获取服务器内部检查器ServerIntrospector。由feign执行流程到图5-1可见，IClientConfig为DefaultClientConfigImpl、ILoadBalancer为ZoneAwareLoadBalancer、ServerIntrospector为NacosServerIntrospector。同时你打开config中可以看到一些配置，默认的连接超时时间connectTimeout为1秒、执行操作超时时间readTimeout也为1秒。
将配置IClientConfig、负载均衡器ILoadBalancer、服务器内部检查器ServerIntrospector封装到FeignLoadBalancer
缓存起来提高系统性能，不需要多次创建造成资源浪费

5.2、feign如何实现负载均衡

feign并没有自己实现负载均衡功能，从上面也看到了FeignLoadBalancer封装了ILoadBalancer，可见它是委托ILoadBalancer实现负载均衡的。从前面的文章我们也已经介绍了Ribbon实现负载均衡的原理，而ILoadBalancer是Ribbon项目里面的一个核心组件接口，所以我们还是会用到之前的一些原理的。而在讲解之前我们先看看FeignLoadBalancer的父子关系图：

从关系图可见只有 FeignLoadBalancer是OpenFeign项目里面的，其它的都是Ribbon项目的。我们点击一下executeWithLoadBalancer()方法，来到了FeignLoadBalancer的抽象父类AbstractLoadBalancerAwareClient。

    public T executeWithLoadBalancer(final S request, final IClientConfig requestConfig) throws ClientException {LoadBalancerCommand<T> command = buildLoadBalancerCommand(request, requestConfig);try {// ServerOperation在这里创建了匿名内部类，那么call()方法是由匿名内部类调用的return command.submit(new ServerOperation<T>() {@Overridepublic Observable<T> call(Server server) {// 拼接最终的URIURI finalUri = reconstructURIWithServer(server, request.getUri());// 替换为最终的URIS requestForServer = (S) request.replaceUri(finalUri);try {// 处理请求return Observable.just(AbstractLoadBalancerAwareClient.this.execute(requestForServer, requestConfig));} catch (Exception e) {return Observable.error(e);}}}).toBlocking().single();} catch (Exception e) {Throwable t = e.getCause();if (t instanceof ClientException) {throw (ClientException) t;} else {throw new ClientException(e);}}}

主要逻辑：

将请求数据以及配置数据封装到LoadBalancerCommand中，LoadBalancerCommand一个命令，用于从负载平衡器执行中生成 Observer。负载平衡器负责以下工作：1）选择一个服务器；2）调用 Server onSubscribe 调用方法；3）如果有，调用ExecutionListener；4）异常时重试，由 RetryHandler 控制；5）向 LoadBalancerStats 提供反馈，即记录请求响应的结果，以便计算权重等等，在上一篇分析权重计算时就用到了其中的统计数据。这些功能在5.3体现。
command.submit()下面分析。其中的参数为ServerOperation类型，ServerOperation在这里创建了匿名内部类，那么call()方法是由匿名内部类调用的。call()方法逻辑就是根据负载均衡选取的服务以及前面移除服务名的uri拼接成最终的请求URI，然后替换为最终请求URI，接着就是真正执行请求的逻辑了。
command.submit()后就是command.toBlocking() .single()，即阻塞同步执行，等待返回结果。

5.3、负载均衡调用

public Observable<T> submit(final ServerOperation<T> operation) {final ExecutionInfoContext context = new ExecutionInfoContext();if (listenerInvoker != null) {try {listenerInvoker.onExecutionStart();} catch (AbortExecutionException e) {return Observable.error(e);}}// 在同一个服务器上最大重试次数final int maxRetrysSame = retryHandler.getMaxRetriesOnSameServer();// 在下一个服务器上最大重试次数final int maxRetrysNext = retryHandler.getMaxRetriesOnNextServer();// Use the load balancer使用负载平衡器Observable<T> o = (server == null ? selectServer() : Observable.just(server)).concatMap(new Func1<Server, Observable<T>>() {@Override// Called for each server being selected为选定的每个服务器调用public Observable<T> call(Server server) {context.setServer(server);final ServerStats stats = loadBalancerContext.getServerStats(server);// Called for each attempt and retry调用每次尝试和重试Observable<T> o = Observable.just(server).concatMap(new Func1<Server, Observable<T>>() {@Overridepublic Observable<T> call(final Server server) {context.incAttemptCount();loadBalancerContext.noteOpenConnection(stats);if (listenerInvoker != null) {try {listenerInvoker.onStartWithServer(context.toExecutionInfo());} catch (AbortExecutionException e) {return Observable.error(e);}}final Stopwatch tracer = loadBalancerContext.getExecuteTracer().start();// ServerOperation匿名内部类调用callreturn operation.call(server).doOnEach(new Observer<T>() {private T entity;@Overridepublic void onCompleted() {recordStats(tracer, stats, entity, null);// TODO: What to do if onNext or onError are never called?}@Overridepublic void onError(Throwable e) {recordStats(tracer, stats, null, e);logger.debug("Got error {} when executed on server {}", e, server);if (listenerInvoker != null) {listenerInvoker.onExceptionWithServer(e, context.toExecutionInfo());}}@Overridepublic void onNext(T entity) {this.entity = entity;if (listenerInvoker != null) {listenerInvoker.onExecutionSuccess(entity, context.toExecutionInfo());}}                            private void recordStats(Stopwatch tracer, ServerStats stats, Object entity, Throwable exception) {tracer.stop();// 记录请求完成的信息，用于统计，如：权重等loadBalancerContext.noteRequestCompletion(stats, entity, exception,tracer.getDuration(TimeUnit.MILLISECONDS), retryHandler);}});}});// 重试if (maxRetrysSame > 0) // retryPolicy()中返回Func2匿名内部类，// 该类的call方法会根据最大重试次数判断是否重试o = o.retry(retryPolicy(maxRetrysSame, true));return o;}});if (maxRetrysNext > 0 && server == null) o = o.retry(retryPolicy(maxRetrysNext, false));return o.onErrorResumeNext(new Func1<Throwable, Observable<T>>() {@Overridepublic Observable<T> call(Throwable e) {if (context.getAttemptCount() > 0) {if (maxRetrysNext > 0 && context.getServerAttemptCount() == (maxRetrysNext + 1)) {e = new ClientException(ClientException.ErrorType.NUMBEROF_RETRIES_NEXTSERVER_EXCEEDED,"Number of retries on next server exceeded max " + maxRetrysNext+ " retries, while making a call for: " + context.getServer(), e);}else if (maxRetrysSame > 0 && context.getAttemptCount() == (maxRetrysSame + 1)) {e = new ClientException(ClientException.ErrorType.NUMBEROF_RETRIES_EXEEDED,"Number of retries exceeded max " + maxRetrysSame+ " retries, while making a call for: " + context.getServer(), e);}}if (listenerInvoker != null) {listenerInvoker.onExecutionFailed(e, context.toFinalExecutionInfo());}return Observable.error(e);}});}

主要逻辑：

如果监听器调用不为空，则执行监听器调用
使用负载均衡器。在使用前，第一次server为空，从selectServer()负载均衡出一个服务，下面分析。Observable的concatMap()方法接收Func1类型参数，该参数是Func1的匿名内部类引用，里面涉及一个call()调用：1）如果有监听器则调用监听器逻辑；2）ServerOperation匿名内部类调用call()，即上面5.2说的call()；3）不管响应正常还是出错，都记录响应结果以便在计算权重等情况下需用到这部分数据；4）retryPolicy()中返回Func2匿名内部类，该类的call方法会根据最大重试次数判断是否重试

5.4、负载均衡选取一个服务器

    private Observable<Server> selectServer() {return Observable.create(new OnSubscribe<Server>() {@Overridepublic void call(Subscriber<? super Server> next) {try {Server server = loadBalancerContext.getServerFromLoadBalancer(loadBalancerURI, loadBalancerKey);   next.onNext(server);next.onCompleted();} catch (Exception e) {next.onError(e);}}});}

selectServer()会负载均衡选取一个服务器，是由LoadBalancerContext#getServerFromLoadBalancer()方法获取到ILoadBalancer。而ILoadBalancer根据上图以及上面的流程也可知道是Ribbon的ZoneAwareLoadBalancer。那么ILoadBalancer#chooseServer()，实际上就是ZoneAwareLoadBalancer#chooseServer()方法调用，根据前面文章Ribbon实现负载均衡原理，我们就理解其中的意思了。

再根据前面移除服务名的uri：http:///admin/feign/feign?str=CeaM以及负载均衡选取的服务server：192.168.88.1:9705，经过ServerOperation匿名内部类调用call()过程中，拼接最终的请求URI为：http://192.168.88.1:9705/admin/feign/feign?str=CeaM，最后根据它以及请求配置(如上面说的：连接超时时间1秒、处理业务超时时间1秒)执行execute完成请求，底层用的是自己封装的http请求工具。