基于SpringBoot实现轻量级的动态定时任务调度的方法

涂流如

目录

• 设计思路
• 整体设计
• 任务模型
• 任务对象加载过程
• 部分源码
• 效果

在使用SpringBoot框架进行开发时，一般都是通过@Scheduled注解进行定时任务的开发：

2. @Component
3. public class TestTask
4. {
5. @Scheduled(cron="0/5 * * * * ? ") //每5秒执行一次
6. public void execute(){
7. SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
8. log.info("任务执行" + df.format(new Date()));
9. }
10. }

复制代码

但是这种方式存在一个问题，那就是任务的周期控制是死的，必须编写在代码中，如果遇到需要在系统运行过程中想中止、立即执行、修改执行周期等动态操作的需求时，使用注解的方式便不能满足了，当然为了满足此种需求可以额外再引入其他任务调度插件（例如XXL-Job等），但是引入其他组件是需要衡量成本的，额外的依赖成本、组件的维护成本、开发的复杂度等等，所以如果系统体量不是那么大，完全没必要通过增加组件来完成，可以基于SpringBoot框架实现一套内置轻量级的任务调度。

设计思路

整体设计

这里我们把定时任务以类作为基础单位，即一个类为一个任务，然后通过配置数据的方式，进行任务的读取，通过反射生成任务对象，使用SpringBoot本身的线程池任务调度，完成动态的定时任务驱动，同时通过接口支撑实现相应的REST API对外暴露接口

任务模型

首先基于模板模式，设计基础的任务执行流程抽象类，定义出一个定时任务需要执行的内容和步骤和一些通用的方法函数，后续具体的定时任务直接继承该父类，实现该父类的before、start、after三个抽象函数即可，所有公共操作均在抽象父类完成

特殊说明：

    基于此方法创建的类是不归Spring的容器管理的，所以自定义的任务子类中是无法使用SpringBoot中的任何注解，尤其在自定义任务类中如果需要依赖其他Bean时，需要借助抽象父类AbstractBaseCronTask中已经实现的 T getServer(Class className)来完成，getServer的实现如下：

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

复制代码

是通过SpringBoot中的ApplicationContext接口来获取Spring的上下文，以此来满足可以获取Spring中其他Bean的诉求。

例如，有个定时任务TaskOne类，它需要使用UserService类中的 caculateMoney()的方法，势必这个定时任务需要依赖UserService类，而TaskOne并非是Spring创建的对象，而是我们人为干预生成的对象，所以它是不在Spring的Bean管理范围的，自然也就无法使用@Autowird等方式注入UserService类，此时就需要使用getServer方法来获取UserService对象

1. //自定义定时任务类
2. public class TaskOne extends AbstractBaseCronTask {
3. private UserService userService;
4. public TestTask(TaskEntity taskEntity) {
5. super(taskEntity);
6. }
7. @Override
8. public void beforeJob() {
9. //任务运行第一步，先将userService进行变量注入
10. userService = getServer(UserService.class);
11. ……
12. }
13. @Override
14. public void startJob() {
15. if(XXXX){
16. //直接调用getServer获取需要的bean
17. User user = getServer(UserMapper.class).findUser("111223")
18. userService.caluateMoney(user);
19. //……其他代码
20. }
21. }
22. @Override
23. public void afterJob() {
24. }
25. }

复制代码

任务对象加载过程

 核心逻辑在于利用反射，在SpringBoot启动后动态创建相应的定时任务类，并将其放置到SpringBoot的定时线程池中进行维护，同时将该对象同步存放至内存中一份，便于可以实时调用，当进行修改任务相关配置时，需要重新加载一次内容。

1. public class TaskScheduleServerImpl implements TaskScheduleServer {
2. //正在运行的任务
3. private static ConcurrentHashMap<String, ScheduledFuture> runningTasks = new ConcurrentHashMap<>();
4. //线程池任务调度
5. private ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
6. public boolean addTaskToScheduling(TaskEntity task) {
7. if(!runningTasks.containsKey(task.getTaskId())){
8. try{
9. Class<?> clazz = Class.forName(task.getTaskClass());
10. Constructor c = clazz.getConstructor(TaskEntity.class);
11. AbstractBaseCronTask runnable = (AbstractBaseCronTask) c.newInstance(task);
12. //反射方式生成对象不属于Spring容器管控，对于Spring的bean使用需要手动注入
13. runnable.setApplicationContext(context);
14. CronTrigger cron = new CronTrigger(task.getTaskCron());
15. //put到runTasks
16. runningTasks.put(task.getTaskId(), Objects.requireNonNull(this.threadPoolTaskScheduler.schedule(runnable, cron)));
17. //存入内存中，便于外部调用
18. ramTasks.put(task.getTaskId(),runnable);
19. task.setTaskRamStatus(1);
20. taskInfoOpMapper.updateTaskInfo(task);
21. return true;
22. }catch (Exception e){
23. log.error("定时任务加载失败..."+e);
24. }
25. }
26. return false;
27. }
28. }

复制代码

部分源码

这里将配置内容放入数据库中，直接以数据库中的表作为任务配置的基础

1. /**
2. * 任务对象
3. **/
4. @Data
5. public class TaskEntity implements Serializable {
6. //任务唯一ID
7. private String taskId;
8. //任务名称
9. private String taskName;
10. //任务描述
11. private String taskDesc;
12. //执行周期配置
13. private String taskCron;
14. //任务类的全路径
15. private String taskClass;
16. //任务的额外配置
17. private String taskOutConfig;
18. //任务创建时间
19. private String taskCreateTime;
20. //任务是否启动，1启用，0不启用
21. private Integer taskIsUse;
22. //是否随系统启动立即执行
23. private Integer taskBootUp;
24. //任务上次执行状态
25. private Integer taskLastRun;
26. //任务是否加载至内存中
27. private Integer taskRamStatus;
28. }

复制代码

核心逻辑，加载定时任务接口及其实现类

1. public interface TaskScheduleServer {
2. ConcurrentHashMap<String, AbstractBaseCronTask> getTaskSchedulingRam();
3. /**
4. * 初始化任务调度
5. */
6. void initScheduling();
7. /**
8. * 添加任务至内存及容器
9. * @param taskEntity 任务实体
10. * @return boolean
11. */
12. boolean addTaskToScheduling(TaskEntity taskEntity);
13. /**
14. * 从任务调度器中移除任务
15. * @param id 任务id
16. * @return Boolean
17. */
18. boolean removeTaskFromScheduling(String id);
19. /**
20. * 执行指定任务
21. * @param id 任务id
22. * @return double 耗时
23. */
24. double runTaskById(String id);
25. /**
26. * 清空任务
27. */
28. void claearAllTask();
29. /**
30. * 加载所有任务
31. */
32. void loadAllTask();
33. /**
34. * 运行开机自启任务
35. */
36. void runBootUpTask();
37. }
38. @Slf4j
39. @Component
40. public class TaskScheduleServerImpl implements TaskScheduleServer {
41. …………
42. @Override
43. public double runTaskById(String id) {
44. TaskEntity task = taskInfoOpMapper.queryTaskInfoById(id);
45. if(null!=task) {
46. if (runningTasks.containsKey(task.getTaskId())){
47. ramTasks.get(task.getTaskId()).run();
48. return ramTasks.get(task.getTaskId()).getRunTime();
49. }
50. }
51. return 0d;
52. }
53. @Override
54. public void claearAllTask() {
55. ramTasks.clear();
56. log.info("【定时任务控制器】清除内存任务 完成");
57. runningTasks.clear();
58. log.info("【定时任务控制器】清除线程任务 完成");
59. threadPoolTaskScheduler.shutdown();
60. }
61. @Override
62. public void loadAllTask() {
63. List<TaskEntity> allTask = taskInfoOpMapper.queryTaskInfo(null);
64. for (TaskEntity task : allTask) {
65. if(addTaskToScheduling(task)){
66. log.info("【定时任务初始化】装填任务:{} [ 任务执行周期：{} ] [ bootup：{}]",task.getTaskName(),task.getTaskCron(),task.getTaskBootUp());
67. }
68. }
69. }
70. @Override
71. public void runBootUpTask() {
72. TaskEntity entity = new TaskEntity().taskBootUp(1);
73. List<TaskEntity> list = taskInfoOpMapper.queryTaskInfo(entity);
74. for(TaskEntity task:list){
75. runTaskById(task.getTaskId());
76. }
77. }
78. }

复制代码

在SpringBoot中的加载类

1. @Order(3)
2. @Component
3. @Slf4j
4. public class AfterAppStarted implements ApplicationRunner {
5. TaskScheduleServer taskScheduleServer;
6. @Autowired
7. public void setTaskScheduleServer(TaskScheduleServer taskScheduleServer) {
8. this.taskScheduleServer = taskScheduleServer;
9. }
10. @Override
11. public void run(ApplicationArguments args) throws Exception {
12. //运行随系统启动的定时任务
13. taskScheduleServer.runBootUpTask();
14. }
15. }

复制代码

对外暴露控制接口及其Service

1. @RestController
2. @RequestMapping("/taskScheduling/manage")
3. @Api(tags = "数据源管理服务")
4. public class TaskSchedulingController {
5. TaskScheduleManagerService taskScheduleManagerService;
6. @Autowired
7. public void setTaskScheduleManagerService(TaskScheduleManagerService taskScheduleManagerService) {
8. this.taskScheduleManagerService = taskScheduleManagerService;
9. }
10. @PostMapping("/search")
11. @Operation(summary = "分页查询任务")
12. public Response searchData(@RequestBody SearchTaskDto param){
13. return Response.success(taskScheduleManagerService.searchTaskForPage(param));
14. }
15. @GetMapping("/detail")
16. @Operation(summary = "具体任务对象")
17. public Response searchDetail(String taskId){
18. return Response.success(taskScheduleManagerService.searchTaskDetail(taskId));
19. }
20. @GetMapping("/shutdown")
21. @Operation(summary = "关闭指定任务")
22. public Response shutdownTask(String taskId){
23. return Response.success(taskScheduleManagerService.shutdownTask(taskId));
24. }
25. @GetMapping("/open")
26. @Operation(summary = "开启指定任务")
27. public Response openTask(String taskId){
28. return Response.success(taskScheduleManagerService.openTask(taskId));
29. }
30. @GetMapping("/run")
31. @Operation(summary = "运行指定任务")
32. public Response runTask(String taskId){
33. return Response.success(taskScheduleManagerService.runTask(taskId));
34. }
35. @PostMapping("/update")
36. @Operation(summary = "更新指定任务")
37. public Response updateTask(@RequestBody TaskEntity taskEntity){
38. return Response.success(taskScheduleManagerService.updateTaskBusinessInfo(taskEntity));
39. }
40. }

复制代码

相关接口实现类

1. @Service
2. public class TaskScheduleManagerServiceImpl implements TaskScheduleManagerService {
3. private TaskInfoOpMapper taskInfoOpMapper;
4. private TaskScheduleServer taskScheduleServer;
5. @Autowired
6. public void setTaskInfoOpMapper(TaskInfoOpMapper taskInfoOpMapper) {
7. this.taskInfoOpMapper = taskInfoOpMapper;
8. }
9. @Autowired
10. public void setTaskScheduleServer(TaskScheduleServer taskScheduleServer) {
11. this.taskScheduleServer = taskScheduleServer;
12. }
13. @Override
14. public IPage<TaskEntity> searchTaskForPage(SearchTaskDto dto) {
15. Page<TaskEntity> pageParam = new Page<>(1,10);
16. pageParam.setAsc("task_id");
17. return taskInfoOpMapper.queryTaskInfoPage(pageParam,dto.getFilterKey(),dto.getBootUp(),dto.getLastRunStatus());
18. }
19. @Override
20. public TaskEntity searchTaskDetail(String taskId) {
21. if(!StringUtils.isEmpty(taskId)){
22. return taskInfoOpMapper.queryTaskInfoById(taskId);
23. }
24. return null;
25. }
26. @Override
27. public TaskRunRetDto runTask(String taskId) {
28. AbstractBaseCronTask task = taskScheduleServer.getTaskSchedulingRam().get(taskId);
29. TaskRunRetDto result = new TaskRunRetDto(TaskRunRetDto.TaskOperation.run, 0);
30. if(null != task) {
31. double time = taskScheduleServer.runTaskById(taskId);
32. result.setResult(1);
33. return result.extend(time).taskInfo(task.getThisTaskInfo());
34. } else {
35. return result.extend("任务未启用");
36. }
37. }
38. @Override
39. public TaskRunRetDto shutdownTask(String taskId) {
40. AbstractBaseCronTask task = taskScheduleServer.getTaskSchedulingRam().get(taskId);
41. TaskRunRetDto result = new TaskRunRetDto(TaskRunRetDto.TaskOperation.shutdown, 0);
42. if(null != task) {
43. boolean flag = taskScheduleServer.removeTaskFromScheduling(taskId);
44. if(flag) {
45. result.setResult(1);
46. }
47. return result.extend("任务成功关闭").taskInfo(task.getThisTaskInfo());
48. } else {
49. return result.extend("任务未启用");
50. }
51. }
52. @Override
53. public TaskRunRetDto openTask(String taskId) {
54. TaskEntity task = taskInfoOpMapper.queryTaskInfoById(taskId);
55. TaskRunRetDto result = new TaskRunRetDto(TaskRunRetDto.TaskOperation.open, 0);
56. if(null != task) {
57. if (!taskScheduleServer.getTaskSchedulingRam().containsKey(taskId)) {
58. boolean flag = taskScheduleServer.addTaskToScheduling(task);
59. if(flag) {
60. result.setResult(1);
61. }
62. return result.extend("任务开启成功").taskInfo(task);
63. } else {
64. return result.extend("任务处于启动状态").taskInfo(task);
65. }
66. }else {
67. return result.extend("任务不存在！");
68. }
69. }
70. @Override
71. public TaskRunRetDto updateTaskBusinessInfo(TaskEntity entity) {
72. TaskEntity task = searchTaskDetail(entity.getTaskId());
73. TaskRunRetDto result = new TaskRunRetDto(TaskRunRetDto.TaskOperation.update, 0).taskInfo(entity);
74. String config = entity.getTaskOutConfig();
75. if(null != config && !JSONUtil.isJson(config) && !JSONUtil.isJsonArray(config)){
76. result.setResult(0);
77. result.extend("更新任务失败,任务配置必须为JSON或空");
78. result.taskInfo(entity);
79. return result;
80. }
81. task.setTaskCron(entity.getTaskCron());
82. task.setTaskOutConfig(entity.getTaskOutConfig());
83. task.setTaskName(entity.getTaskName());
84. task.setTaskDesc(entity.getTaskDesc());
85. int num = taskInfoOpMapper.updateTaskInfo(task);
86. if (num == 1) {
87. result.setResult(1);
88. result.extend("成功更新任务");
89. result.taskInfo(entity);
90. //重新刷新任务
91. taskScheduleServer.removeTaskFromScheduling(entity.getTaskId());
92. taskScheduleServer.addTaskToScheduling(task);
93. }
94. return result;
95. }

复制代码

效果

数据库中配置任务

任务代码

1. public class TestTask extends AbstractBaseCronTask {
2. public TestTask(TaskEntity taskEntity) {
3. super(taskEntity);
4. }
5. @Override
6. public void beforeJob() {
7. log.info("测试任务开始");
8. }
9. @Override
10. public void startJob() {
11. }
12. @Override
13. public void afterJob() {
14. }
15. }

复制代码

任务查看

执行效果

到此这篇关于基于SpringBoot实现轻量级的动态定时任务调度的文章就介绍到这了,更多相关SpringBoot动态定时任务调度内容请搜索晓枫资讯以前的文章或继续浏览下面的相关文章希望大家以后多多支持晓枫资讯！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

张勇530

路过，支持一下

Kun617

顶顶更健康！！！

永恒伤心人杜

感谢楼主，顶。