Spring Cloud升级之路 - Hoxton - 9. 针对网关非 Get 请求的重试

public class RetryGatewayFilterFactory
		extends AbstractGatewayFilterFactory {

	/**
	 * Retry iteration key.ServerWebExchange的某个Attribute的key
	 * 这个Attribute用来在每次调用的时候，+1，看是否超过了重试次数
	 */
	public static final String RETRY_ITERATION_KEY = "retry_iteration";

	public RetryGatewayFilterFactory() {
		super(RetryConfig.class);
	}

	@Override
	public GatewayFilter apply(RetryConfig retryConfig) {
	    //检验配置
		retryConfig.validate();
		Repeat statusCodeRepeat = null;
		//如果配置了可重试的HTTP响应状态码，则检查响应码是否可以重试
		if (!retryConfig.getStatuses().isEmpty() || !retryConfig.getSeries().isEmpty()) {
			Predicate> repeatPredicate = context -> {
				ServerWebExchange exchange = context.applicationContext();
				//检查是否超过了重试次数
				if (exceedsMaxIterations(exchange, retryConfig)) {
					return false;
				}

                //判断是否可以重试
				HttpStatus statusCode = exchange.getResponse().getStatusCode();

				boolean retryableStatusCode = retryConfig.getStatuses()
						.contains(statusCode);

				if (!retryableStatusCode && statusCode != null) { 
					// try the series
					retryableStatusCode = retryConfig.getSeries().stream()
							.anyMatch(series -> statusCode.series().equals(series));
				}

				final boolean finalRetryableStatusCode = retryableStatusCode;

                //判断是否是可以重试的HTTP方法
				HttpMethod httpMethod = exchange.getRequest().getMethod();
				boolean retryableMethod = retryConfig.getMethods().contains(httpMethod);
                //返回是否是可以重试的方法以及是否可以重试的HTTP响应状态码
				return retryableMethod && finalRetryableStatusCode;
			};

            //每次重试，都要重置路由，重新解析路由
			statusCodeRepeat = Repeat.onlyIf(repeatPredicate)
					.doOnRepeat(context -> reset(context.applicationContext()));
            
            //设置Backoff
			BackoffConfig backoff = retryConfig.getBackoff();
			if (backoff != null) {
				statusCodeRepeat = statusCodeRepeat.backoff(getBackoff(backoff));
			}
		}

		// TODO: support timeout, backoff, jitter, etc... in Builder

        //判断异常是否可以重试
		Retry exceptionRetry = null;
		if (!retryConfig.getExceptions().isEmpty()) {
			Predicate> retryContextPredicate = context -> {

				ServerWebExchange exchange = context.applicationContext();

				if (exceedsMaxIterations(exchange, retryConfig)) {
					return false;
				}

				Throwable exception = context.exception();
				for (Class retryableClass : retryConfig
						.getExceptions()) {
					if (retryableClass.isInstance(exception) || (exception != null
							&& retryableClass.isInstance(exception.getCause()))) {
						trace("exception or its cause is retryable %s, configured exceptions %s",
								() -> getExceptionNameWithCause(exception),
								retryConfig::getExceptions);

						HttpMethod httpMethod = exchange.getRequest().getMethod();
						boolean retryableMethod = retryConfig.getMethods()
								.contains(httpMethod);
						trace("retryableMethod: %b, httpMethod %s, configured methods %s",
								() -> retryableMethod, () -> httpMethod,
								retryConfig::getMethods);
						return retryableMethod;
					}
				}
				trace("exception or its cause is not retryable %s, configured exceptions %s",
						() -> getExceptionNameWithCause(exception),
						retryConfig::getExceptions);
				return false;
			};
			exceptionRetry = Retry.onlyIf(retryContextPredicate)
					.doOnRetry(context -> reset(context.applicationContext()))
					.retryMax(retryConfig.getRetries());
			BackoffConfig backoff = retryConfig.getBackoff();
			if (backoff != null) {
				exceptionRetry = exceptionRetry.backoff(getBackoff(backoff));
			}
		}

		GatewayFilter gatewayFilter = apply(retryConfig.getRouteId(), statusCodeRepeat,
				exceptionRetry);
		return new GatewayFilter() {
			@Override
			public Mono filter(ServerWebExchange exchange,
					GatewayFilterChain chain) {
				return gatewayFilter.filter(exchange, chain);
			}

			@Override
			public String toString() {
				return filterToStringCreator(RetryGatewayFilterFactory.this)
						.append("retries", retryConfig.getRetries())
						.append("series", retryConfig.getSeries())
						.append("statuses", retryConfig.getStatuses())
						.append("methods", retryConfig.getMethods())
						.append("exceptions", retryConfig.getExceptions()).toString();
			}
		};
	}

	private String getExceptionNameWithCause(Throwable exception) {
		if (exception != null) {
			StringBuilder builder = new StringBuilder(exception.getClass().getName());
			Throwable cause = exception.getCause();
			if (cause != null) {
				builder.append("{cause=").append(cause.getClass().getName()).append("}");
			}
			return builder.toString();
		}
		else {
			return "null";
		}
	}

	private Backoff getBackoff(BackoffConfig backoff) {
		return Backoff.exponential(backoff.firstBackoff, backoff.maxBackoff,
				backoff.factor, backoff.basedOnPreviousValue);
	}

	public boolean exceedsMaxIterations(ServerWebExchange exchange,
			RetryConfig retryConfig) {
		Integer iteration = exchange.getAttribute(RETRY_ITERATION_KEY);
		//是否超过了可重试次数
		boolean exceeds = iteration != null && iteration >= retryConfig.getRetries();
		return exceeds;
	}

	public void reset(ServerWebExchange exchange) {
		//这个方法主要是
		Set addedHeaders = exchange.getAttributeOrDefault(
				CLIENT_RESPONSE_HEADER_NAMES, Collections.emptySet());
		addedHeaders
				.forEach(header -> exchange.getResponse().getHeaders().remove(header));
		removeAlreadyRouted(exchange);
	}


	public GatewayFilter apply(String routeId, Repeat repeat,
			Retry retry) {
		if (routeId != null && getPublisher() != null) {
			// send an event to enable caching
			getPublisher().publishEvent(new EnableBodyCachingEvent(this, routeId));
		}
		return (exchange, chain) -> {
			trace("Entering retry-filter");

			// chain.filter returns a Mono
			Publisher publisher = chain.filter(exchange)
					// .log("retry-filter", Level.INFO)
					.doOnSuccessOrError((aVoid, throwable) -> {
						int iteration = exchange
								.getAttributeOrDefault(RETRY_ITERATION_KEY, -1);
						int newIteration = iteration + 1;
						trace("setting new iteration in attr %d", () -> newIteration);
						exchange.getAttributes().put(RETRY_ITERATION_KEY, newIteration);
					});

			if (retry != null) {
				// retryWhen returns a Mono
				// retry needs to go before repeat
				publisher = ((Mono) publisher)
						.retryWhen(retry.withApplicationContext(exchange));
			}
			if (repeat != null) {
				// repeatWhen returns a Flux
				// so this needs to be last and the variable a Publisher
				publisher = ((Mono) publisher)
						.repeatWhen(repeat.withApplicationContext(exchange));
			}

			return Mono.fromDirect(publisher);
		};
	}

	@SuppressWarnings("unchecked")
	public static class RetryConfig implements HasRouteId {
        //路由id
		private String routeId;
        //重试次数，不包括调用的第一次，默认为3，也就是可能会调用4次
		private int retries = 3;
        //针对哪些HTTP状态码重试，一个Series对应一组HttpStatus
		private List series = toList(Series.SERVER_ERROR);
        //针对哪些HTTP状态码重试，一个HttpStatus就是一个HTTP状态码
		private List statuses = new ArrayList<>();
        //针对哪些HTTP方法重试
		private List methods = toList(HttpMethod.GET);
        //针对的哪些异常重试
		private List> exceptions = toList(IOException.class,
				TimeoutException.class);
        //重试间隔策略
		private BackoffConfig backoff;
        
        public void validate() {
            //重试次数必须大于10
			Assert.isTrue(this.retries > 0, "retries must be greater than 0");
			//可重试的series，可重试的状态码还有可重试的异常不能都为空，否则没有可以重试的场景了
			Assert.isTrue(
					!this.series.isEmpty() || !this.statuses.isEmpty()
							|| !this.exceptions.isEmpty(),
					"series, status and exceptions may not all be empty");
			//重试的Http方法不能为空
			Assert.notEmpty(this.methods, "methods may not be empty");
			if (this.backoff != null) {
				this.backoff.validate();
			}
		}
        
        //省略构造器，getter，setter还有一些工具方法
	}

	public static class BackoffConfig {
	    //第一次重试时间间隔
		private Duration firstBackoff = Duration.ofMillis(5);
        //最大等待间隔
		private Duration maxBackoff;
        //增长比例
		private int factor = 2;
        //是否保留上一次请求的重试间隔时间，下次从这个时间间隔开始重试
		private boolean basedOnPreviousValue = true;
		//省略构造器，getter，setter
		public void validate() {
		    //第一次重试间隔不能为空
			Assert.notNull(this.firstBackoff, "firstBackoff must be present");
		}
	}

}

    @Override
    public Mono filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        ServerHttpRequest request = exchange.getRequest();
        //获取微服务名称
        String serviceName = request.getHeaders().getFirst(CommonConstant.SERVICE_NAME);
        HttpMethod method = exchange.getRequest().getMethod();
        //生成 GatewayFilter,保存到 gatewayFilterMap
        GatewayFilter gatewayFilter = gatewayFilterMap.computeIfAbsent(serviceName + ":" + method, k -> {
            Map retryConfigMap = apiGatewayRetryConfig.getRetry();
            //通过微服务名称，获取重试配置
            RetryConfig retryConfig = retryConfigMap.containsKey(serviceName) ? retryConfigMap.get(serviceName) : apiGatewayRetryConfig.getDefault();
            //重试次数为0，则不重试
            if (retryConfig.getRetries() == 0) {
                return null;
            }
            //针对非GET请求，强制限制重试并且只能重试下面的异常b
            if (!HttpMethod.GET.equals(method)) {
                RetryConfig newConfig = new RetryConfig();
                BeanUtils.copyProperties(retryConfig, newConfig);
                //限制所有方法都可以重试，由于外层限制了不为GET，这里相当于不为GET的所有方法
                newConfig.setMethods(HttpMethod.values());
                newConfig.setSeries();
                newConfig.setStatuses();
                newConfig.setExceptions(//链接超时
                        io.netty.channel.ConnectTimeoutException.class,
                        //No route to host
                        java.net.ConnectException.class,
                        //针对Resilience4j的异常
                        CallNotPermittedException.class);
                retryConfig = newConfig;
            }
            return this.apply(retryConfig);
        });
        return gatewayFilter != null ? gatewayFilter.filter(exchange, chain) : chain.filter(exchange);
    }