Spring Retry实现重试机制的示例详解

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.7.2</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <modelVersion>4.0.0</modelVersion>
    <artifactId>spring-retry</artifactId>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.retry</groupId>
            <artifactId>spring-retry</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-aop</artifactId>
        </dependency>
    </dependencies>
</project>

@RestController
//开启Spring-Retry重试机制
@EnableRetry
@SpringBootApplication
public class App {
    public static void main(String[] args) {
        SpringApplication.run(App.class,args);
    }
    @Resource
    private RetryService retryService;
    @GetMapping("/test")
    public String test(@RequestParam("code") Integer code) throws Exception{
        retryService.retry(code);
        return "ok";
    }
}

package com.coco.service.impl;
import com.coco.service.RetryService;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Recover;
import org.springframework.retry.annotation.Retryable;
import org.springframework.stereotype.Service;
import java.io.IOException;
@Service
public class RetryServiceImpl implements RetryService {
    /**
     * value：抛出指定异常才会重试
     * include：和value一样，默认为空，当exclude也为空时，默认所有异常
     * exclude：指定不处理的异常
     * maxAttempts：最大重试次数，默认3次
     * backoff：重试等待策略，
     * 默认使用@Backoff，@Backoff的value默认为1000L，我们设置为2000； 以毫秒为单位的延迟（默认 1000）
     * multiplier（指定延迟倍数）默认为0，表示固定暂停1秒后进行重试，如果把multiplier设置为1.5，则第一次重试为2秒，第二次为3秒，第三次为4.5秒。
     */
    @Retryable(value = RuntimeException.class,maxAttempts = 3,backoff = @Backoff(delay = 2000,multiplier = 1.5))
    @Override
    public void retry(int code) throws Exception {
        System.out.println("retry被调用了");
        if (code==0){
            throw new IOException("调用失败，重试");
        }
        System.out.println("调用成功");
    }
    /**
     * Spring-Retry还提供了@Recover注解，用于@Retryable重试失败后处理方法。
     * 如果不需要回调方法，可以直接不写回调方法，那么实现的效果是，重试次数完了后，如果还是没成功没符合业务判断，就抛出异常。
     * 可以看到传参里面写的是 Exception e，这个是作为回调的接头暗号（重试次数用完了，还是失败，我们抛出这个Exception e通知触发这个回调方法）。
     * 注意事项：
     * 方法的返回值必须与@Retryable方法一致
     * 方法的第一个参数，必须是Throwable类型的，建议是与@Retryable配置的异常一致，其他的参数，需要哪个参数，写进去就可以了（@Recover方法中有的）
     * 该回调方法与重试方法写在同一个实现类里面
     *
     * 由于是基于AOP实现，所以不支持类里自调用方法
     * 如果重试失败需要给@Recover注解的方法做后续处理，那这个重试的方法不能有返回值，只能是void
     * 方法内不能使用try catch，只能往外抛异常
     * @Recover注解来开启重试失败后调用的方法(注意,需跟重处理方法在同一个类中)，此注解注释的方法参数一定要是@Retryable抛出的异常，否则无法识别，可以在该方法中进行日志处理。
     */
    @Recover
    public void recover(Exception e, int code){
        System.out.println("回调方法执行！！！！");
        //记日志到数据库 或者调用其余的方法
        System.out.println("异常信息:"+e.getMessage());
    }
}

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@EnableAspectJAutoProxy(proxyTargetClass = false)
//注解里我们尤其要关注@Import注解，因为这是Spring将一个Bean注入到容器中的
@Import(RetryConfiguration.class)
@Documented
public @interface EnableRetry {
   /**
    * Indicate whether subclass-based (CGLIB) proxies are to be created as opposed to
    * standard Java interface-based proxies. The default is {@code false}.
    * @return whether to proxy or not to proxy the class
    */
   boolean proxyTargetClass() default false;
}

@Override
public void afterPropertiesSet() throws Exception {
   this.retryContextCache = findBean(RetryContextCache.class);
   this.methodArgumentsKeyGenerator = findBean(MethodArgumentsKeyGenerator.class);
   this.newMethodArgumentsIdentifier = findBean(NewMethodArgumentsIdentifier.class);
   this.retryListeners = findBeans(RetryListener.class);
   this.sleeper = findBean(Sleeper.class);
   Set<Class<? extends Annotation>> retryableAnnotationTypes = new LinkedHashSet<Class<? extends Annotation>>(1);
   retryableAnnotationTypes.add(Retryable.class);
   //构建AOP切面和通知
   this.pointcut = buildPointcut(retryableAnnotationTypes);
   this.advice = buildAdvice();
   if (this.advice instanceof BeanFactoryAware) {
      ((BeanFactoryAware) this.advice).setBeanFactory(this.beanFactory);
   }
}

@Override
public Object invoke(final MethodInvocation invocation) throws Throwable {
   String name;
   if (StringUtils.hasText(this.label)) {
      name = this.label;
   }
   else {
      name = invocation.getMethod().toGenericString();
   }
   final String label = name;
   //初始化重试机制的回调函数，这里是重点，在重试执行我们的业务逻辑的时候，就会进入到
   //这里回调函数中，然后执行doWithRetry()方法，但是第一次只是初始化，并不会进入到这里面
   RetryCallback<Object, Throwable> retryCallback = new MethodInvocationRetryCallback<Object, Throwable>(
         invocation, label) {
      @Override
      public Object doWithRetry(RetryContext context) throws Exception {
         context.setAttribute(RetryContext.NAME, this.label);
         if (this.invocation instanceof ProxyMethodInvocation) {
            context.setAttribute("___proxy___", ((ProxyMethodInvocation) this.invocation).getProxy());
            try {
               return ((ProxyMethodInvocation) this.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");
         }
      }
   };
   //还记得我们在自己RetryServiceImpl中实现了一个方法recover()，并且用@Recover标记
   //如果我们实现了这个方法，那么this.recoverer就不为空，就会进入到if分支里面去
   //最后调用this.retryOperations.execute()方法
   if (this.recoverer != null) {
      ItemRecovererCallback recoveryCallback = new ItemRecovererCallback(invocation.getArguments(),
            this.recoverer);
      try {
         Object recovered = this.retryOperations.execute(retryCallback, recoveryCallback);
         return recovered;
      }
      finally {
         RetryContext context = RetrySynchronizationManager.getContext();
         if (context != null) {
            context.removeAttribute("__proxy__");
         }
      }
   }
   //如果我们自己没有实现recover()方法，那么this.recoverer就等于null，就会直接进入到这里面来了
   return this.retryOperations.execute(retryCallback);
}

public final <T, E extends Throwable> T execute(RetryCallback<T, E> retryCallback,
      RecoveryCallback<T> recoveryCallback) throws E {
   //继续进入doExecute方法
   return doExecute(retryCallback, recoveryCallback, null);
}

protected <T, E extends Throwable> T doExecute(RetryCallback<T, E> retryCallback,
      RecoveryCallback<T> recoveryCallback, RetryState state) throws E, ExhaustedRetryException {
   //RetryPolicy这个对象包含二个属性
   //maxAttempts：也就是重试的最大次数，当达到这个次数之后就不会再次重试了
   //retryableClassifier：还记得我们加在方法上的@Retryable(value = RuntimeException.class,maxAttempts = 3,backoff = @Backoff(delay = 2000,multiplier = 1.5))
   //这里设置了一个异常类型，表示的是只有返回的是这个类型的异常才会进行重试
   //如果返回的是其它类型的异常就不会进行重试，所以retryableClassifier这个值就是保存注解
   //里面value设置的异常类型
   RetryPolicy retryPolicy = this.retryPolicy;
   BackOffPolicy backOffPolicy = this.backOffPolicy;
   //初始化我们当前线程重试的上下文
   //在上下文中有一个很重的属性count,初始化的时候这个值为0，后续重试一次，这个值就会加1
   RetryContext context = open(retryPolicy, state);
   //将上下文保存到ThreadLocal中，也是防止并发安全
   RetrySynchronizationManager.register(context);
   Throwable lastException = null;
   boolean exhausted = false;
   try {
      // 给客户一个机会来增强上下文。。。，这里不是重点
      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),意思就是是否可以重试
      while (canRetry(retryPolicy, context) && !context.isExhaustedOnly()) {
         try {
            lastException = null;
            //如果可以重试，就会执行doWithRetry()方法
            //在之前我们分析RetryOperationsInterceptor类中的invoke()方法的时候，在那里
            //已经实现了回调方法，所以此时就会进入到那个回调方法中
            return retryCallback.doWithRetry(context);
         }
         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;
                  throw ex;
               }
            }
            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;
         }
      }
      exhausted = true;
      return handleRetryExhausted(recoveryCallback, context, state);
   }
   catch (Throwable e) {
      throw RetryTemplate.<E>wrapIfNecessary(e);
   }
   finally {
      //清除上下文信息
      close(retryPolicy, context, state, lastException == null || exhausted);
      doCloseInterceptors(retryCallback, context, lastException);
      //将ThreadLocal中的上下文信息清除1掉
      RetrySynchronizationManager.clear();
   }
}

protected boolean canRetry(RetryPolicy retryPolicy, RetryContext context) {
   //进入这个方法
   return retryPolicy.canRetry(context);
}

public boolean canRetry(RetryContext context) {
   Throwable t = context.getLastThrowable();
   //retryForException(t)：判断返回的异常是否跟我们注解设置的异常类型一致，
   //                      在分析RetryPolicy对象中有个属性就保存了我们注解设置的异常类型
   //context.getRetryCount() < getMaxAttempts()：重试次数是否已经达到了我们设置的最大次数
   return (t == null || retryForException(t)) && context.getRetryCount() < getMaxAttempts();
}

//这段代码就是RetryOperationsInterceptor拦截器中的invoke()方法，我把这段代码截取出来了
RetryCallback<Object, Throwable> retryCallback = new MethodInvocationRetryCallback<Object, Throwable>(invocation, name) {
    //执行这段方法
    public Object doWithRetry(RetryContext context) throws Exception {
        context.setAttribute("context.name", this.label);
        if (this.invocation instanceof ProxyMethodInvocation) {
            context.setAttribute("___proxy___", ((ProxyMethodInvocation)this.invocation).getProxy());
            try {
                // 这里就是执行我们自己的业务逻辑了，如果有异常就抛出，然后在重试机制的
                // while循环中捕获，继而判断异常是否符合并且重试次数是否达到，如果条件符合
                //就继续重试执行，如果不符合，就不会再重试了
                return ((ProxyMethodInvocation)this.invocation).invocableClone().proceed();
            } catch (Exception var3) {
                throw var3;
            } catch (Error var4) {
                throw var4;
            } catch (Throwable var5) {
                throw new IllegalStateException(var5);
            }
        } 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");
        }
    }
};