在 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

@Configuration
public class ScheduleConfig {

    // 用来存入线程执行情况, 方便于停止定时任务时使用
    public static ConcurrentHashMap<String, ScheduledFuture> map = new ConcurrentHashMap<String, ScheduledFuture>();

    @Bean
    public ThreadPoolTaskScheduler threadPoolTaskScheduler(){
        ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
        threadPoolTaskScheduler.setPoolSize(10);                        // 线程池大小
        threadPoolTaskScheduler.setThreadNamePrefix("taskExecutor-");   // 线程名称
        threadPoolTaskScheduler.setAwaitTerminationSeconds(60);         // 等待时长
        threadPoolTaskScheduler.setWaitForTasksToCompleteOnShutdown(true);  // 调度器shutdown被调用时等待当前被调度的任务完成
        return threadPoolTaskScheduler;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

第二步

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

@RestController
public class TestController {

    @Autowired
    private DynamicTask task;

    @RequestMapping("start")
    public void startTask() {
        com.demo.task.startCron();
    }

    // 测试访问： http://localhost:8080/stopById?taskId=任务一
    @RequestMapping("stopById")
    public void stopById(String taskId) {
        com.demo.task.stop(taskId);
    }

    @RequestMapping("stopAll")
    public void stopAll() {
        com.demo.task.stopAll();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

第三步

核心逻辑

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

• threadPoolTaskScheduler.schedule(com.demo.task, cron) 用来调度任务

• boolean cancelled = scheduledFuture.isCancelled(); 用来判断是否已经取消

• scheduledFuture.cancel(true) 用来将当前的任务取消

下面是核心代码逻辑：

@Component
public class DynamicTask {
    private final static Logger logger = LoggerFactory.getLogger(DynamicTask.class);

    @Autowired
    private ThreadPoolTaskScheduler threadPoolTaskScheduler;    // 注入线程池任务调度类

    public void startCron(){

        getTasks().forEach(customizeTask -> {
            // 开始执行调度
            ScheduledFuture scheduledFuture = threadPoolTaskScheduler.schedule(customizeTask, new CronTrigger(customizeTask.getCron()));
            // 将 scheduledFuture 保存下来用于停止任务使用
            ScheduleConfig.cache.put(customizeTask.getName(), scheduledFuture);
        });
    }

    public void stop(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);        // 移除缓存
        }
    }

    public void stopAll(){
        if (ScheduleConfig.cache.isEmpty()) return;
        ScheduleConfig.cache.values().forEach(scheduledFuture -> scheduledFuture.cancel(true) );
    }

    private List<CustomizeTask> getTasks(){
        return Arrays.asList(new CustomizeTask("任务一", "0/2 * * * * ?"),
                new CustomizeTask("任务二", "0/3 * * * * ?"));
    }

    // 自定义任务，这里用来对任务进行封装
    private class CustomizeTask implements Runnable {
        private String name;    // 任务名字
        private String cron;    // 触发条件

        CustomizeTask(String name, String cron) {
            this.name = name;
            this.cron = cron;
        }

        public String getCron(){
            return this.cron;
        }

        public String getName(){
            return this.name;
        }

        @Override
        public void run() {
            logger.info("当前任务名称：{}", name );
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

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

private class CustomizeTask implements Runnable {
    private String name;    // 任务名字
    private String cron;    // 触发条件
    private String data;    // 传输的数据参数
    private String method;  // 需要调用的方法
    
    CustomizeTask(String name, String cron, String data, String method) {
        this.name = name;
        this.cron = cron;
        this.data = data;
        this.method = method;
    }

    public String getCron(){
        return this.cron;
    }

    public String getName(){
        return this.name;
    }

    @Override
    public void run() {
        // 通过反射获取到调用的方法，传入调用的参数 data
        logger.info("当前任务名称：{}", name );
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

测试

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

http://localhost:8080/start
1

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

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

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

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

http://localhost:8080/stopAll
1

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