SpringBoot定时任务动态扩展ScheduledTaskRegistrar

末蓝、 2023-09-23 22:08 102阅读 0赞

**摘要**：本文主要介绍基于`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

新增调度任务

![format_png][]

![format_png 1][]

删除调度任务

![format_png 2][]

![format_png 3][]

[format_png]: https://img-blog.csdnimg.cn/img_convert/d332328864f15ceffae61296af20febf.webp?x-oss-process=image/format,png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/1fa0271e2e5ca3f37f0df4ac89d62e58.webp?x-oss-process=image/format,png
[format_png 2]: https://img-blog.csdnimg.cn/img_convert/a52c5e7439913177c8e0c34fc9f194f8.webp?x-oss-process=image/format,png
[format_png 3]: https://img-blog.csdnimg.cn/img_convert/024aa36058164d72ac30a3e4ca5b5b01.webp?x-oss-process=image/format,png