跳到主要内容

08、SpringCloud Zuul 服务重试:源码解析

我们一般部署服务的时候,都会部署一个网关服务,内部所有的其他微服务的调用,都将通过网关路由过去,不对外直接暴露,对外只暴露网关服务。而且一般内部服务会部署多个实例,zuul集成了ribbon,会自动负载均衡的方式去调用内部服务。

当内部服务滚动重启的时候,通过网关的一个请求刚好路由到重启的那个实例的话,因为默认没有开启zuul的请求重试策略,该请求将会报错,其实理想的方式可以通过重试路由到另外一个活动的服务实例上去。

要开启zuul网关请求重试,首先需要添加spring-retry依赖:

<dependency>
    <groupId>org.springframework.retry</groupId>
    <artifactId>spring-retry</artifactId>
</dependency>

然后配置:

zuul.retryable=true

这样,所有路由都将会进行重试。(此属性默认是false,所以不会重试)

有时候我们不希望所有路由都有重试机制,我们可以配置指定路由进行重试:

zuul.routes.<routename>.retryable=true

这里的routename默认情况下就是你的服务名(我们可以通过管理端点/routes看到都有哪些路由

也可以查看更详细的路由信息:/routes?format=details,

端点实现类:org.springframework.cloud.netflix.zuul.RoutesMvcEndpoint)。

例如我有一个rcmd-service-data的服务,我可以这样配置:

zuul.retryable=false
zuul.routes.rcmd-service-data.retryable=true

这样,就只有rcmd-service-data这个服务开启了重试机制。我们通过/routes?format=details端点也可以看到:

 

我们知道zuul请求也是通过Ribbon负载均衡客户端去调用其他服务的,所以我们的重试策略也是在具体的ribbon配置中指定:

rcmd-service-data:
  ribbon:
    # Max number of retries on the same server (excluding the first try)
    MaxAutoRetries: 1 
    # Max number of next servers to retry (excluding the first server)
    MaxAutoRetriesNextServer: 2 #当允许在其他服务器上重试的时候,会调用IRule.choose选择可用服务实例中的其他一台服务实例进行调用
    # Whether all operations can be retried for this client
    OkToRetryOnAllOperations: true  #默认为false,则只允许GET请求被重试
    ReadTimeout: 5000
    ConnectTimeout: 2000

重试的时候还有补偿策略,例如重试时间间隔

默认是没有间隔:org.springframework.retry.backoff.NoBackOffPolicy

我们可以实现自己的补偿策略,也可以用内部实现的一些补偿策略(需要定义一个bean),如指数级的补偿策略(1秒,2秒,4秒类似这种指数级睡眠间隔增长,不超过10秒):

@Configuration
public class MyConfiguration {
    @Bean
    LoadBalancedBackOffPolicyFactory backOffPolciyFactory() {
        return new LoadBalancedBackOffPolicyFactory() {
            @Override
            public BackOffPolicy createBackOffPolicy(String service) {
                return new ExponentialBackOffPolicy();
            }
        };
    }
}

也可以正对某些响应状态码进行重试(当调用rcmd-service-data返回404,502的时候,进行重试,其他状态码不重试):

rcmd-service-data:
  ribbon:
    retryableStatusCodes: 404,502

以上差不多就是网关重试相关的能够配置的点了.

下面从源码层面看看重试的实现。

首先我们需要大致有个概念就是,zuul网关是通过ribbon负载均衡客户端来调用内部服务的,然后ribbon客户端默认是用HttpClient来发起http请求调用(当然还可以通过配置使用okhttp或者ribbon自带的RestClient),在org.springframework.cloud.netflix.ribbon.apache.HttpClientRibbonConfiguration中进行自动装配的:

@Bean
	@ConditionalOnMissingBean(AbstractLoadBalancerAwareClient.class)
	@ConditionalOnClass(name = "org.springframework.retry.support.RetryTemplate")
	public RetryableRibbonLoadBalancingHttpClient retryableRibbonLoadBalancingHttpClient(
		IClientConfig config, ServerIntrospector serverIntrospector,
		ILoadBalancer loadBalancer, RetryHandler retryHandler,
		LoadBalancedRetryPolicyFactory loadBalancedRetryPolicyFactory, CloseableHttpClient httpClient,
		LoadBalancedBackOffPolicyFactory loadBalancedBackOffPolicyFactory,
		LoadBalancedRetryListenerFactory loadBalancedRetryListenerFactory) {
		RetryableRibbonLoadBalancingHttpClient client = new RetryableRibbonLoadBalancingHttpClient(
			httpClient, config, serverIntrospector, loadBalancedRetryPolicyFactory,
			loadBalancedBackOffPolicyFactory, loadBalancedRetryListenerFactory);
		client.setLoadBalancer(loadBalancer);
		client.setRetryHandler(retryHandler);
		Monitors.registerObject("Client_" + this.name, client);
		return client;
	}

ribbon内部优惠构造hystrix command来执行请求

所以我们常说ribbon的超时时间设置要大于等于hystrix 超时时间,不然导致command还没执行完,ribbon却超时了

由HttpClientRibbonCommand.run方法发起请求

org.springframework.cloud.netflix.zuul.filters.route.support.AbstractRibbonCommand#run

@Override
protected ClientHttpResponse run() throws Exception {
    final RequestContext context = RequestContext.getCurrentContext();

    RQ request = createRequest();
    RS response;
    
    boolean retryableClient = this.client instanceof AbstractLoadBalancingClient
            && ((AbstractLoadBalancingClient)this.client).isClientRetryable((ContextAwareRequest)request);
    
    if (retryableClient) {
        response = this.client.execute(request, config); //此处的client就是
RetryableRibbonLoadBalancingHttpClient
} else {
    response = this.client.executeWithLoadBalancer(request, config);
}
context.set("ribbonResponse", response); // Explicitly close the HttpResponse if the Hystrix command timed out to// release the underlying HTTP connection held by the response.//if (this.isResponseTimedOut()) {if (response != null) {response.close();}}return new RibbonHttpResponse(response);}

如果可以重试的话,调用的是RetryableRibbonLoadBalancingHttpClient.execute方法:

@Override
	public RibbonApacheHttpResponse execute(final RibbonApacheHttpRequest request, final IClientConfig configOverride) throws Exception {
		final RequestConfig.Builder builder = RequestConfig.custom();
		IClientConfig config = configOverride != null ? configOverride : this.config;
		builder.setConnectTimeout(config.get(
				CommonClientConfigKey.ConnectTimeout, this.connectTimeout));
		builder.setSocketTimeout(config.get(
				CommonClientConfigKey.ReadTimeout, this.readTimeout));
		builder.setRedirectsEnabled(config.get(
				CommonClientConfigKey.FollowRedirects, this.followRedirects));

		final RequestConfig requestConfig = builder.build();
		final LoadBalancedRetryPolicy retryPolicy = loadBalancedRetryPolicyFactory.create(this.getClientName(), this);
		RetryCallback<RibbonApacheHttpResponse, IOException> retryCallback = new RetryCallback<RibbonApacheHttpResponse, IOException>() {
			@Override
			public RibbonApacheHttpResponse doWithRetry(RetryContext context) throws IOException {
				//on retries the policy will choose the server and set it in the context
				//extract the server and update the request being made
				RibbonApacheHttpRequest newRequest = request;
				if(context instanceof LoadBalancedRetryContext) {
					ServiceInstance service = ((LoadBalancedRetryContext)context).getServiceInstance();
					if(service != null) {
						//Reconstruct the request URI using the host and port set in the retry context
						newRequest = newRequest.withNewUri(UriComponentsBuilder.newInstance().host(service.getHost())
								.scheme(service.getUri().getScheme()).userInfo(newRequest.getURI().getUserInfo())
								.port(service.getPort()).path(newRequest.getURI().getPath())
								.query(newRequest.getURI().getQuery()).fragment(newRequest.getURI().getFragment())
								.build().encode().toUri());
					}
				}
				newRequest = getSecureRequest(newRequest, configOverride);
				HttpUriRequest httpUriRequest = newRequest.toRequest(requestConfig);
				final HttpResponse httpResponse = RetryableRibbonLoadBalancingHttpClient.this.delegate.execute(httpUriRequest);
				if(retryPolicy.retryableStatusCode(httpResponse.getStatusLine().getStatusCode())) {
					throw new HttpClientStatusCodeException(RetryableRibbonLoadBalancingHttpClient.this.clientName,
							httpResponse, HttpClientUtils.createEntity(httpResponse), httpUriRequest.getURI());
				}
				return new RibbonApacheHttpResponse(httpResponse, httpUriRequest.getURI());
			}
		};
		RibbonRecoveryCallback<RibbonApacheHttpResponse, HttpResponse> recoveryCallback = new RibbonRecoveryCallback<RibbonApacheHttpResponse, HttpResponse>() {
			@Override
			protected RibbonApacheHttpResponse createResponse(HttpResponse response, URI uri) {
				return new RibbonApacheHttpResponse(response, uri);
			}
 		};
		return this.executeWithRetry(request, retryPolicy, retryCallback, recoveryCallback);
	}
	
	

进而又调用了自身的executeWithRetry方法:

private RibbonApacheHttpResponse executeWithRetry(RibbonApacheHttpRequest request, LoadBalancedRetryPolicy retryPolicy,
                                      RetryCallback<RibbonApacheHttpResponse, IOException> callback,
                                      RecoveryCallback<RibbonApacheHttpResponse> recoveryCallback) throws Exception {
   RetryTemplate retryTemplate = new RetryTemplate();
   boolean retryable = isRequestRetryable(request); //从org.springframework.cloud.netflix.zuul.filters.route.RibbonCommandContext中获取retryable配置
   retryTemplate.setRetryPolicy(retryPolicy == null || !retryable ? new NeverRetryPolicy()
         : new RetryPolicy(request, retryPolicy, this, this.getClientName()));
   BackOffPolicy backOffPolicy = loadBalancedBackOffPolicyFactory.createBackOffPolicy(this.getClientName());
   retryTemplate.setBackOffPolicy(backOffPolicy == null ? new NoBackOffPolicy() : backOffPolicy);
   RetryListener[] retryListeners = this.loadBalancedRetryListenerFactory.createRetryListeners(this.getClientName());
   if (retryListeners != null && retryListeners.length != 0) {
      retryTemplate.setListeners(retryListeners);
   }
   return retryTemplate.execute(callback, recoveryCallback);
}
private boolean isRequestRetryable(ContextAwareRequest request) {
   return request.getContext() == null ? true :
      BooleanUtils.toBooleanDefaultIfNull(request.getContext().getRetryable(), true);
}

executeWithRetry方法又会调回execute方法中定义的RetryCallback匿名实现类的doWithRetry方法:

RetryCallback<RibbonApacheHttpResponse, IOException> retryCallback = new RetryCallback<RibbonApacheHttpResponse, IOException>() {
    @Override
    public RibbonApacheHttpResponse doWithRetry(RetryContext context) throws IOException {
        //on retries the policy will choose the server and set it in the context
        //extract the server and update the request being made
        RibbonApacheHttpRequest newRequest = request;
        if(context instanceof LoadBalancedRetryContext) {
            ServiceInstance service = ((LoadBalancedRetryContext)context).getServiceInstance();//这个getServiceInstance在next server retry的时候会变
            if(service != null) {
                //Reconstruct the request URI using the host and port set in the retry context
                newRequest = newRequest.withNewUri(UriComponentsBuilder.newInstance().host(service.getHost())
                        .scheme(service.getUri().getScheme()).userInfo(newRequest.getURI().getUserInfo())
                        .port(service.getPort()).path(newRequest.getURI().getPath())
                        .query(newRequest.getURI().getQuery()).fragment(newRequest.getURI().getFragment())
                        .build().encode().toUri());
            }
        }
        newRequest = getSecureRequest(newRequest, configOverride);
        HttpUriRequest httpUriRequest = newRequest.toRequest(requestConfig);
        final HttpResponse httpResponse = RetryableRibbonLoadBalancingHttpClient.this.delegate.execute(httpUriRequest);
        if(retryPolicy.retryableStatusCode(httpResponse.getStatusLine().getStatusCode())) { //这里就是跟配置中配置的状态码进行匹配
            throw new HttpClientStatusCodeException(RetryableRibbonLoadBalancingHttpClient.this.clientName,
                    httpResponse, HttpClientUtils.createEntity(httpResponse), httpUriRequest.getURI());
        }
        return new RibbonApacheHttpResponse(httpResponse, httpUriRequest.getURI());
    }
};

我们需要缕一缕这部分代码逻辑,首先是这句:

boolean retryable = isRequestRetryable(request); //从org.springframework.cloud.netflix.zuul.filters.route.RibbonCommandContext中获取retryable配置

表名了从org.springframework.cloud.netflix.zuul.filters.route.RibbonCommandContext获取到的是否可以重试的配置,而

RibbonCommandContext又是从:

org.springframework.cloud.netflix.zuul.filters.route.RibbonRoutingFilter#buildCommandContext中创建出来的:

protected RibbonCommandContext buildCommandContext(RequestContext context) {
    HttpServletRequest request = context.getRequest();

    MultiValueMap<String, String> headers = this.helper
            .buildZuulRequestHeaders(request);
    MultiValueMap<String, String> params = this.helper
            .buildZuulRequestQueryParams(request);
    String verb = getVerb(request);
    InputStream requestEntity = getRequestBody(request);
    if (request.getContentLength() < 0 && !verb.equalsIgnoreCase("GET")) {
        context.setChunkedRequestBody();
    }

    String serviceId = (String) context.get(SERVICE_ID_KEY);
    Boolean retryable = (Boolean) context.get(RETRYABLE_KEY);
    Object loadBalancerKey = context.get(LOAD_BALANCER_KEY);

    String uri = this.helper.buildZuulRequestURI(request);

    // remove double slashes
    uri = uri.replace("//", "/");

    long contentLength = useServlet31 ? request.getContentLengthLong(): request.getContentLength();

    return new RibbonCommandContext(serviceId, verb, uri, retryable, headers, params,
            requestEntity, this.requestCustomizers, contentLength, loadBalancerKey);
}

这句:

Boolean retryable = (Boolean) context.get(RETRYABLE_KEY);

retryable又是从:

com.netflix.zuul.context.RequestContext中获取的,那么从RequestContext里面的这个key又是从哪设置进去的呢?我们进而搜索发现在:

org.springframework.cloud.netflix.zuul.filters.pre.PreDecorationFilter

这个zuul filter中设置的该值,run方法代码片段:

Route route = this.routeLocator.getMatchingRoute(requestURI);
if (route.getRetryable() != null) {
        ctx.put(RETRYABLE_KEY, route.getRetryable());
    }

此处的routeLocator是:org.springframework.cloud.netflix.zuul.filters.CompositeRouteLocator.

这里用了一种设计模式(组合模式),不过里面就一个locator:

org.springframework.cloud.netflix.zuul.filters.discovery.DiscoveryClientRouteLocator,继承自:

org.springframework.cloud.netflix.zuul.filters.SimpleRouteLocator

最终会调用到:

org.springframework.cloud.netflix.zuul.filters.SimpleRouteLocator#getRoute方法:

protected Route getRoute(ZuulRoute route, String path) {
    if (route == null) {
        return null;
    }
    if (log.isDebugEnabled()) {
        log.debug("route matched=" + route);
    }
    String targetPath = path;
    String prefix = this.properties.getPrefix();
    if(prefix.endsWith("/")) {
        prefix = prefix.substring(0, prefix.length() - 1);
    }
    if (path.startsWith(prefix + "/") && this.properties.isStripPrefix()) {
        targetPath = path.substring(prefix.length());
    }
    if (route.isStripPrefix()) {
        int index = route.getPath().indexOf("*") - 1;
        if (index > 0) {
            String routePrefix = route.getPath().substring(0, index);
            targetPath = targetPath.replaceFirst(routePrefix, "");
            prefix = prefix + routePrefix;
        }
    }
    Boolean retryable = this.properties.getRetryable();
    if (route.getRetryable() != null) {
        retryable = route.getRetryable();
    }
    return new Route(route.getId(), targetPath, route.getLocation(), prefix,
            retryable,
            route.isCustomSensitiveHeaders() ? route.getSensitiveHeaders() : null, 
            route.isStripPrefix());
}

看到这句:

Boolean retryable = this.properties.getRetryable();

最终取的就是本文开头的配置文件中的配置属性。

看到这里,我们终于把retryable属性的来龙去脉搞清楚了,下面说说怎么重试到另外一个实例上去的。

我们在看回:

org.springframework.cloud.netflix.ribbon.apache.RetryableRibbonLoadBalancingHttpClient#execute方法。

其中有一段代码:

RetryCallback<RibbonApacheHttpResponse, IOException> retryCallback = new RetryCallback<RibbonApacheHttpResponse, IOException>() {
   @Override
   public RibbonApacheHttpResponse doWithRetry(RetryContext context) throws IOException {
      //on retries the policy will choose the server and set it in the context
      //extract the server and update the request being made
      RibbonApacheHttpRequest newRequest = request;
      if(context instanceof LoadBalancedRetryContext) {
         ServiceInstance service = ((LoadBalancedRetryContext)context).getServiceInstance();
         if(service != null) {
            //Reconstruct the request URI using the host and port set in the retry context
            newRequest = newRequest.withNewUri(UriComponentsBuilder.newInstance().host(service.getHost())
                  .scheme(service.getUri().getScheme()).userInfo(newRequest.getURI().getUserInfo())
                  .port(service.getPort()).path(newRequest.getURI().getPath())
                  .query(newRequest.getURI().getQuery()).fragment(newRequest.getURI().getFragment())
                  .build().encode().toUri());
         }
      }
      newRequest = getSecureRequest(newRequest, configOverride);
      HttpUriRequest httpUriRequest = newRequest.toRequest(requestConfig);
      final HttpResponse httpResponse = RetryableRibbonLoadBalancingHttpClient.this.delegate.execute(httpUriRequest);
      if(retryPolicy.retryableStatusCode(httpResponse.getStatusLine().getStatusCode())) {
         throw new HttpClientStatusCodeException(RetryableRibbonLoadBalancingHttpClient.this.clientName,
               httpResponse, HttpClientUtils.createEntity(httpResponse), httpUriRequest.getURI());
      }
      return new RibbonApacheHttpResponse(httpResponse, httpUriRequest.getURI());
   }
};

前面已经提到过:

ServiceInstance service = ((LoadBalancedRetryContext)context).getServiceInstance();

这一行代码getServiceInstance可能会变,即跟你最开始的路由不同,比如我有A,B两个实例,最开始的时候确定路由到A实例上,当A挂了,再重试的时候可能这个ServiceInstance可能是 Instance B。既然有get那么肯定有set它的地方.前面我们说了要实现重试,必须引入spring-retry依赖,然后才会通过org.springframework.retry.support.RetryTemplate#doExecute来执行重试请求,我们看看这个方法的代码片段:

while (canRetry(retryPolicy, context) && !context.isExhaustedOnly()) {

   try {
      if (this.logger.isDebugEnabled()) {
         this.logger.debug("Retry: count=" + context.getRetryCount());
      }
      // Reset the last exception, so if we are successful
      // the close interceptors will not think we failed...
      lastException = null;
      return retryCallback.doWithRetry(context); //这里调用了RetryableRibbonLoadBalancingHttpClient#execute中定义的RetryCallback的doWithRetry方法
   }
   catch (Throwable e) {

      lastException = e;

      try {
         registerThrowable(retryPolicy, state, context, e);//在失败的时候执行的逻辑,这里是关键,里面有重新选择服务实例的逻辑
      }
      catch (Exception ex) {
         throw new TerminatedRetryException("Could not register throwable",
               ex);
      }
      finally {
         doOnErrorInterceptors(retryCallback, context, e);
      }

      if (canRetry(retryPolicy, context) && !context.isExhaustedOnly()) {
         try {
            backOffPolicy.backOff(backOffContext);
         }
         catch (BackOffInterruptedException ex) {
            lastException = e;
            // back off was prevented by another thread - fail the retry
            if (this.logger.isDebugEnabled()) {
               this.logger
                     .debug("Abort retry because interrupted: count="
                           + context.getRetryCount());
            }
            throw ex;
         }
      }

      if (this.logger.isDebugEnabled()) {
         this.logger.debug(
               "Checking for rethrow: count=" + context.getRetryCount());
      }

      if (shouldRethrow(retryPolicy, context, state)) {
         if (this.logger.isDebugEnabled()) {
            this.logger.debug("Rethrow in retry for policy: count="
                  + context.getRetryCount());
         }
         throw RetryTemplate.<E>wrapIfNecessary(e);
      }

   }

   /*
    * A stateful attempt that can retry may rethrow the exception before now,
    * but if we get this far in a stateful retry there's a reason for it,
    * like a circuit breaker or a rollback classifier.
    */
   if (state != null && context.hasAttribute(GLOBAL_STATE)) {
      break;
   }
}

这行代码是关键,在请求发生异常的时候catch住异常,然后会执行:

registerThrowable(retryPolicy, state, context, e)

其中的retryPolicy是在RetryableRibbonLoadBalancingHttpClient中定义的内部类:

org.springframework.cloud.netflix.ribbon.apache.RetryableRibbonLoadBalancingHttpClient.RetryPolicy,所以其实调用委托给了:

org.springframework.cloud.netflix.ribbon.RibbonLoadBalancedRetryPolicy#registerThrowable方法:

@Override
public void registerThrowable(LoadBalancedRetryContext context, Throwable throwable) {
   //if this is a circuit tripping exception then notify the load balancer
   if (lbContext.getRetryHandler().isCircuitTrippingException(throwable)) {
      updateServerInstanceStats(context);
   }
   
   //Check if we need to ask the load balancer for a new server.
   //Do this before we increment the counters because the first call to this method
   //is not a retry it is just an initial failure.
   if(!canRetrySameServer(context)  && canRetryNextServer(context)) {
      context.setServiceInstance(loadBalanceChooser.choose(serviceId));
   }
   //This method is called regardless of whether we are retrying or making the first request.
   //Since we do not count the initial request in the retry count we don't reset the counter
   //until we actually equal the same server count limit.  This will allow us to make the initial
   //request plus the right number of retries.
   if(sameServerCount >= lbContext.getRetryHandler().getMaxRetriesOnSameServer() && canRetry(context)) {
      //reset same server since we are moving to a new server
      sameServerCount = 0;
      nextServerCount++;
      if(!canRetryNextServer(context)) {
         context.setExhaustedOnly();
      }
   } else {
      sameServerCount++;
   }

}

这个类中就用到了最开始说到的MaxAutoRetries,MaxAutoRetriesNextServer,OkToRetryOnAllOperations这三个属性。

主要看这段:

if(!canRetrySameServer(context)  && canRetryNextServer(context)) {
  context.setServiceInstance(loadBalanceChooser.choose(serviceId));
}

当能在其他服务器上重试的时候,然后通过loadBalanceChooser.choose(serviceId)重新选择了另外一个服务实例。然后set进了context。

整个流程大致上就是这样,具体的实现细节可以自行深入查看一下以上提及的类的代码。

另外还有一个注意事项,此源码是基于Edgware.SR2版本,在Edgware.SR1的时候有BUG,在:

org.springframework.cloud.netflix.ribbon.apache.RetryableRibbonLoadBalancingHttpClient#execute中定义的匿名类RetryCallback中这行代码

newRequest = getSecureRequest(newRequest, configOverride);

在SR1版本的时候是:

newRequest = getSecureRequest(request, configOverride);

导致retry next server无效,每次都被设置回了最开始route的实例,我也是在debug的时候发现的,开始用的SR1版本,我说怎么retry老是无效。。。后面去翻了一下官方的issue,发现:

https://github.com/spring-cloud/spring-cloud-netflix/issues/2667

此BUG在spring-cloud-netfix 1.4.3修复,SR1用的是1.4.2,SR2用的是1.4.3.

小伙伴们发现有问题,请确认你的版本是否和我说的版本一致,根据需要升级一下spring cloud版本。。。

版权声明:「DDKK.COM 弟弟快看,程序员编程资料站」本站文章,版权归原作者所有