从0到1带你手撸一个请求重试组件,不信你学不会!

简介: 从0到1带你手撸一个请求重试组件,不信你学不会!

背景介绍


在实际的项目应用场景中,经常会需要遇到远程服务接口的调用,时不时会出现一些接口调用超时,或者函数执行失败需要重试的情况,例如下边的这种场景:


某些不太稳定的接口,需要依赖于第三方的远程调用,例如数据加载,数据上传相关的类型。


方案整理


基于try catch机制


这种方式来做重试处理的话,会比较简单粗暴。


public void test(){
    try{
        //执行远程调用方法
        doRef();
    }catch(Exception e){
        //重新执行远程调用方法
        doRef();
    }
}


当出现了异常的时候,立即执行远程调用,此时可能忽略了几个问题:


  1. 如果重试出现了问题,是否还能继续重试


  1. 第一次远程调用出现了异常,此时可能第三方服务此时负载已达到瓶颈,或许需要间隔一段时间再发送远程调用的成功率会高些。


  1. 多次重试都失败之后如何通知调用方自己。


使用Spring的Retry组件


Spring的Retry组件提供了非常丰富的功能用于请求重试。接入这款组件的方式也很简单, 首先需要引入相关的依赖配置:


<dependency>
    <groupId>org.springframework.retry</groupId>
    <artifactId>spring-retry</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-aop</artifactId>
</dependency>


然后是在启动类上加入一个@EnableRetry注解


@SpringBootApplication
@EnableRetry
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class);
    }
}


最后是在需要被执行的函数头部加入这一@Retryable注解:


@RestController
@RequestMapping(value = "/retry")
public class RetryController {
    @GetMapping(value = "/test")
    @Retryable(value = Exception.class, maxAttempts = 3, backoff = @Backoff(delay = 2000, multiplier = 1.5))
    public int retryServiceOne(int code) throws Exception {
        System.out.println("retryServiceOne 被调用,时间" + LocalTime.now());
        System.out.println("执行当前线程为:" + Thread.currentThread().getName());
        if(code==0){
            throw new Exception("业务执行异常!");
        }
        System.out.println("retryServiceOne 执行成功!");
        return 200;
    }
}


测试结果:


请求url:http://localhost:8080/retry/test?code=0

控制台会输出相关的调用信息:


image.png


从输出记录来看,确实是spring封装好的retry组件帮我们在出现了异常的情况下会重复调用该方法多次,并且每次调用都会有对应的时间间隔。


好的,看到了这里,目前大概了解了Spring的这款重试组件该如何去使用,那么我们再来深入思考一下,如果需要通过我们手写去实现一款重试组件需要考虑哪些因素呢?下边我和大家分享下自己的一些设计思路,可能有些部分设计得并不是特别完善。


手写一款重试组件


首先我们需要定义一个retry注解:


@Documented
@Target(value = ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface Retry {
    int maxAttempts() default 3;
    int delay() default 3000;
    Class<? extends Throwable>[] value() default {};
    Class<? extends RetryStrategy> strategy() default FastRetryStrategy.class;
    Class<? extends RetryListener> listener() default AbstractRetryListener.class;
}


这款注解里面主要属性有:


  • 最大重试次数
  • 每次重试的间隔时间
  • 关注异常(仅当抛出了相应异常的条件下才会重试)
  • 重试策略(默认是快速重试)
  • 重试监听器


为了减少代码的耦合性,所以这里我将重试接口的拦截和处理都归到了aop层面去处理,因此需要引入一个对应的依赖配置:


<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-aop</artifactId>
</dependency>



重试部分的Aop模块代码如下所示:


@Aspect
@Component
public class RetryAop {
    @Resource
    private ApplicationContext applicationContext;
    @Pointcut("@annotation(org.idea.qiyu.framework.retry.jdk.config.Retry)")
    public void pointCut() {
    }
    @Around(value = "pointCut()")
    public Object doBiz(ProceedingJoinPoint point) {
        MethodSignature methodSignature = (MethodSignature) point.getSignature();
        Method method = methodSignature.getMethod();
        Retry retry = method.getDeclaredAnnotation(Retry.class);
        RetryStrategy retryStrategy = applicationContext.getBean(retry.strategy());
        RetryTask retryTask = new RetryTaskImpl(point);
        retryStrategy.initArgs(retry, retryTask);
        try {
            Object result = point.proceed();
            return result;
        } catch (Throwable throwable) {
            retryStrategy.retryTask();
        }
        return null;
    }
    private class RetryTaskImpl implements RetryTask {
        private ProceedingJoinPoint proceedingJoinPoint;
        private Object result;
        private volatile Boolean asyncRetryState = null;
        public RetryTaskImpl(ProceedingJoinPoint proceedingJoinPoint) {
            this.proceedingJoinPoint = proceedingJoinPoint;
        }
        public ProceedingJoinPoint getProceedingJoinPoint() {
            return proceedingJoinPoint;
        }
        public void setProceedingJoinPoint(ProceedingJoinPoint proceedingJoinPoint) {
            this.proceedingJoinPoint = proceedingJoinPoint;
        }
        public Object getResult() {
            return result;
        }
        public void setResult(Object result) {
            this.result = result;
        }
        public Boolean getAsyncRetryState() {
            return asyncRetryState;
        }
        public void setAsyncRetryState(Boolean asyncRetryState) {
            this.asyncRetryState = asyncRetryState;
        }
        @Override
        public Object getRetryResult() {
            return result;
        }
        @Override
        public Boolean getRetryStatus() {
            return asyncRetryState;
        }
        @Override
        public void setRetrySuccess() {
            this.setAsyncRetryState(true);
        }
        @Override
        public void doTask() throws Throwable {
            this.result = proceedingJoinPoint.proceed();
        }
    }
}


这里解释一下,这个模块主要是拦截带有 @Retry 注解的方法,然后将需要执行的部分放入到一个RetryTask类型的对象当中,内部的doTask函数会触发真正的方法调用。


RetryTask接口的代码如下:


public interface RetryTask {
     Object getRetryResult();
     Boolean getRetryStatus();
     void setRetrySuccess();
     void doTask() throws Throwable;
}


首次函数执行的过程中,会有一个try catch的捕获,如果出现了异常情况就会进入了retryTask函数内部:


image.png


在进入retryTask函数当中,则相当于进入了具体的重试策略函数执行逻辑中。


从代码截图可以看出,重试策略是从Spring容器中加载出来的,这是需要提前注入到Spring容器。


重试策略接口:


public interface RetryStrategy {
    /**
     * 初始化一些参数配置
     *
     * @param retry
     * @param retryTask
     */
    void initArgs(Retry retry,RetryTask retryTask);
    /**
     * 重试策略
     */
    void retryTask();
}


默认的重试策略为快速重试策略,相关代码为:


public class FastRetryStrategy implements RetryStrategy, ApplicationContextAware {
    private Retry retry;
    private RetryTask retryTask;
    private ApplicationContext applicationContext;
    private ExecutorService retryThreadPool;
    public FastRetryStrategy() {
    }
    public ExecutorService getRetryThreadPool() {
        return retryThreadPool;
    }
    public void setRetryThreadPool(ExecutorService retryThreadPool) {
        this.retryThreadPool = retryThreadPool;
    }
    @Override
    public void initArgs(Retry retry, RetryTask retryTask) {
        this.retry = retry;
        this.retryTask = retryTask;
    }
    @Override
    public void retryTask() {
        if (!FastRetryStrategy.class.equals(retry.strategy())) {
            System.err.println("error retry strategy");
            return;
        }
        //安全类型bean查找
        String[] beanNames = applicationContext.getBeanNamesForType(retry.listener());
        RetryListener retryListener = null;
        if (beanNames != null && beanNames.length > 0) {
            retryListener = applicationContext.getBean(retry.listener());
        }
        Class<? extends Throwable>[] exceptionClasses = retry.value();
        RetryListener finalRetryListener = retryListener;
        //如果没有支持异步功能,那么在进行重试的时候就会一直占用着服务器的业务线程,导致服务器线程负载暴增
        retryThreadPool.submit(new Runnable() {
            @Override
            public void run() {
                for (int i = 1; i <= retry.maxAttempts(); i++) {
                    int finalI = i;
                    try {
                        retryTask.doTask();
                        retryTask.setRetrySuccess();
                        return;
                    } catch (Throwable e) {
                        for (Class<? extends Throwable> clazz : exceptionClasses) {
                            if (e.getClass().equals(clazz) || e.getClass().isInstance(clazz)) {
                                if (finalRetryListener != null) {
                                    finalRetryListener.notifyObserver();
                                }
                                System.err.println("[FastRetryStrategy] retry again,attempt's time is " + finalI + ",tims is " + System.currentTimeMillis());
                                try {
                                    Thread.sleep(retry.delay());
                                } catch (InterruptedException ex) {
                                    ex.printStackTrace();
                                }
                                continue;
                            }
                        }
                    }
                }
            }
        });
    }
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
        ExecutorService executorService = (ExecutorService) applicationContext.getBean("retryThreadPool");
        this.setRetryThreadPool(executorService);
    }
}


重试的过程中专门采用了一个单独的线程池来执行相应逻辑,这样可以避免一直消耗着服务器的业务线程,导致业务线程被长时间占用影响整体吞吐率。


另外,当重试出现异常的时候,还可以通过回调对应的监听器组件做一些记录:例如日志记录,操作记录写入等等操作。


public interface RetryListener {
    /**
     * 通知观察者
     */
    void notifyObserver();
}


默认抽象类


public abstract class AbstractRetryListener implements RetryListener {
    @Override
    public abstract void notifyObserver();
}


自定义的一个监听器对象:


public class DefaultRetryListener implements RetryListener {
    @Override
    public void notifyObserver() {
        System.out.println("this is a DefaultRetryListener");
    }
}


好了,此时基本的配置都差不多了,如果需要使用的话,则需要进行一些bean的初始化配置:


@Configuration
public class RetryConfig {
    @Bean
    public FastRetryStrategy fastRetryStrategy(){
        return new FastRetryStrategy();
    }
    @Bean
    public RetryListener defaultRetryListener(){
        return new DefaultRetryListener();
    }
    @Bean
    public ExecutorService retryThreadPool(){
        ExecutorService executorService = new ThreadPoolExecutor(2,4,0L, TimeUnit.SECONDS,new LinkedBlockingQueue<>());
        return executorService;
    }
}


这里面主要将重试策略,重试监听器,重试所使用的线程池都分别进行了装载配置到Spring容器当中。


测试方式:


通过http请求url的方式进行验证:http://localhost:8080/do-test?code=2


@RestController
public class TestController {
    public static int count = 0;
    @Retry(maxAttempts = 5, delay = 100, value = {ArithmeticException.class}, strategy = FastRetryStrategy.class, listener = DefaultRetryListener.class)
    @GetMapping(value = "/do-test")
    public String doTest(int code) {
        count++;
        System.out.println("code is :" + code + " result is :" + count % 3 + " count is :" + count);
        if (code == 1) {
            System.out.println("--this is a test");
        } else {
            if (count % 5 != 0) {
                System.out.println(4 / 0);
            }
        }
        return "success";
    }
}


请求之后可以看到控制台输出了对应的内容:


image.png


不足点


  1. 需要指定完全匹配的异常才能做到相关的重试处理,这部分的处理步骤会比较繁琐,并不是特别灵活。


  1. 一定要是出现了异常才能进行重试,但是往往有些时候可能会返回一些错误含义的DTO对象,这方面的处理并不是那么灵活。


guava-retryer的重试组件就在上述的几个不足点中有所完善,关于其具体使用就不在本文中介绍了,感兴趣的小伙伴可以去了解下这款组件的使用细节。


相关文章
|
4月前
|
数据采集 存储 JSON
推荐3款自动爬虫神器,再也不用手撸代码了
推荐3款自动爬虫神器,再也不用手撸代码了
终于搞明白什么叫做回调了
终于搞明白什么叫做回调了
62 0
|
Java 微服务
手撸一个动态Feign,实现一个“万能”接口调用
手撸一个动态Feign,实现一个“万能”接口调用
|
前端开发 Java Spring
求求你不要写满屏的 try...catch 了,这才是优雅的处理方式,真香...
求求你不要写满屏的 try...catch 了,这才是优雅的处理方式,真香...
278 0
求求你不要写满屏的 try...catch 了,这才是优雅的处理方式,真香...
|
前端开发 测试技术 API
同事的接口文档我每次看着就头大,毛病多多。。。(2)
同事的接口文档我每次看着就头大,毛病多多。。。(2)
127 0
同事的接口文档我每次看着就头大,毛病多多。。。(2)
|
JSON 前端开发 JavaScript
同事的接口文档我每次看着就头大,毛病多多。。。(1)
同事的接口文档我每次看着就头大,毛病多多。。。(1)
151 0
同事的接口文档我每次看着就头大,毛病多多。。。(1)
|
消息中间件 存储 前端开发
面试官让我手写队列,差点没写出来,回来后赶忙把重点记下来
栈和队列是一对好兄弟,前面我们介绍过一篇栈的文章(栈,不就后进先出),栈的机制相对简单,后入先出,就像进入一个狭小的山洞,山洞只有一个出入口,只能后进先出(在外面的先出去,堵在里面先进去的就有点倒霉)。而队列就好比是一个隧道,后面的人跟着前面走,前面人先出去(先入先出)。日常的排队就是队列运转形式的一个描述!
117 0
面试官让我手写队列,差点没写出来,回来后赶忙把重点记下来
|
前端开发 JavaScript 索引
2022年了!再来手撕一下前端瀑布流代码吧!
**前言: **知识是学不完的,可是我们为什么还是要不停的去学习呢。原因很简单,因为我们要产生更多的知识,让更多的人学不完!前端技术也是在不停的革新,我们要做那个让别人有学不完的知识的人
1021 0
2022年了!再来手撕一下前端瀑布流代码吧!
|
前端开发 JavaScript 容器
【一个让你停不下来的动效】——难道不来瞅瞅?(含源码+思路)
【一个让你停不下来的动效】——难道不来瞅瞅?(含源码+思路)
170 0