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

新增排程任務

删除排程任務