Spring-retry的使用及原理

2020-05-14

字数统计: 3.2k字 | 阅读时长: 13分

Spring Retry提供了自动重新调用失败的操作的功能。这在错误可能是暂时性的（例如瞬时网络故障）的情况下很有用。Spring Retry提供对流程和基于策略的行为的声明式控制，易于扩展和自定义。接下来，本文将带大家了解 Spring Retry 的使用方法和部分源码解析

引入spring-retry 相关包

使用spring-retry ，我们只需引入spring-retry 和 aop 的包即可，以 maven 为例：

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

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

使用 spring-retry

@EnableRetry

首先我们需要使用@EnableRetry 注解启用Retry，注册 Retryable 的切点和拦截方法

@Retryable

在我们想要重试的方法上增加@Retryable 注解

@Service
@EnableRetry
public class SpringRetryTest {

    @Retryable
    public void test() {
        System.out.println("test");
        throw new RuntimeException();
    }
}

这样我们就可以实现一个简单的重试功能了，是不是非常的方便！当然 Spring Retry 支持很多复杂功能的实现，接下来我们就来看一下他的强大功能吧！

@Retryable 参数

interceptor

自定义重试拦截器bean名称，用于可重试方法。与其他属性互斥。

//新增自定义拦截器
@Component
public class RetryTestInterceptor implements MethodInterceptor {

    @Override
    public Object invoke(MethodInvocation invocation) throws Throwable {
      	//实现具体的拦截策略
        System.out.println("retry test interceptor");
        return invocation.proceed();
    }
}
//参数为自定义拦截器的 bean 名称
@Retryable(interceptor = "retryTestInterceptor")

value
include

value 与 include 含义相同，表示可重试的异常类型。默认为空，如果同时exclude 也为空则会重试所有异常。但在使用时需要注意
1
@Retryable(value = {RuntimeException.class})
exclude

不可重试的异常类型。默认为空（如果include也为为空，将重试所有异常）。如果include为空但exclude 不为空，则重试非 exclude 中的异常
1
@Retryable(exclude = {RuntimeException.class})
label

统计报告的唯一标签。如果未提供，则调用者可以选择忽略它或提供默认值。
stateful

标记为表示重试是有状态的，默认为 false，在 interceptor 参数中有提到，参数不同使用的拦截器不同，后面我们会讲到不同拦截的作用
maxAttempts

最大重试次数，默认是 3

maxAttemptsExpression

最大尝试次数的表达式，表达式一旦设置了值，则会覆盖 maxAttempts 的值

1 2	//执行 testBean 的 attempts() 方法，获取重试次数 @Retryable(maxAttemptsExpression = "#{@testBean.attempts()}")

backoff

用于重试退避策略，比如每隔2s 重试一次，每次重试间隔时间等等，详见@Backoff注解

1 2	//第一次延迟 1s 重试，第二次延迟 2s，第三次延迟 4s，... @Retryable(backoff = @Backoff(delay = 1000L, multiplier = 2))

exceptionExpression

异常处理表达式，ExpressionRetryPolicy中使用，执行完父类的 canRetry 之后，需要校验 exceptionExpression 的值，为 true 则可以重试
1
2
//执行 testBean 的 shouldRetry() 方法，如果为 true，则允许重试
@Retryable(exceptionExpression = "#{@testBean.shouldRetry()}")
listeners

重试监听器的 bean 名称。

重试策略

Spring Retry 提供了多种重试策略，比如

SimpleRetryPolicy

默认重试策略，简单的重试策略，它对一组命名的异常*（和子类）重试固定次数。尝试次数包括初始尝试

AlwaysRetryPolicy

始终允许重试策略，
MaxAttemptsRetryPolicy

简单重试策略，仅通过重试次数判断是否能够重试。不建议直接使用它。
TimeoutRetryPolicy

超时重试策略，仅在尚未超时的情况下允许重试。
NeverRetryPolicy

允许第一次尝试，但不允许重试。

…

还有很多重试策略就不一一介绍了，大家可以自行了解，当然我们也可以通过实现 RetryPolicy 自定义重试策略。

退避策略

退避策略就是我们上面提到的 @Backoff 注解实现的功能，那么我们首先看一下@Backoff 的参数

@Backoff 参数

value

默认为 1000，与 delay 作用相同，表示延迟的毫秒数。当 delay 非 0 时，此参数忽略。
delay

默认为 0。在指数情况下用作初始值，在统一情况下用作*的最小值。当此元素的值为0时，将采用元素value的值，否则将采用此元素的值，并且将忽略value。
maxDelay

默认为 0。重试之间的最大等待时间（以毫秒为单位）。如果小于delay，那么将应用默认值为30000L
multipler

默认为 0。如果为正，则用作乘法器以生成下一个退避延迟。返回一个乘法器，用于计算下一个退避延迟
delayExpression

评估标准退避期的表达式。在指数情况下用作初始值*，在均匀情况下用作最小值。覆盖 delay。
maxDelayExpression

该表达式计算重试之间的最大等待时间（以毫秒为单位）。如果小于 delay，那么将应用30000L 为默认值。覆盖 maxDelay。
multiplierExpression

评估为用作乘数的值，以生成退避的下一个延迟。覆盖multiplier。返回一个乘数表达式，用于计算下一个退避延迟
random

默认为 false，在指数情况下 multiplier> 0 将此值设置为 true 可以使后退延迟随机化，从而使最大延迟乘以前一延迟，并且两个值之间的分布是均匀的。

@Backoff 的参数会影响我们使用哪种退避策略

FixedBackOffPolicy

默认退避策略，每 1 秒重试 1 次
ExponentialBackOffPolicy

指数退避策略，当设置 multiplier 时使用，每次重试时间间隔为当前延迟时间 * multiplier。
ExponentialRandomBackOffPolicy

指数随机退避策略。在指数退避策略的基础上增加了随机性。具体策略查看 getSleepAndIncrement() 方法
UniformRandomBackOffPolicy

均匀随机策略，设置 maxDely 但没有设置 multiplier 时使用，重试间隔会在 maxDelay 和 delay 间随机

同样的，我们可以通过实现 BackOffPolicy 来实现自定义的退避策略

@Recover

作为恢复处理程序的方法调用的注释。重试方法最终会调用标注了@Recover 的方法

合适的恢复*处理程序具有 Throwable 类型作为第一个参数，并且具有与要从其中进行恢复的 @Retryable 方法相同类型的返回值。 Throwable 第一个参数是可选的（但是，如果没有其他参数匹配，则不带该参数的方法将被调用）。从失败方法的参数列表中依次填充后续参数。

Spring Retry 原理

@EnableRetry 作用

@EnablRetry 中使用了两个特殊的注解

@EnableAspectJAutoProxy

这个注解的作用是开启 aop 的功能，默认使用 jdk 的动态代理。如果proxyTargetClass参数为 true，则使用 cglib 的动态代理。
Import

Import 引入了 RetryConfiguration 的 bean 。我们重点看下这个 bean。

RetryConfiguration

我们可以看到 RetryConfiguration 继承了 AbstractPointcutAdvisor，所以 RetryConfiguration 需要实现 getAdvice() 和 getPointcut()接口，所以这个 bean 的作用就是为@Retryable注解注册pointcut切点和advice 增强。我们再来看他的 init 方法

其中 buildPointcut 方法就为 Retryable 设置了切入点，而 buildAdvice 则为 Retryable 设置了增强，增强的处理类为 AnnotationAwareRetryOperationsInterceptor，后续处理 Retryable 注解就需要从这个拦截类进行了。

@Retryable 增强处理

标记了@Retryable 注解的类或方法会通过 AnnotationAwareRetryOperationsInterceptor 类增强，这里会处理@Retryable 的相关参数

首先看一下 AnnotationAwareRetryOperationsInterceptor 的 invoke() 方法

@Override
	public Object invoke(MethodInvocation invocation) throws Throwable {
    //获得真正处理重试的代理类
		MethodInterceptor delegate = getDelegate(invocation.getThis(), invocation.getMethod());
		if (delegate != null) {
      // 代理类存在，则执行代理类的 invoke() 方法
			return delegate.invoke(invocation);
		}
		else {
      //否则，直接执行目标方法
			return invocation.proceed();
		}
	}

这里 getDelegate() 会处理 @Retryable 的相关参数以及决定使用哪种重试策略和退避策略。

private MethodInterceptor getDelegate(Object target, Method method) {
   ConcurrentMap<Method, MethodInterceptor> cachedMethods = this.delegates.get(target);
   if (cachedMethods == null) {
      cachedMethods = new ConcurrentHashMap<Method, MethodInterceptor>();
   }
   MethodInterceptor delegate = cachedMethods.get(method);
   if (delegate == null) {
      MethodInterceptor interceptor = NULL_INTERCEPTOR;
      //获取方法上的 @Retryable 注解
      Retryable retryable = AnnotatedElementUtils.findMergedAnnotation(method, Retryable.class);
      if (retryable == null) {
         //获取类上的 @Retryable 注解
         retryable = AnnotatedElementUtils.findMergedAnnotation(method.getDeclaringClass(), Retryable.class);
      }
      if (retryable == null) {
         //获取目标类或方法上的 @Retryable 注解
         retryable = findAnnotationOnTarget(target, method, Retryable.class);
      }
      if (retryable != null) {
         if (StringUtils.hasText(retryable.interceptor())) {
            // 自定义拦截处理
            interceptor = this.beanFactory.getBean(retryable.interceptor(), MethodInterceptor.class);
         }
         else if (retryable.stateful()) {
           	// 有状态的重试拦截
            interceptor = getStatefulInterceptor(target, method, retryable);
         }
         else {
            // 无状态的重试拦截
            interceptor = getStatelessInterceptor(target, method, retryable);
         }
      }
      cachedMethods.putIfAbsent(method, interceptor);
      delegate = cachedMethods.get(method);
   }
   this.delegates.putIfAbsent(target, cachedMethods);
   return delegate == NULL_INTERCEPTOR ? null : delegate;
}

该方法会返回 @Retryable 最终使用的处理类，我们重点看一下 getStatelessInterceptor 的处理，getStatefulInterceptor 中多了 @CircuitBreaker 熔断相关的处理。

private MethodInterceptor getStatelessInterceptor(Object target, Method method, Retryable retryable) {
   //创建 RetryTemplate，同时注册 listeners
   RetryTemplate template = createTemplate(retryable.listeners());
   //设置重试策略
   template.setRetryPolicy(getRetryPolicy(retryable));
   //设置退避策略
   template.setBackOffPolicy(getBackoffPolicy(retryable.backoff()));
   //通过 StatelessRetryInterceptorBuilder 创建 RetryOperationsInterceptor 拦截，初始化重试模板等信息
   return RetryInterceptorBuilder.stateless().retryOperations(template).label(retryable.label())
         .recoverer(getRecoverer(target, method)).build();
}

重试策略和退避策略比较简单，根据@Retryable 和 @Backoff 中的参数做相应处理。

得到 RetryOperationsInterceptor 之后，会调用它的 invoke 方法。

public Object invoke(final MethodInvocation invocation) throws Throwable {

   String name;
   if (StringUtils.hasText(label)) {
      name = label;
   }
   else {
      name = invocation.getMethod().toGenericString();
   }
   final String label = name;

   RetryCallback<Object, Throwable> retryCallback = new MethodInvocationRetryCallback<Object, Throwable>(
         invocation, label) {

      public Object doWithRetry(RetryContext context) throws Exception {

         context.setAttribute(RetryContext.NAME, label);

         /*
          * If we don't copy the invocation carefully it won't keep a reference to
          * the other interceptors in the chain. We don't have a choice here but to
          * specialise to ReflectiveMethodInvocation (but how often would another
          * implementation come along?).
          */
         if (invocation instanceof ProxyMethodInvocation) {
            try {
               //此处执行真实的业务方法，这里 clone 了一个 invocation，我暂时不清楚这么做的原因
               return ((ProxyMethodInvocation) invocation).invocableClone().proceed();
            }
            catch (Exception e) {
               throw e;
            }
            catch (Error e) {
               throw e;
            }
            catch (Throwable e) {
               throw new IllegalStateException(e);
            }
         }
         else {
            throw new IllegalStateException(
                  "MethodInvocation of the wrong type detected - this should not happen with Spring AOP, "
                        + "so please raise an issue if you see this exception");
         }
      }

   };

   if (recoverer != null) {
      ItemRecovererCallback recoveryCallback = new ItemRecovererCallback(invocation.getArguments(), recoverer);
      return this.retryOperations.execute(retryCallback, recoveryCallback);
   }
   //最终执行 RetryTemplate 的 execute() 方法
   return this.retryOperations.execute(retryCallback);

}

终于，到了 retry 逻辑处理的核心部分，我们看一下 RetryTemplate 的 doExecute() 方法

protected <T, E extends Throwable> T doExecute(RetryCallback<T, E> retryCallback,
      RecoveryCallback<T> recoveryCallback, RetryState state) throws E, ExhaustedRetryException {
	 // 获得重试策略
   RetryPolicy retryPolicy = this.retryPolicy;
   // 退避策略
   BackOffPolicy backOffPolicy = this.backOffPolicy;

   // 重试上下文，允许重试策略自行初始化
   RetryContext context = open(retryPolicy, state);
   if (this.logger.isTraceEnabled()) {
      this.logger.trace("RetryContext retrieved: " + context);
   }

   // 确保上下文对于需要的客户全局可用   
   RetrySynchronizationManager.register(context);

   Throwable lastException = null;

   boolean exhausted = false;
   try {
			// 执行我们需要的 listeners 的 open() 方法
      boolean running = doOpenInterceptors(retryCallback, context);

      if (!running) {
         throw new TerminatedRetryException("Retry terminated abnormally by interceptor before first attempt");
      }

      BackOffContext backOffContext = null;
      Object resource = context.getAttribute("backOffContext");

      if (resource instanceof BackOffContext) {
         backOffContext = (BackOffContext) resource;
      }

      if (backOffContext == null) {
         backOffContext = backOffPolicy.start(context);
         if (backOffContext != null) {
            context.setAttribute("backOffContext", backOffContext);
         }
      }

      // 这里执行重试策略
      while (canRetry(retryPolicy, context) && !context.isExhaustedOnly()) {

         try {
            lastException = null;
            // 执行业务方法
            return retryCallback.doWithRetry(context);
         }
         // 出现异常，需要重试
         catch (Throwable e) {
            lastException = e;
            try {
               // 注册出现的异常，重试上下文的失败次数+1
               registerThrowable(retryPolicy, state, context, e);
            }
            catch (Exception ex) {
               throw new TerminatedRetryException("Could not register throwable", ex);
            }
            finally {
               // 执行 listeners 的 onError() 方法
               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);
            }

         }

				 // 可以重试的有状态尝试可能会在此之前重新抛出异常，但是，如果我们在有状态重试中走得这么远，则有一定的原因，例如断路器或回滚分类器。
         if (state != null && context.hasAttribute(GLOBAL_STATE)) {
            break;
         }
      }

      if (state == null && this.logger.isDebugEnabled()) {
         this.logger.debug("Retry failed last attempt: count=" + context.getRetryCount());
      }

      exhausted = true;
      //最后尝试失败后要采取的措施。如果有状态，清理缓存。如果有恢复回调，请执行该回调并返回其结果。 否则引发异常
      return handleRetryExhausted(recoveryCallback, context, state);

   }
   catch (Throwable e) {
      throw RetryTemplate.<E>wrapIfNecessary(e);
   }
   finally {
      //如有必要，清理缓存并关闭提供的上下文
      close(retryPolicy, context, state, lastException == null || exhausted);
      // 执行 listeners 的 close() 方法
      doCloseInterceptors(retryCallback, context, lastException);
      // 在结束时清除当前上下文
      RetrySynchronizationManager.clear();
   }
}

到这里，Spring Retry 的重试实现就完成了，关于 Spring Retry 提供的熔断功能，我们下次再进行分析，希望这边文章能够帮助到大家。

欢迎关注我的公众号“shy先生”，很高兴能与大家一起成长。