Spring SchedulingConfigurer 实现动态定时任务

柔情只为你懂 2022-11-15 11:53 348阅读 0赞

![4645e47d5f7baa7efa8e6d77c914c9da.gif][]

点击上方蓝字，关注我们

![48218734dc9322e5b2f22ed5dd0ba301.gif][]

![4ad8fb554c130b15fe4be697741dc81a.png][]

##  ##

一、前言

大家在日常工作中，一定使用过 Spring 的 `@Scheduled` 注解吧，通过该注解可以非常方便的帮助我们实现任务的定时执行。

但是该注解是不支持运行时动态修改执行间隔的，不知道你在业务中有没有这些需求和痛点：

•在服务运行时能够动态修改定时任务的执行频率和执行开关，而无需重启服务和修改代码•能够基于配置，在不同环境/机器上，实现定时任务执行频率的差异化

这些都可以通过 Spring 的 `SchedulingConfigurer` 注解来实现。

这个注解其实大家并不陌生，如果有使用过 @Scheduled 的话，因为 @Scheduled 默认是单线程执行的，因此如果存在多个任务同时触发，可能触发阻塞。使用 SchedulingConfigurer 可以配置用于执行 @Scheduled 的线程池，来避免这个问题。

@Configuration
    public class ScheduleConfig implements SchedulingConfigurer {
        @Override
        public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
            //设定一个长度10的定时任务线程池
            taskRegistrar.setScheduler(Executors.newScheduledThreadPool(10));
        }
    }

但其实这个接口，还可以实现动态定时任务的功能，下面来演示如何实现。

##  ##

二、功能实现

> 后续定义的类开头的 `DS` 是 `Dynamic Schedule` 的缩写。

使用到的依赖，除了 Spring 外，还包括：

<dependency>
        <groupId>org.apache.commons</groupId>
        <artifactId>commons-lang3</artifactId>
    </dependency>
    
    <dependency>
        <groupId>org.apache.commons</groupId>
        <artifactId>commons-collections4</artifactId>
        <version>4.4</version>
    </dependency>
    
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <scope>provided</scope>
        <version>1.18.18</version>
    </dependency>

### 2.1 @EnableScheduling ###

首先需要开启 `@EnableScheduling` 注解，直接在启动类添加即可：

@EnableScheduling
    @SpringBootApplication
    public class DSApplication {
        public static void main(String[] args) {
            SpringApplication.run(DSApplication.class, args);
        }
    }

### 2.2 IDSTaskInfo ###

定义一个任务信息的接口，后续所有用于动态调整的任务信息对象，都需要实现该接口。

•`id`：该任务信息的唯一 ID，用于唯一标识一个任务•`cron`：该任务执行的 cron 表达式。•`isValid`：任务开关•`isChange`：用于标识任务参数是否发生了改变

public interface IDSTaskInfo {
        /**
         * 任务 ID
         */
        long getId();
    
        /**
         * 任务执行 cron 表达式
         */
        String getCron();
    
        /**
         * 任务是否有效
         */
        boolean isValid();
    
        /**
         * 判断任务是否发生变化
         */
        boolean isChange(IDSTaskInfo oldTaskInfo);
    }

### 2.3 DSContainer ###

顾名思义，是存放 IDSTaskInfo 的容器。

具有以下成员变量：

•`scheduleMap`：用于暂存 IDSTaskInfo 和实际任务 ScheduledTask 的映射关系。其中：

•task\_id：作为主键，确保一个 IDSTaskInfo 只会被注册进一次•T：暂存当初注册时的 IDSTaskInfo，用于跟最新的 IDSTaskInfo 比较参数是否发生变化•ScheduledTask：暂存当初注册时生成的任务，如果需要取消任务的话，需要拿到该对象•Semaphore：确保每个任务实际执行时只有一个线程执行，不会产生并发问题

•`taskRegistrar`：Spring 的任务注册管理器，用于注册任务到 Spring 容器中

具有以下成员方法：

•`void checkTask(final T taskInfo, final TriggerTask triggerTask)`：检查 IDSTaskInfo，判断是否需要注册/取消任务。具体的逻辑包括：

•如果任务已经注册：

•如果任务无效：则取消任务•如果任务有效：

•如果任务配置发生了变化：则取消任务并重新注册任务

•如果任务没有注册：

•如果任务有效：则注册任务

•`Semaphore getSemaphore()`：获取信号量属性。

import lombok.extern.slf4j.Slf4j;
    import org.apache.commons.lang3.tuple.Pair;
    import org.springframework.scheduling.config.ScheduledTask;
    import org.springframework.scheduling.config.ScheduledTaskRegistrar;
    import org.springframework.scheduling.config.TriggerTask;
    
    import java.util.Map;
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.Semaphore;
    
    @Slf4j
    public class DSContainer<T extends IDSTaskInfo> {
        /**
         * IDSTaskInfo和真实任务的关联关系
         *
         * <task_id, <Task, <Scheduled, Semaphore>>>
         */
        private final Map<Long, Pair<T, Pair<ScheduledTask, Semaphore>>> scheduleMap = new ConcurrentHashMap<>();
    
        private final ScheduledTaskRegistrar taskRegistrar;
    
        public DSContainer(ScheduledTaskRegistrar scheduledTaskRegistrar) {
            this.taskRegistrar = scheduledTaskRegistrar;
        }
    
        /**
         * 注册任务
         * @param taskInfo 任务信息
         * @param triggerTask 任务的触发规则
         */
        public void checkTask(final T taskInfo, final TriggerTask triggerTask) {
            final long taskId = taskInfo.getId();
    
            if (scheduleMap.containsKey(taskId)) {
                if (taskInfo.isValid()) {
                    final T oldTaskInfo = scheduleMap.get(taskId).getLeft();
    
                    if(oldTaskInfo.isChange(taskInfo)) {
                        log.info("DSContainer will register again because task config change, taskId: {}", taskId);
                        cancelTask(taskId);
                        registerTask(taskInfo, triggerTask);
                    }
                } else {
                    log.info("DSContainer will cancelTask because task not valid, taskId: {}", taskId);
                    cancelTask(taskId);
                }
            } else {
                if (taskInfo.isValid()) {
                    log.info("DSContainer will registerTask, taskId: {}", taskId);
                    registerTask(taskInfo, triggerTask);
                }
            }
        }
    
        /**
         * 获取 Semaphore，确保任务不会被多个线程同时执行
         */
        public Semaphore getSemaphore(final long taskId) {
            return this.scheduleMap.get(taskId).getRight().getRight();
        }
    
        private void registerTask(final T taskInfo, final TriggerTask triggerTask) {
            final ScheduledTask latestTask = taskRegistrar.scheduleTriggerTask(triggerTask);
            this.scheduleMap.put(taskInfo.getId(), Pair.of(taskInfo, Pair.of(latestTask, new Semaphore(1))));
        }
    
        private void cancelTask(final long taskId) {
            final Pair<T, Pair<ScheduledTask, Semaphore>> pair = this.scheduleMap.remove(taskId);
            if (pair != null) {
                pair.getRight().getLeft().cancel();
            }
        }
    }

### 2.4 AbstractDSHandler ###

下面定义实际的动态线程池处理方法，这里采用抽象类实现，将共用逻辑封装起来，方便扩展。

具有以下抽象方法：

•`ExecutorService getWorkerExecutor()`：提供用于真正执行任务时的线程池。•`List<T> listTaskInfo()`：获取所有的任务信息。•`void doProcess(T taskInfo)`：实现实际执行任务的业务逻辑。

具有以下公共方法：

•`void configureTasks(ScheduledTaskRegistrar taskRegistrar)`：创建 DSContainer 对象，并创建一个单线程的任务定时执行，调用 scheduleTask() 方法处理实际逻辑。•`void scheduleTask()`：首先加载所有任务信息，然后基于 cron 表达式生成 TriggerTask 对象，调用 checkTask() 方法确认是否需要注册/取消任务。当达到执行时间时，调用 execute() 方法，执行任务逻辑。•`void execute(final T taskInfo)`：获取信号量，成功后执行任务逻辑。

import lombok.extern.slf4j.Slf4j;
    import org.apache.commons.collections4.CollectionUtils;
    import org.springframework.scheduling.annotation.SchedulingConfigurer;
    import org.springframework.scheduling.config.ScheduledTaskRegistrar;
    import org.springframework.scheduling.config.TriggerTask;
    import org.springframework.scheduling.support.CronTrigger;
    
    import java.util.List;
    import java.util.Objects;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Semaphore;
    import java.util.concurrent.TimeUnit;
    
    @Slf4j
    public abstract class AbstractDSHandler<T extends IDSTaskInfo> implements SchedulingConfigurer {
    
        private DSContainer<T> dsContainer;
    
        private final String CLASS_NAME = getClass().getSimpleName();
    
        /**
         * 获取用于执行任务的线程池
         */
        protected abstract ExecutorService getWorkerExecutor();
    
        /**
         * 获取所有的任务信息
         */
        protected abstract List<T> listTaskInfo();
    
        /**
         * 做具体的任务逻辑
         */
        protected abstract void doProcess(T taskInfo);
    
        @Override
        public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
            dsContainer = new DSContainer<>(taskRegistrar);
            // 每隔 100ms 调度一次，用于读取所有任务
            taskRegistrar.addFixedDelayTask(this::scheduleTask, 1000);
        }
    
        /**
         * 调度任务，加载所有任务并注册
         */
        private void scheduleTask() {
            CollectionUtils.emptyIfNull(listTaskInfo()).forEach(taskInfo -> 
                    dsContainer.checkTask(taskInfo, new TriggerTask(() -> 
                            this.execute(taskInfo), triggerContext -> new CronTrigger(taskInfo.getCron()).nextExecutionTime(triggerContext)
                    ))
            );
        }
    
        private void execute(final T taskInfo) {
            final long taskId = taskInfo.getId();
    
            try {
                Semaphore semaphore = dsContainer.getSemaphore(taskId);
                if (Objects.isNull(semaphore)) {
                    log.error("{} semaphore is null, taskId: {}", CLASS_NAME, taskId);
                    return;
                }
                if (semaphore.tryAcquire(3, TimeUnit.SECONDS)) {
                    try {
                        getWorkerExecutor().execute(() -> doProcess(taskInfo));
                    } finally {
                        semaphore.release();
                    }
                } else {
                    log.warn("{} too many executor, taskId: {}", CLASS_NAME, taskId);
                }
            } catch (InterruptedException e) {
                log.warn("{} interruptedException error, taskId: {}", CLASS_NAME, taskId);
            } catch (Exception e) {
                log.error("{} execute error, taskId: {}", CLASS_NAME, taskId, e);
            }
        }
    }

##  ##

三、快速测试

至此就完成了动态任务的框架搭建，下面让我们来快速测试下。为了尽量减少其他技术带来的复杂度，本次测试不涉及数据库和真实的定时任务，完全采用模拟实现。

### 3.1 模拟定时任务 ###

为了模拟一个定时任务，我定义了一个 `foo()` 方法，其中只输出一句话。后续我将通过定时调用该方法，来模拟定时任务。

import lombok.extern.slf4j.Slf4j;
    
    import java.time.LocalTime;
    
    @Slf4j
    public class SchedulerTest {
        public void foo() {
            log.info("{} Execute com.github.jitwxs.sample.ds.test.SchedulerTest#foo", LocalTime.now());
        }
    }

### 3.2 实现 IDSTaskInfo ###

首先定义 IDSTaskInfo，我这里想通过反射来实现调用 `foo()` 方法，因此 `reference` 表示的是要调用方法的全路径。另外我实现了 `isChange()` 方法，只要 cron、isValid、reference 发生了变动，就认为该任务的配置发生了改变。

import com.github.jitwxs.sample.ds.config.IDSTaskInfo;
    import lombok.Builder;
    import lombok.Data;
    
    @Data
    @Builder
    public class SchedulerTestTaskInfo implements IDSTaskInfo {
        private long id;
    
        private String cron;
    
        private boolean isValid;
    
        private String reference;
    
        @Override
        public boolean isChange(IDSTaskInfo oldTaskInfo) {
            if(oldTaskInfo instanceof SchedulerTestTaskInfo) {
                final SchedulerTestTaskInfo obj = (SchedulerTestTaskInfo) oldTaskInfo;
                return !this.cron.equals(obj.cron) || this.isValid != obj.isValid || !this.reference.equals(obj.getReference());
            } else {
                throw new IllegalArgumentException("Not Support SchedulerTestTaskInfo type");
            }
        }
    }

### 3.3 实现 AbstractDSHandler ###

有几个需要关注的：

（1）`getWorkerExecutor()` 我随便写了个 2，其实 SchedulerTestTaskInfo 对象只有一个（即调用 foo() 方法的定时任务）。

（2）`listTaskInfo()` 返回值我使用了 volatile 变量，便于我修改它，模拟任务信息数据的改变。

（3）`doProcess()` 方法中，读取到 reference 后，使用反射进行调用，模拟定时任务的执行。

（4）额外实现了 `ApplicationListener` 接口，当服务启动后，每隔一段时间修改下任务信息，模拟业务中调整配置。

•服务启动后，foo() 定时任务将每 10s 执行一次。•10s 后，将 foo() 定时任务执行周期从每 10s 执行调整为 1s 执行。•10s 后，关闭 foo() 定时任务执行。•10s 后，开启 foo() 定时任务执行。

import com.github.jitwxs.sample.ds.config.AbstractDSHandler;
    import org.springframework.context.ApplicationEvent;
    import org.springframework.context.ApplicationListener;
    import org.springframework.stereotype.Component;
    
    import java.lang.reflect.Method;
    import java.util.Collections;
    import java.util.List;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.LockSupport;
    
    @Component
    public class SchedulerTestDSHandler extends AbstractDSHandler<SchedulerTestTaskInfo> implements ApplicationListener {
        public volatile List<SchedulerTestTaskInfo> taskInfoList = Collections.singletonList(
                SchedulerTestTaskInfo.builder()
                        .id(1)
                        .cron("0/10 * * * * ? ")
                        .isValid(true)
                        .reference("com.github.jitwxs.sample.ds.test.SchedulerTest#foo")
                        .build()
        );
    
        @Override
        protected ExecutorService getWorkerExecutor() {
            return Executors.newFixedThreadPool(2);
        }
    
        @Override
        protected List<SchedulerTestTaskInfo> listTaskInfo() {
            return taskInfoList;
        }
    
        @Override
        protected void doProcess(SchedulerTestTaskInfo taskInfo) {
            final String reference = taskInfo.getReference();
            final String[] split = reference.split("#");
            if(split.length != 2) {
                return;
            }
    
           try {
               final Class<?> clazz = Class.forName(split[0]);
               final Method method = clazz.getMethod(split[1]);
               method.invoke(clazz.newInstance());
           } catch (Exception e) {
               e.printStackTrace();
           }
        }
    
        @Override
        public void onApplicationEvent(ApplicationEvent applicationEvent) {
            Executors.newScheduledThreadPool(1).scheduleAtFixedRate(() -> {
                LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(10));
    
                // setting 1 seconds execute
                taskInfoList = Collections.singletonList(
                        SchedulerTestTaskInfo.builder()
                                .id(1)
                                .cron("0/1 * * * * ? ")
                                .isValid(true)
                                .reference("com.github.jitwxs.sample.ds.test.SchedulerTest#foo")
                                .build()
                );
    
                LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(10));
    
                // setting not valid
                taskInfoList = Collections.singletonList(
                        SchedulerTestTaskInfo.builder()
                                .id(1)
                                .cron("0/1 * * * * ? ")
                                .isValid(false)
                                .reference("com.github.jitwxs.sample.ds.test.SchedulerTest#foo")
                                .build()
                );
    
                LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(10));
    
                // setting valid
                taskInfoList = Collections.singletonList(
                        SchedulerTestTaskInfo.builder()
                                .id(1)
                                .cron("0/1 * * * * ? ")
                                .isValid(true)
                                .reference("com.github.jitwxs.sample.ds.test.SchedulerTest#foo")
                                .build()
                );
            }, 12, 86400, TimeUnit.SECONDS);
        }
    }

### 3.4 运行程序 ###

整个应用包结构如下：

![null][]

运行程序后，在控制台可以观测到如下输出：

![null][null 1]

##  ##

四、后记

以上完成了动态定时任务的介绍，你能够根据本篇文章，实现以下需求吗：

•本文基于 cron 表达式实现了频率控制，你能改用 fixedDelay 或 fixedRate 实现吗？•基于数据库/配置文件/配置中心，实现对服务中定时任务的动态频率调整和任务的启停。•开发一个数据表历史数据清理功能，能够动态配置要清理的表、清理的规则、清理的周期。•开发一个数据表异常数据告警功能，能够动态配置要扫描的表、告警的规则、扫描的周期。

往期推荐

[大数据批处理框架Spring Batch全面解析][Spring Batch]

[软件架构设计说明书该怎么写？][Link 1]

[基于Redis位图实现用户签到功能][Redis]

[后端程序员也可以用Grafana做出漂亮可视化界面！][Grafana]

[面试官：请你谈谈ConcurrentHashMap][ConcurrentHashMap]

![5e62150a32dd266973faca1c6b48302e.png][]

欢迎分享转发，有帮助的话点个“在看”

![e878e7f05a31a07776bdb67fe7f019f7.gif][]

[4645e47d5f7baa7efa8e6d77c914c9da.gif]: /images/20221022/5303ea9bccd74404a11f3e415faed09a.png
[48218734dc9322e5b2f22ed5dd0ba301.gif]: /images/20221022/05cdc4f88b4c4c83b126313e432dde95.png
[4ad8fb554c130b15fe4be697741dc81a.png]: /images/20221022/fdd32be0a1304f2f96baa170a7ce697c.png
[null]: /images/20221022/b05b5fb2742446b2ae0084ac98b41dd5.png
[null 1]: /images/20221022/d1e58c8c45af4c36b6bf595a4fffd37d.png
[Spring Batch]: http://mp.weixin.qq.com/s?__biz=MzU2NDc4MjE2Ng%3D%3D&chksm=fc44e401cb336d177b91184451491cea3a65571e86590917cbb80323441d74ee9f87d2057b7c&idx=1&mid=2247487800&scene=21&sn=06e80a846694667aedc5298451e3c8f7#wechat_redirect
[Link 1]: http://mp.weixin.qq.com/s?__biz=MzU2NDc4MjE2Ng%3D%3D&chksm=fc44e40fcb336d192b19a3a1b71aba782de80c30c69d7185ccdbe5ddc436f20e7776937609e4&idx=1&mid=2247487798&scene=21&sn=1a72dd891a03c7e6e146fed1902057e5#wechat_redirect
[Redis]: http://mp.weixin.qq.com/s?__biz=MzU2NDc4MjE2Ng%3D%3D&chksm=fc44e40bcb336d1da5e03ea3b94a8e1e1e6959e5e93a0abdb200616b311f6b07c9b26d2af2a6&idx=1&mid=2247487794&scene=21&sn=ede93c761ab1513515376c8827e5bc28#wechat_redirect
[Grafana]: http://mp.weixin.qq.com/s?__biz=MzU2NDc4MjE2Ng%3D%3D&chksm=fc44e40dcb336d1bc8d40a83dee297540d3daed46db3b1269977c7b8d772496252e5b06069fb&idx=1&mid=2247487796&scene=21&sn=2e29878fd492549873d74549ca1c7f98#wechat_redirect
[ConcurrentHashMap]: http://mp.weixin.qq.com/s?__biz=MzU2NDc4MjE2Ng%3D%3D&chksm=fc44e5f9cb336cef02534365a10c008a04d952a7a3927c879a9f93494e2d921bbdb50f8d5c71&idx=1&mid=2247487680&scene=21&sn=f3b7b1dc9ae76c2bc3ced416350331f3#wechat_redirect
[5e62150a32dd266973faca1c6b48302e.png]: /images/20221022/7b61824dbf9641bdaaaa4458bd60d0a2.png
[e878e7f05a31a07776bdb67fe7f019f7.gif]: /images/20221022/03223c336d614a698a06ee927b9db9ef.png