基于SpringBoot實現輕量級的動態定時任務調度

在使用SpringBoot框架進行開發時，一般都是通過@Scheduled注解進行定時任務的開發：


@Component
public class TestTask
{@Scheduled(cron="0/5 * *  * * ? ")   //每5秒執行一次public void execute(){SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); log.info("任務執行" + df.format(new Date()));}
}

但是這種方式存在一個問題，那就是任務的周期控制是死的，必須編寫在代碼中，如果遇到需要在系統運行過程中想中止、立即執行、修改執行周期等動態操作的需求時，使用注解的方式便不能滿足了，當然為了滿足此種需求可以額外再引入其他任務調度插件（例如XXL-Job等），但是引入其他組件是需要衡量成本的，額外的依賴成本、組件的維護成本、開發的復雜度等等，所以如果系統體量不是那么大，完全沒必要通過增加組件來完成，可以基于SpringBoot框架實現一套內置輕量級的任務調度。

設計思路

整體設計

這里我們把定時任務以類作為基礎單位，即一個類為一個任務，然后通過配置數據的方式，進行任務的讀取，通過反射生成任務對象，使用SpringBoot本身的線程池任務調度，完成動態的定時任務驅動，同時通過接口支撐實現相應的REST API對外暴露接口

任務模型

首先基于模板模式，設計基礎的任務執行流程抽象類，定義出一個定時任務需要執行的內容和步驟和一些通用的方法函數，后續具體的定時任務直接繼承該父類，實現該父類的before、start、after三個抽象函數即可，所有公共操作均在抽象父類完成

特殊說明：

? ? 基于此方法創建的類是不歸Spring的容器管理的，所以自定義的任務子類中是無法使用SpringBoot中的任何注解，尤其在自定義任務類中如果需要依賴其他Bean時，需要借助抽象父類AbstractBaseCronTask中已經實現的<T> T getServer(Class<T> className)來完成，getServer的實現如下：

public <T> T getServer(Class<T> className){return applicationContext.getBean(className);}

是通過SpringBoot中的ApplicationContext接口來獲取Spring的上下文，以此來滿足可以獲取Spring中其他Bean的訴求。

例如，有個定時任務TaskOne類，它需要使用UserService類中的 caculateMoney()的方法，勢必這個定時任務需要依賴UserService類，而TaskOne并非是Spring創建的對象，而是我們人為干預生成的對象，所以它是不在Spring的Bean管理范圍的，自然也就無法使用@Autowird等方式注入UserService類，此時就需要使用getServer方法來獲取UserService對象

//自定義定時任務類
public class TaskOne extends AbstractBaseCronTask {private UserService userService;public TestTask(TaskEntity taskEntity) {super(taskEntity);}@Overridepublic void beforeJob() {//任務運行第一步，先將userService進行變量注入userService = getServer(UserService.class);……}@Overridepublic void startJob() {if(XXXX){//直接調用getServer獲取需要的beanUser user = getServer(UserMapper.class).findUser("111223")userService.caluateMoney(user);//……其他代碼}}@Overridepublic void afterJob() {}
}

任務對象加載過程

?核心邏輯在于利用反射，在SpringBoot啟動后動態創建相應的定時任務類，并將其放置到SpringBoot的定時線程池中進行維護，同時將該對象同步存放至內存中一份，便于可以實時調用，當進行修改任務相關配置時，需要重新加載一次內容。

public class TaskScheduleServerImpl implements TaskScheduleServer {//正在運行的任務private static ConcurrentHashMap<String, ScheduledFuture> runningTasks = new ConcurrentHashMap<>();//線程池任務調度private ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();public boolean addTaskToScheduling(TaskEntity task) {if(!runningTasks.containsKey(task.getTaskId())){try{Class<?> clazz = Class.forName(task.getTaskClass());Constructor c = clazz.getConstructor(TaskEntity.class);AbstractBaseCronTask runnable = (AbstractBaseCronTask) c.newInstance(task);//反射方式生成對象不屬于Spring容器管控，對于Spring的bean使用需要手動注入runnable.setApplicationContext(context);CronTrigger cron = new CronTrigger(task.getTaskCron());//put到runTasksrunningTasks.put(task.getTaskId(), Objects.requireNonNull(this.threadPoolTaskScheduler.schedule(runnable, cron)));//存入內存中，便于外部調用ramTasks.put(task.getTaskId(),runnable);task.setTaskRamStatus(1);taskInfoOpMapper.updateTaskInfo(task);return true;}catch (Exception e){log.error("定時任務加載失敗..."+e);}}return false;}
}

部分源碼

這里將配置內容放入數據庫中，直接以數據庫中的表作為任務配置的基礎

/**
* 任務對象
**/
@Data
public class TaskEntity implements Serializable {//任務唯一IDprivate String taskId;//任務名稱private String taskName;//任務描述private String taskDesc;//執行周期配置private String taskCron;//任務類的全路徑private String taskClass;//任務的額外配置private String taskOutConfig;//任務創建時間private String taskCreateTime;//任務是否啟動，1啟用，0不啟用private Integer taskIsUse;//是否隨系統啟動立即執行private Integer taskBootUp;//任務上次執行狀態private Integer taskLastRun;//任務是否加載至內存中 private Integer taskRamStatus;
}

核心邏輯，加載定時任務接口及其實現類

public interface TaskScheduleServer {ConcurrentHashMap<String, AbstractBaseCronTask> getTaskSchedulingRam();/*** 初始化任務調度*/void initScheduling();/*** 添加任務至內存及容器* @param taskEntity 任務實體* @return boolean*/boolean addTaskToScheduling(TaskEntity taskEntity);/*** 從任務調度器中移除任務* @param id 任務id* @return Boolean*/boolean removeTaskFromScheduling(String id);/*** 執行指定任務* @param id 任務id* @return double 耗時*/double runTaskById(String id);/*** 清空任務*/void claearAllTask();/*** 加載所有任務*/void loadAllTask();/*** 運行開機自啟任務*/void runBootUpTask();}@Slf4j
@Component
public class TaskScheduleServerImpl implements TaskScheduleServer {…………@Overridepublic double runTaskById(String id) {TaskEntity task = taskInfoOpMapper.queryTaskInfoById(id);if(null!=task) {if (runningTasks.containsKey(task.getTaskId())){ramTasks.get(task.getTaskId()).run();return ramTasks.get(task.getTaskId()).getRunTime();}}return 0d;}@Overridepublic void claearAllTask() {ramTasks.clear();log.info("【定時任務控制器】清除內存任務 完成");runningTasks.clear();log.info("【定時任務控制器】清除線程任務 完成");threadPoolTaskScheduler.shutdown();}@Overridepublic void loadAllTask() {List<TaskEntity> allTask = taskInfoOpMapper.queryTaskInfo(null);for (TaskEntity task : allTask) {if(addTaskToScheduling(task)){log.info("【定時任務初始化】裝填任務:{} [ 任務執行周期：{} ] [ bootup：{}]",task.getTaskName(),task.getTaskCron(),task.getTaskBootUp());}}}@Overridepublic void runBootUpTask() {TaskEntity entity = new TaskEntity().taskBootUp(1);List<TaskEntity> list = taskInfoOpMapper.queryTaskInfo(entity);for(TaskEntity task:list){runTaskById(task.getTaskId());}}
}

在SpringBoot中的加載類

@Order(3)
@Component
@Slf4j
public class AfterAppStarted implements ApplicationRunner {TaskScheduleServer taskScheduleServer;@Autowiredpublic void setTaskScheduleServer(TaskScheduleServer taskScheduleServer) {this.taskScheduleServer = taskScheduleServer;}@Overridepublic void run(ApplicationArguments args) throws Exception {//運行隨系統啟動的定時任務taskScheduleServer.runBootUpTask();}}

對外暴露控制接口及其Service

@RestController
@RequestMapping("/taskScheduling/manage")
@Api(tags = "數據源管理服務")
public class TaskSchedulingController {TaskScheduleManagerService taskScheduleManagerService;@Autowiredpublic void setTaskScheduleManagerService(TaskScheduleManagerService taskScheduleManagerService) {this.taskScheduleManagerService = taskScheduleManagerService;}@PostMapping("/search")@Operation(summary = "分頁查詢任務")public Response searchData(@RequestBody SearchTaskDto param){return Response.success(taskScheduleManagerService.searchTaskForPage(param));}@GetMapping("/detail")@Operation(summary = "具體任務對象")public Response searchDetail(String taskId){return Response.success(taskScheduleManagerService.searchTaskDetail(taskId));}@GetMapping("/shutdown")@Operation(summary = "關閉指定任務")public Response shutdownTask(String taskId){return Response.success(taskScheduleManagerService.shutdownTask(taskId));}@GetMapping("/open")@Operation(summary = "開啟指定任務")public Response openTask(String taskId){return Response.success(taskScheduleManagerService.openTask(taskId));}@GetMapping("/run")@Operation(summary = "運行指定任務")public  Response runTask(String taskId){return Response.success(taskScheduleManagerService.runTask(taskId));}@PostMapping("/update")@Operation(summary = "更新指定任務")public Response updateTask(@RequestBody TaskEntity taskEntity){return Response.success(taskScheduleManagerService.updateTaskBusinessInfo(taskEntity));}}

相關接口實現類

@Service
public class TaskScheduleManagerServiceImpl implements TaskScheduleManagerService {private TaskInfoOpMapper taskInfoOpMapper;private TaskScheduleServer taskScheduleServer;@Autowiredpublic void setTaskInfoOpMapper(TaskInfoOpMapper taskInfoOpMapper) {this.taskInfoOpMapper = taskInfoOpMapper;}@Autowiredpublic void setTaskScheduleServer(TaskScheduleServer taskScheduleServer) {this.taskScheduleServer = taskScheduleServer;}@Overridepublic IPage<TaskEntity> searchTaskForPage(SearchTaskDto dto) {Page<TaskEntity> pageParam = new Page<>(1,10);pageParam.setAsc("task_id");return taskInfoOpMapper.queryTaskInfoPage(pageParam,dto.getFilterKey(),dto.getBootUp(),dto.getLastRunStatus());}@Overridepublic TaskEntity searchTaskDetail(String taskId) {if(!StringUtils.isEmpty(taskId)){return taskInfoOpMapper.queryTaskInfoById(taskId);}return null;}@Overridepublic TaskRunRetDto runTask(String taskId) {AbstractBaseCronTask task = taskScheduleServer.getTaskSchedulingRam().get(taskId);TaskRunRetDto result = new TaskRunRetDto(TaskRunRetDto.TaskOperation.run, 0);if(null != task) {double time = taskScheduleServer.runTaskById(taskId);result.setResult(1);return result.extend(time).taskInfo(task.getThisTaskInfo());} else {return result.extend("任務未啟用");}}@Overridepublic TaskRunRetDto shutdownTask(String taskId) {AbstractBaseCronTask task = taskScheduleServer.getTaskSchedulingRam().get(taskId);TaskRunRetDto result = new TaskRunRetDto(TaskRunRetDto.TaskOperation.shutdown, 0);if(null != task) {boolean flag = taskScheduleServer.removeTaskFromScheduling(taskId);if(flag) {result.setResult(1);}return result.extend("任務成功關閉").taskInfo(task.getThisTaskInfo());} else {return result.extend("任務未啟用");}}@Overridepublic TaskRunRetDto openTask(String taskId) {TaskEntity task = taskInfoOpMapper.queryTaskInfoById(taskId);TaskRunRetDto result = new TaskRunRetDto(TaskRunRetDto.TaskOperation.open, 0);if(null != task) {if (!taskScheduleServer.getTaskSchedulingRam().containsKey(taskId)) {boolean flag = taskScheduleServer.addTaskToScheduling(task);if(flag) {result.setResult(1);}return result.extend("任務開啟成功").taskInfo(task);} else {return result.extend("任務處于啟動狀態").taskInfo(task);}}else {return result.extend("任務不存在！");}}@Overridepublic TaskRunRetDto updateTaskBusinessInfo(TaskEntity entity) {TaskEntity task = searchTaskDetail(entity.getTaskId());TaskRunRetDto result = new TaskRunRetDto(TaskRunRetDto.TaskOperation.update, 0).taskInfo(entity);String config = entity.getTaskOutConfig();if(null != config && !JSONUtil.isJson(config) && !JSONUtil.isJsonArray(config)){result.setResult(0);result.extend("更新任務失敗,任務配置必須為JSON或空");result.taskInfo(entity);return result;}task.setTaskCron(entity.getTaskCron());task.setTaskOutConfig(entity.getTaskOutConfig());task.setTaskName(entity.getTaskName());task.setTaskDesc(entity.getTaskDesc());int num = taskInfoOpMapper.updateTaskInfo(task);if (num == 1) {result.setResult(1);result.extend("成功更新任務");result.taskInfo(entity);//重新刷新任務taskScheduleServer.removeTaskFromScheduling(entity.getTaskId());taskScheduleServer.addTaskToScheduling(task);}return result;}

效果

數據庫中配置任務

任務代碼

public class TestTask extends AbstractBaseCronTask {public TestTask(TaskEntity taskEntity) {super(taskEntity);}@Overridepublic void beforeJob() {log.info("測試任務開始");}@Overridepublic void startJob() {}@Overridepublic void afterJob() {}
}

任務查看