SpringBoot定时任务动态扩展ScheduledTaskRegistrar详解

摘要

本文主要介绍基于SpringBoot定时任务ScheduledTaskRegistrar的动态扩展，实现定时任务的动态新增和删除。

ScheduledTaskRegistrar类简要描述

平常使用方式配置

• Application启动类上添加注解@EnableScheduling

@EnableScheduling
@SpringBootApplication(exclude = {DataSourceAutoConfiguration.class})
public class Application {
   public static void main(String[] args) {
       SpringApplication.run(Application.class, args);
   }
}

• 在需要定时的方法上添加定时注解@Scheduled(cron = "0/10 * * * * ?")

@Slf4j
@Component
public class OtherScheduler {
   @Scheduled(cron = "0/10 * * * * ?")
   public void print(){
       log.info("每10S打印一次");
   }
   @Scheduled(cron = "0/5 * * * * ?")
   public void print5(){
       log.info("每5S打印一次");
   }
}

原理分析

默认的方式启动把ScheduledAnnotationBeanPostProcessor该类实例化到SpringBoot的Bean管理中，并且该类持有一个ScheduledTaskRegistrar属性，然后扫描出来拥有@Scheduled注解的方法，添加到定时任务中。

• 添加定时任务到列表中

扫描到@Scheduled注解的时候调用了该方法添加任务

public void addCronTask(Runnable task, String expression) {
if (!CRON_DISABLED.equals(expression)) {
addCronTask(new CronTask(task, expression));
}
}

• 启动定时任务

在对象实例化完成后，调用了afterPropertiesSet方法，该方法实际使用中执行了

public void afterPropertiesSet() {
scheduleTasks();
}
protected void scheduleTasks() {
if (this.taskScheduler == null) {
this.localExecutor = Executors.newSingleThreadScheduledExecutor();
this.taskScheduler = new ConcurrentTaskScheduler(this.localExecutor);
}
if (this.triggerTasks != null) {
for (TriggerTask task : this.triggerTasks) {
addScheduledTask(scheduleTriggerTask(task));
}
}
if (this.cronTasks != null) {
for (CronTask task : this.cronTasks) {
addScheduledTask(scheduleCronTask(task));
}
}
if (this.fixedRateTasks != null) {
for (IntervalTask task : this.fixedRateTasks) {
addScheduledTask(scheduleFixedRateTask(task));
}
}
if (this.fixedDelayTasks != null) {
for (IntervalTask task : this.fixedDelayTasks) {
addScheduledTask(scheduleFixedDelayTask(task));
}
}
}
private void addScheduledTask(@Nullable ScheduledTask task) {
if (task != null) {
this.scheduledTasks.add(task);
}
}
// 启动任务核心方法
public ScheduledTask scheduleCronTask(CronTask task) {
ScheduledTask scheduledTask = this.unresolvedTasks.remove(task);
boolean newTask = false;
if (scheduledTask == null) {
scheduledTask = new ScheduledTask(task);
newTask = true;
}
if (this.taskScheduler != null) {
scheduledTask.future = this.taskScheduler.schedule(task.getRunnable(), task.getTrigger());
}
else {
addCronTask(task);
this.unresolvedTasks.put(task, scheduledTask);
}
return (newTask ? scheduledTask : null);
}

DynamicScheduledTaskRegistrar 动态任务注册类

下面改动主要涉及到线程池数量、新增任务、删除任务、销毁任务四个方面；

public class DynamicScheduledTaskRegistrar extends ScheduledTaskRegistrar {
   private static final Logger log = LoggerFactory.getLogger(DynamicScheduledTaskRegistrar.class);
   private final Map<String,ScheduledTask> scheduledTaskMap = new LinkedHashMap<>(16);
   public DynamicScheduledTaskRegistrar(){
       super();
       // 两种实现方案
       //ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(10);
       //TaskScheduler taskScheduler = new ConcurrentTaskScheduler(scheduledExecutorService);
       // 第二种实现方案
       ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
       taskScheduler.setPoolSize(8);
       taskScheduler.setRemoveOnCancelPolicy(true);
       taskScheduler.setThreadNamePrefix("dynamic-scheduled-task-");
       taskScheduler.initialize();
       this.setScheduler(taskScheduler);
   }
   /**
    * 新增任务
    * @param taskName
    * @param cron
    * @param runnable
    */
   public Boolean addCronTask(String taskName,String cron,Runnable runnable){
       if(scheduledTaskMap.containsKey(taskName)){
           log.error("定时任务["+ taskName+"]已存在，添加失败");
           return Boolean.FALSE;
       }
       CronTask cronTask = new CronTask(runnable,cron);
       ScheduledTask scheduledTask = this.scheduleCronTask(cronTask);
       scheduledTaskMap.put(taskName,scheduledTask);
       log.info("定时任务["+taskName+"]新增成功");
       return Boolean.TRUE;
   }
   /**
    * 删除任务
    * @param taskName
    */
   public void cancelCronTask(String taskName){
       ScheduledTask scheduledTask = scheduledTaskMap.get(taskName);
       if(null != scheduledTask){
           scheduledTask.cancel();
           scheduledTaskMap.remove(taskName);
       }
       log.info("定时任务["+taskName+"]删除成功");
   }
   @Override
   public void destroy() {
       super.destroy();
       scheduledTaskMap.values().forEach(ScheduledTask::cancel);
   }
}

线程池数量问题

由于默认是单线程的，如果任务阻塞时间过长则会导致后续的任务阻塞，所以尽量是异步任务或者是线程池数量大一点，则可以避免这个问题

DynamicScheduledTaskService

@Service
public class DynamicScheduledTaskService {
   private static final Logger log = LoggerFactory.getLogger(DynamicScheduledTaskService.class);
   private final DynamicScheduledTaskRegistrar dynamicScheduledTaskRegistrar = new DynamicScheduledTaskRegistrar();
   /**
    * 新增任务
    * @param taskName
    * @param cron
    */
   public void add(String taskName,String cron){
       Boolean result = dynamicScheduledTaskRegistrar.addCronTask(taskName,cron,() -> print(taskName));
       log.info("定时任务添加结果：" + result);
   }
   /**
    * 取消任务
    * @param taskName
    */
   public void cancel(String taskName){
       dynamicScheduledTaskRegistrar.cancelCronTask(taskName);
   }
   private void print(String taskName){
       log.info(taskName+"开始");
       try{
           Thread.sleep(9000L);
           log.info(taskName+"结束111");
       }catch (Exception ex){
       }
       log.info(taskName+"结束");
   }
}

SchedulerController

@RestController
@RequestMapping(value = "scheduler")
public class SchedulerController {
   @Autowired
   private DynamicScheduledTaskService dynamicScheduledTaskService;
   @GetMapping(value = "add")
   public Object add(String taskName,String cron){
       dynamicScheduledTaskService.add(taskName,cron);
       return "SUCCESS";
   }
   @GetMapping(value = "cancel")
   public Object cancel(String jobName){
       dynamicScheduledTaskService.cancel(jobName);
       return "SUCCESS";
   }
}

测试结果

新增的任务都睡眠了9S

新增调度任务

删除调度任务

来源：https://juejin.cn/post/7187628561430216765