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

SpringBoot ScheduledTaskRegistrar SpringBoot定时任务动态扩展 2023-01-12 18:01:29 946人浏览薄情痞子

Python 官方文档：入门教程 => 点击学习

摘要

目录摘要ScheduledTaskReGIStrar类简要描述平常使用方式配置原理分析DynamicScheduledTaskRegistrar 动态任务注册类线程池数量问题新增调度

摘要

本文主要介绍基于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&lt;String,ScheduledTask&gt; scheduledTaskMap = new LinkedHashMap&lt;&gt;(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);
    }
    
    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;
    }
    
    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();
    
    public void add(String taskName,String cron){
        Boolean result = dynamicScheduledTaskRegistrar.addCronTask(taskName,cron,() -> print(taskName));
        log.info("定时任务添加结果：" + result);
    }
    
    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";
    }
}

测试结果