在 SpringBoot 中使用 ThreadPoolTaskScheduler 实现定时任务

项目工程地址

简介

之前看道同事项目上面使用过 ThreadPoolTaskScheduler 作为定时任务，就简单的看了下没有整理文档，现在就深入研究下，整理下知识点。

ThreadPoolTaskScheduler 这个类是在

org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler

这个包中。

ThreadPoolTaskScheduler 是

spring taskSchedule

接口的实现，可以用来做定时任务使用。

ThreadPoolTaskScheduler 四个版本定时任务方法：

• schedule(Runnable task, Date stateTime)，在指定时间执行一次定时任务
• schedule(Runnable task, Trigger trigger)，动态创建指定表达式cron的定时任务，threadPoolTaskScheduler.schedule(() -> {}, triggerContext -> newCronTrigger("").nextExecutionTime(triggerContext));
• scheduleAtFixedRate，指定间隔时间执行一次任务，间隔时间为前一次执行开始到下次任务开始时间
• scheduleWithFixedDelay，指定间隔时间执行一次任务，间隔时间为前一次任务完成到下一次开始时间

具体实现

说明

我们需要向 spring 容器中注入一个 ThreadPoolTaskScheduler 的 bean，用于调度定时任务，以及需要增加一个缓存用于存入当前执行任务的 scheduleFuture 对象，将 ScheduledFuture 对象缓存的原因是在于，为了方面于停止对应的任务。

我们点击 ThreadPoolTaskScheduler 类，看具体的源码的时候会发现有这么一段断码：

这里是表示初始化的线程池的大小为 1，也就是说线程调度器设置只有一个线程容量，如果存在多个任务被触发时，会等第一个任务执行完毕才会执行下一个任务。所以这里还是需要自己去定义线程的大小，避免因为默认的线程池导致出现奇奇怪怪的问题。

第一步

创建 ThreadPoolTaskScheduler Bean 以及 用于存放定时任务的 map

@ConfigurationpublicclassScheduleConfig{// 用来存入线程执行情况, 方便于停止定时任务时使用publicstaticConcurrentHashMap<String,ScheduledFuture> cache=newConcurrentHashMap<String,ScheduledFuture>();@BeanpublicThreadPoolTaskSchedulerthreadPoolTaskScheduler(){ThreadPoolTaskScheduler threadPoolTaskScheduler =newThreadPoolTaskScheduler();
        threadPoolTaskScheduler.setPoolSize(10);// 线程池大小
        threadPoolTaskScheduler.setThreadNamePrefix("taskExecutor-");// 线程名称
        threadPoolTaskScheduler.setAwaitTerminationSeconds(60);// 等待时长
        threadPoolTaskScheduler.setWaitForTasksToCompleteOnShutdown(true);// 调度器shutdown被调用时等待当前被调度的任务完成return threadPoolTaskScheduler;}}

第二步

增加用于外部访问的接口 controller，等程序启动完成之后，我们需要调用对应的方法进行访问测试

@RestControllerpublicclassTestController{@AutowiredprivateDynamicTask task;@RequestMapping("start")publicvoidstartTask(){
        com.demo.task.startCron();}// 测试访问： http://localhost:8080/stopById?taskId=任务一@RequestMapping("stopById")publicvoidstopById(String taskId){
        com.demo.task.stop(taskId);}@RequestMapping("stopAll")publicvoidstopAll(){
        com.demo.task.stopAll();}}

第三步

核心逻辑

在这里当在执行

threadPoolTaskScheduler.schedule()

时，会传入一个自定义的 com.demo.task，以及一个 trigger。
调用完成之后会返回一个 scheduledFuture，这个就是当前的任务调度器，停止的时候需要找到这个调度器，用这个调用器来终止。

• threadPoolTaskScheduler.schedule(com.demo.task, cron) 用来调度任务
• boolean cancelled = scheduledFuture.isCancelled(); 用来判断是否已经取消
• scheduledFuture.cancel(true) 用来将当前的任务取消

下面是核心代码逻辑：

@ComponentpublicclassDynamicTask{privatefinalstaticLogger logger =LoggerFactory.getLogger(DynamicTask.class);@AutowiredprivateThreadPoolTaskScheduler threadPoolTaskScheduler;// 注入线程池任务调度类publicvoidstartCron(){getTasks().forEach(customizeTask ->{// 开始执行调度ScheduledFuture scheduledFuture = threadPoolTaskScheduler.schedule(customizeTask,newCronTrigger(customizeTask.getCron()));// 将 scheduledFuture 保存下来用于停止任务使用ScheduleConfig.cache.put(customizeTask.getName(), scheduledFuture);});}publicvoidstop(String taskId){if(ScheduleConfig.cache.isEmpty())return;if(ScheduleConfig.cache.get(taskId)==null)return;ScheduledFuture scheduledFuture =ScheduleConfig.cache.get(taskId);if(scheduledFuture !=null){
            scheduledFuture.cancel(true);// 这里需要使用指定的 scheduledFuture 来停止当前的线程ScheduleConfig.cache.remove(taskId);// 移除缓存}}publicvoidstopAll(){if(ScheduleConfig.cache.isEmpty())return;ScheduleConfig.cache.values().forEach(scheduledFuture -> scheduledFuture.cancel(true));}privateList<CustomizeTask>getTasks(){returnArrays.asList(newCustomizeTask("任务一","0/2 * * * * ?"),newCustomizeTask("任务二","0/3 * * * * ?"));}// 自定义任务，这里用来对任务进行封装privateclassCustomizeTaskimplementsRunnable{privateString name;// 任务名字privateString cron;// 触发条件CustomizeTask(String name,String cron){this.name = name;this.cron = cron;}publicStringgetCron(){returnthis.cron;}publicStringgetName(){returnthis.name;}@Overridepublicvoidrun(){
            logger.info("当前任务名称：{}", name );}}}

对于上面的任务类

CustomizeTask

可以自定义进行封装，可以增加成员属性调用的 服务方法类 以及 调用参数，在 run 方法上面就可以去调用对应的服务方法了

privateclassCustomizeTaskimplementsRunnable{privateString name;// 任务名字privateString cron;// 触发条件privateString data;// 传输的数据参数privateString method;// 需要调用的方法CustomizeTask(String name,String cron,String data,String method){this.name = name;this.cron = cron;this.data = data;this.method = method;}publicStringgetCron(){returnthis.cron;}publicStringgetName(){returnthis.name;}@Overridepublicvoidrun(){// 通过反射获取到调用的方法，传入调用的参数 data
        logger.info("当前任务名称：{}", name );}}

测试

当源码搭建完成之后，启动服务，我们测试下，首先我们先访问下启动任务

http://localhost:8080/start

这里可以看到控制台交替输出任务的名称

接着，我们再来测试下停止其中的某个服务，当访问下面的链接之后，会发现任务一就已经被停止了

http://localhost:8080/stopById?taskId=任务一

最后，我们访问下停止所有的任务：

http://localhost:8080/stopAll

就会发现控制台没有任务在执行了。