前瞻

阅读本篇前,请先阅读smart-retry源码阅读文章

背景

smart-retry有以下缺点

• 只支持入参有且仅有一个
• 每一个重试方法都对应一个定时任务,会造成线程的过度使用
• 不支持抛出指定异常后重试

基于此,对smart-retry做了升级改造

改造后特性

针对smart-retry的确定的改进

• 支持多个入参方法重试
• 只提供重试接口给用户,具体定时任务选用由用户决定,灵活性大大增加
• 支持抛出指定异常后重试

新增的功能

• 支持配置在注解上配置是否在执行方法前入库
• 增加了重试记录表,把每次重试都记录下来
• 重试规则由cron表达式改为按时间,或者间隔重试,更通俗易懂

代码跟进改造点

下边会讲述代码如何改造

支持多个入参方法重试

改造前入参,以及参数类型都是单个,改成数组,我列举这两种的前后对比,因为改的地方比较多,我不一一列举

改造前入参

   public Object handle(Object arg) {

改造后入参

public Object handle(Object arg[]) {

改造前入参类型(用于序列化和反序列化)

   public Class<?> getInputArgsType() {

       return inputArgsType;

   }

改造后入参类型(用于序列化和反序列化)

   public Class<?>[] getInputArgsType() {

       return inputArgsType;

   }

只提供重试接口给用户,具体定时任务选用由用户决定,灵活性大大增加

我们先看原来的代码

   @Override

   public void afterSingletonsInstantiated() {

       postedClasseCache.clear();

       this.retryTaskMapper = defaultListableBeanFactory.getBean(RetryTaskMapper.class);

       this.retryRegistry = defaultListableBeanFactory.getBean(RetryRegistry.class);

       boolean beforeTask = environment.getProperty(EnvironmentConstants.RETRY_BEFORETASK, Boolean.class, Boolean.TRUE);

       this.retrySerializer = getRetrySerializerFromBeanFactory(defaultListableBeanFactory);

       if (this.retrySerializer == null) {

           this.retryHandlerPostProcessor = new DefaultRetryHandlerPostProcessor(retryTaskMapper, beforeTask);

       } else {

           this.retryHandlerPostProcessor = new DefaultRetryHandlerPostProcessor(new DefaultRetryTaskFactory(retrySerializer), retryTaskMapper, beforeTask);

       }

       retryHandlers.forEach(this::registerJobBean);

       retryHandlers.clear();

   }

   protected void registerJobBean(RetryHandler retryHandler) {

       if (retryHandler.identity().length() > 50) {

           throw new IllegalArgumentException("identity=" + retryHandler.identity() + " is too long, it must be less than 50");

       }

       RetryHandler retryHandlerProxy = retryHandlerPostProcessor.doPost(retryHandler);

       RetryHandlerRegistration.registry(retryHandlerProxy);

       RetryProcessor retryProcessor = new DefaultRetryProcessor(retryHandler, retryTaskMapper, retrySerializer);

       retryRegistry.register(retryHandler, retryProcessor);

   }

   @Override

   public void register(RetryHandler retryHandler, RetryProcessor retryProcessor) {

       if (StringUtils.isBlank(retryHandler.cron())) {

           throw new IllegalArgumentException("identity=" + retryHandler.identity() + ", 使用Elastic-Job注册器，必须指定RetryHandler/RetryFunction的cron表达式");

       }

       BeanDefinitionBuilder beanDefinitionBuilder = BeanDefinitionBuilder.rootBeanDefinition(SpringJobScheduler.class);

       beanDefinitionBuilder.addConstructorArgValue(new RetryJob(retryProcessor));

       beanDefinitionBuilder.addConstructorArgValue(registryCenter);

       beanDefinitionBuilder.addConstructorArgValue(createLiteJobConfiguration(retryHandler));

       if (jobEventConfiguration != null) {

           beanDefinitionBuilder.addConstructorArgValue(jobEventConfiguration);

       }

       beanDefinitionBuilder.addConstructorArgValue(getElasticJobListeners());

       beanDefinitionBuilder.setInitMethodName("init");

       String jobBeanName = getJobBeanName(retryHandler);

       defaultListableBeanFactory.registerBeanDefinition(jobBeanName, beanDefinitionBuilder.getBeanDefinition());

       //此处的getBean调用是为了手工触发Bean的初始化

       defaultListableBeanFactory.getBean(jobBeanName);

       log.info("identity={}已成功注册到Elastic-Job", retryHandler.identity());

   }

可以看到,每一个retryHandler会单独注册一个定时任务,并且注册的时候需要指定注册到哪个定时任务

再看改造后代码,先拿到所有retryHandler,封装为可执行的RetryFunctionProcessor

   @Override

   public void afterSingletonsInstantiated() {

       postedClasseCache.clear();

       this.retryTaskMapper = defaultListableBeanFactory.getBean(RetryTaskMapper.class);

       RetryFunctionProcessor retryFunctionProcessor = defaultListableBeanFactory.getBean(RetryFunctionProcessor.class);

       this.retrySerializer = getRetrySerializerFromBeanFactory(defaultListableBeanFactory);

       List<RetryProcessor> retryProcessorList = new ArrayList<>();

       for (RetryHandler retryHandler : retryHandlers) {

           if (retryHandler.identity().length() > 50) {

               throw new IllegalArgumentException("identity=" + retryHandler.identity() + " is too long, it must be less than 50");

           }

           RetryHandler retryHandlerProxy =new ImmediatelyRetryHandler(retryHandler, new DefaultRetryTaskFactory(retrySerializer), retryTaskMapper, retryRule);

           RetryHandlerRegistration.registry(retryHandlerProxy);

           RetryProcessor retryProcessor = new DefaultRetryProcessor(retryHandler, retryTaskMapper, retrySerializer, retryRule);

           retryProcessorList.add(retryProcessor);

       }

       

       retryFunctionProcessor.register(retryProcessorList);

       retryHandlers.clear();

   }

RetryFunctionProcessor其实就是把retryProcessorList保存为成员变量,提供给用户去调用,调用方式将在如何使用篇章讲解

支持抛出特定异常后重试

smart-retry只能在抛出RuntimeException的时候,进行重试,改造后,可以支持抛出特定异常后重试,这样当自定义异常时,更加灵活,代码改造点如下

RetryFunction注解增加变量

   /**

    * 如果方法抛出该异常则会创建重试任务

    */

   Class<? extends RuntimeException>[] retryException() default {RuntimeException.class};

在执行方法抛出异常时,判断是否在配置的这些异常范围,来决定是否要进行重试

       } catch (RuntimeException e) {

           boolean isIncludeException = RetryHandlerUtils.isIncludeException(e, retryException);

           if (!isIncludeException) {

               throw e;

           }

支持配置在注解上配置是否在执行方法前入库

smart-retry是支持是否在执行方法前入库的,但是是全局的配置,改造后可以对每一个任务单独配置

   /**

    * 是否在执行任务之前插入数据库 |配置false则表示，只有任务执行报错才插入数据库|

    * @return

    */

   boolean beforeTask() default false;

具体逻辑就比较简单了,就是根据配置,来决定是否在抛出异常前插入数据

增加了重试记录表,把每次重试都记录下来

smart-retry只有一张重试表,比如重试过5次,没办法知道这5次重试的具体详情,只能知道最后一次是成功还是失败等,所以增加了重试记录表,来记录每一次重试的记录,利用了spring的切面的原理,对原代码没有任何侵入

@Aspect

@Slf4j

public class RetryTaskRecordAspect implements ApplicationContextAware {

   private ApplicationContext applicationContext;

   

   @Pointcut("execution(* com.aliyun.gts.bpaas.retry.core.RetryTaskMapper.insert(..)) ||" +

           "execution(* com.aliyun.gts.bpaas.retry.core.RetryTaskMapper.update(..))")

   public void pointCut(){

       

   }

   

   @Around(value = "pointCut()")

   public Object insertTaskRecord(ProceedingJoinPoint proceedingJoinPoint) {

       Object o = null;

       try {

           o = proceedingJoinPoint.proceed();

       } catch (Throwable throwable) {

           log.error("切面insertTaskRecord报错,", throwable);

       }

       Object[] args = proceedingJoinPoint.getArgs();

       RetryTask retryTask = (RetryTask) args[0];

       RetryTaskMapper retryTaskMapper = applicationContext.getBean(RetryTaskMapper.class);

       retryTaskMapper.insertTaskRecord(retryTask);

       return o;

   }

   @Override

   public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {

       this.applicationContext = applicationContext;  

   }

}

重试规则由cron表达式改为按时间,或者间隔重试,更通俗易懂

改造前:smart-retry原来的重试规则,是为每一条重试任务配置一个cron表达式

改造后:重试规则在application.properties中全局配置,更方便管理,cron表达式改为按时间,或者间隔,更容易配置,不易配置错

   /**

    * 重试规则,有两种方式,指定时间或者间隔

    * 1.时间,格式为:onTime-时间-最大重试次数,例如  onTime-23:59:59-3,onTime为类型,23:59:59为时间的时分秒,3位最大重试次数,用中划线隔开

    * 2.间隔,格式为:onInterval-间隔重试规则,例如 onInterval-10,20,30,onInterval为类型,10,20,30表示每次重试的间隔时间,单位为秒

    */

   private String retryRule;

指定时间重试的实现逻辑:获取到下次的执行时间,如果执行时间小于当前时间,则需要加一天,否则,就是该时间

指定间隔重试的实现逻辑:现在的时间,加上重试次数对应的间隔时间即可

代码如下

   /**

    * 获取下次执行时间

    * @param retryRule

    * @param retryCount

    * @return

    */

   public static LocalDateTime getNextExecTime(RetryRule retryRule, int retryCount) {

       

       if (retryRule instanceof OnTimeRetryRule) {

           OnTimeRetryRule onTimeRetryRule = (OnTimeRetryRule) retryRule;

           LocalTime retryTime = onTimeRetryRule.getRetryTime();

           LocalDateTime nextExecTime = LocalDateTime.of(LocalDate.now(), retryTime);

           

           if (nextExecTime.isAfter(LocalDateTime.now())) {

               return nextExecTime;

           } else {

               return nextExecTime.plusDays(1L);

           }

       }

       OnIntervalRetryRule onIntervalRetryRule = (OnIntervalRetryRule) retryRule;

       Long[] retryInterval = onIntervalRetryRule.getRetryInterval();

       

       return plusSeconds(retryInterval[retryCount]);

   }

如何使用

建表

接口重推依赖两张表重试表以及重试记录表

重试表sql

create table sys_retry_task (

task_id bigint not null primary key auto_increment,

identity_name varchar(50) not null COMMENT '任务的唯一标识',

params text COMMENT '参数',

status tinyint not null COMMENT '状态。1: 处理中，2: 成功，3: 失败',

retry_count int not null default 0 COMMENT '重试次数',

remark varchar(1000) COMMENT '备注',

create_date datetime not null,

edit_date datetime,

next_date DATETIME COMMENT '下次执行的时间') ENGINE=InnoDB COMMENT='系统重试表';

create index idx_identityname_status ON sys_retry_task(identity_name asc,status asc);

重试记录表sql

CREATE TABLE `sys_retry_task_record`

(

   `id`            BIGINT(20)    NOT NULL AUTO_INCREMENT,

   `task_id`       BIGINT(20)    NULL     DEFAULT NULL COMMENT '重试任务id',

   `identity_name` VARCHAR(50)   NOT NULL COMMENT '任务的唯一标识' COLLATE 'utf8_general_ci',

   `params`        TEXT          NULL     DEFAULT NULL COMMENT '参数' COLLATE 'utf8_general_ci',

   `status`        TINYINT(4)    NOT NULL COMMENT '状态。1: 处理中，2: 成功，3: 失败',

   `retry_count`   INT(11)       NOT NULL DEFAULT '0' COMMENT '重试次数',

   `remark`        VARCHAR(1000) NULL     DEFAULT NULL COMMENT '备注' COLLATE 'utf8_general_ci',

   `next_date`     DATETIME      NULL     DEFAULT NULL COMMENT '下次执行的时间',

   `edit_date`     DATETIME      NULL     DEFAULT NULL,

   `create_date`   DATETIME      NOT NULL,

   PRIMARY KEY (`id`) USING BTREE,

   INDEX `idx_identityname_status` (`identity_name`, `status`) USING BTREE

)

   COMMENT ='系统重试记录表'

   COLLATE = 'utf8_general_ci'

   ENGINE = InnoDB

引入依赖

       <dependency>

           <groupId>com.aliyun.gts.bpaas</groupId>

           <artifactId>aliyun-gts-retry-starter</artifactId>

           <version>1.0.0-SNAPSHOT</version>

       </dependency>

application.properties配置

# 数据源配置,注意要和建表的数据源和数据库在同一个

spring.datasource.driver-class-name=com.mysql.jdbc.Driver

spring.datasource.url=jdbc:mysql://localhost:3306/test?characterEncoding=utf8&autoReconnect=true&useSSL=false

spring.datasource.username=root

spring.datasource.password=123456

# 重试开关,默认为false

gts.retry.enable=true

# 重试的规则

gts.retry.retry-rule=onInterval-10,30,30,30

# 序列化方式,支持fastjson,jackson以及gson,默认为fastjson

gts.retry.serialize-type=fastjson

下边对重试规则配置gts.retry.retry-rule进行详细说明

重试规则,有两种方式,指定时间或者间隔
1. 时间,格式为:onTime-时间-最大重试次数,例如(onTime-23:59:59-3),onTime为类型,23:59:59为时间的时分秒,3位最大重试次数,用中划线隔开
2. 间隔,格式为:onInterval-间隔重试规则,例如 (onInterval-10,20,30),onInterval为类型,10,20,30表示每次重试的间隔时间,单位为秒

编程界面

接口重推需要的编程界面有两处,第一处是在方法上打上@RetryFunction注解,第二处是在定时任务中调用处理定时任务的方法

在方法上打上@RetryFunction注解

   @RetryFunction(identity = "demo.simplest", beforeTask = true,

           retryListener = SimpleTestRetryListener.class,

           retryException = {RuntimeException.class}, ignoreException = true)

   public void simplestWithId(int id) {

       

       log.info("simplestWithId[{}]执行开始", id);

       // doSomething()

       log.info("simplestWithId[{}]执行完成", id);

   }

下面对注解中的每一个属性做详细的解释

定时任务中触发重试任务示例

@Slf4j

@Component

public class DistributeRetrySchedule extends AbstractGtsSchedulerTaskProcessor {

   

   @Autowired

   private RetryFunctionProcessor retryFunctionProcessor;

   

   @Override

   public String getTaskId(){

       return "retry-job";

   }

   @Override

   public GtsSchedulerTaskResult process(GtsSchedulerTaskParameter parameter) throws Exception{

       DateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

       log.info("执行定时任务，当前时间{}",format.format(new Date()));

       retryFunctionProcessor.processTask();

       return GtsSchedulerTaskResult.successResult();

   }

}

重试任务持久化示例

重试表

重试记录表