Java轻量级动态定时任务框架基于Spring实现高效任务调度与管理-源码网

戴敦邦《西游记》插画_15_国绘画苑_来自小红书网页版.jpg

0 介绍

0.1 背景

在笔者开发的系统中，原先的定时任务采用Spring框架的@Scheduled注解加配置文件进行驱动，应用启动后无法动态管理定时任务，暂停定时任务、调整定时任务运行周期等操作需修改配置文件后重新部署应用才能生效，很不方便。

当时我便想引入定时任务管理功能，开源的xxl-job等项目可实现此类功能，但考虑到只需进行简单的定时任务管理、不想引入过多依赖，增加系统复杂度，因此我便开发了这个轻量级的定时任务管理服务用于管理系统的定时任务。

上线后使用了一年多时间，运行良好，较好的解决了之前遇到的痛点问题。

0.2 代码功能

这是一个轻量的定时任务管理服务，包含定时任务创建、启动、暂停、删除、手动执行、调整周期等功能。

0.3 代码依赖

Spring框架：使用Spring的ThreadPoolTaskScheduler作为底层的调度器，同时运行任务的类需注册为Spring的Bean
关系型数据库：使用了MySQL数据库作为定时任务的持久化存储

0.4 代码结构

该代码主要包含以下几个部分：

ScheduledTaskService：定时任务服务类，提供了定时任务的管理操作，包括创建、启动、暂停、删除、手动执行、调整周期等
TaskSchedulerConfig：定时任务调度器配置类，负责配置定时任务调度器ThreadPoolTaskScheduler
VtaScheduledTaskMapper：定时任务表Mapper，负责与定时任务的数据库表进行交互
VtaScheduledTask：定时任务表实体类，定义了定时任务的基本信息，如任务名称、任务运行的Spring Bean name、任务运行的方法名、方法入参、cron表达式、任务运行状态等
VtaScheduledTaskRunLogMapper：定时任务运行日志表Mapper，记录每一次的任务运行日志
VtaScheduledTaskRunLog：定时任务运行日志表实体类

1 定时任务管理-服务类

@Slf4j
@Service
public class ScheduledTaskService {
    private final Map> taskScheduledFutureMap = new ConcurrentHashMap<>();

    /**
     * Spring定时任务调度器（线程池）
     * 
     * 通过{@link TaskSchedulerConfig#threadPoolTaskScheduler()}配置Bean
     */
    @Resource
    private ThreadPoolTaskScheduler threadPoolTaskScheduler;

    @Resource
    private VtaScheduledTaskMapper vtaScheduledTaskMapper;

    @Resource
    private VtaScheduledTaskRunLogMapper vtaScheduledTaskRunLogMapper;

    /**
     * 从数据库表加载定时任务
     */
    public void loadTask() {
        List tasks = vtaScheduledTaskMapper.selectAll();
        for (VtaScheduledTask task : tasks) {
            if (task.isRunning()) {
                task.setStatus(ScheduledTaskStatusEnum.INIT.code());
                updateTask(task);
            }
            if (Boolean.TRUE.equals(task.getIsStart())) {
                schedule(task);
            }
        }
    }

    /**
     * 分页查询任务列表
     *
     * @param pageRequest 分页查询请求
     * @return 分页数据
     */
    public ResponseDataWithPageInfo findPage(ScheduledTaskPageRequest pageRequest) {
        pageRequest.startPage();
        List list = vtaScheduledTaskMapper.findPage(pageRequest);
        return new ResponseDataWithPageInfo(list);
    }

    public List listAll() {
        return vtaScheduledTaskMapper.selectAll();
    }

    /**
     * 创建任务
     * 

     * 初始为未启动状态
     *
     * @param createRequest 创建请求
     */
    public void createTask(ScheduledTaskCreateRequest createRequest) {
        createRequest.check();
        VtaScheduledTask task = createRequest.toRecord();
        vtaScheduledTaskMapper.insert(task);
    }

    /**
     * 调整定时任务周期（cron表达式）
     *
     * @param taskId 任务id
     * @param cron   cron表达式
     */
    @Transactional(rollbackFor = Exception.class)
    public void updateTaskCron(Integer taskId, String cron) {
        if (!CronExpression.isValidExpression(cron)) {
            throw new BizException("cron表达式不合法");
        }

        VtaScheduledTask task = getTaskById(taskId);
        if (Objects.equals(task.getCron(), cron)) {
            return; // cron无变化，直接返回
        }
        task.setCron(cron);
        vtaScheduledTaskMapper.updateByPrimaryKeySelective(task);
        if (task.getIsStart()) { // 如果是启动状态，则重新调度任务
            cancel(taskId);
            schedule(task);
        }
    }

    /**
     * 启动任务
     *
     * @param taskId 任务id
     */
    @Transactional(rollbackFor = Exception.class)
    public void startTask(Integer taskId) {
        // 更新数据库为启动状态
        updateStartStatus(taskId, true);
        // 调度任务
        VtaScheduledTask task = getTaskById(taskId);
        schedule(task);
    }

    /**
     * 暂停任务
     *
     * @param taskId 任务id
     */
    @Transactional(rollbackFor = Exception.class)
    public void stopTask(Integer taskId) {
        // 更新数据库为未启动状态
        updateStartStatus(taskId, false);
        // 取消任务
        cancel(taskId);
    }

    /**
     * 删除任务
     *
     * @param taskId 任务id
     */
    @Transactional(rollbackFor = Exception.class)
    public void deleteTask(Integer taskId) {
        vtaScheduledTaskMapper.deleteByPrimaryKey(taskId);
        cancel(taskId);
    }

    /**
     * 手动执行一次
     *
     * @param taskId 任务id
     * @throws TaskMultiRunningException 任务正在运行中则抛此异常
     */
    public void executeTask(Integer taskId) throws TaskMultiRunningException {
        VtaScheduledTask task = getTaskById(taskId);
        if (task.isRunning()) {
            throw new TaskMultiRunningException("任务正在运行中");
        }
        Runnable runnable = getRunnable(taskId);
        threadPoolTaskScheduler.execute(runnable);
    }

    private void updateTask(VtaScheduledTask task) {
        vtaScheduledTaskMapper.updateByPrimaryKeySelective(task);
    }

    public VtaScheduledTask getTaskById(Integer taskId) {
        return vtaScheduledTaskMapper.selectByPrimaryKey(taskId);
    }

    public VtaScheduledTask getTaskByName(String taskName) {
        return vtaScheduledTaskMapper.selectByName(taskName);
    }

    private void updateStartStatus(Integer taskId, boolean isStart) {
        VtaScheduledTask record = new VtaScheduledTask();
        record.setId(taskId);
        record.setIsStart(isStart);
        vtaScheduledTaskMapper.updateByPrimaryKeySelective(record);
    }

    /**
     * 调度任务
     * 
     * 同一个任务不会并发执行
     *
     * @param task 任务
     */
    private void schedule(VtaScheduledTask task) {
        Runnable runnable = getRunnable(task.getId());
        String cron = task.getCron();
        ScheduledFuture scheduledFuture = threadPoolTaskScheduler.schedule(runnable, new CronTrigger(cron));
        taskScheduledFutureMap.put(task.getId(), scheduledFuture);
    }

    /**
     * 取消任务调度
     *
     * @param taskId 任务id
     */
    private void cancel(Integer taskId) {
        ScheduledFuture scheduledFuture = taskScheduledFutureMap.get(taskId);
        if (scheduledFuture != null) {
            boolean cancel = scheduledFuture.cancel(true);
            log.info("cancel: {}", cancel);
            taskScheduledFutureMap.remove(taskId);
        }
    }

    // 获取任务对应的Runnable
    private Runnable getRunnable(Integer taskId) {
        return () -> {
            LocalDateTime startTime = LocalDateTime.now();
            VtaScheduledTask task = getTaskById(taskId);
            VtaScheduledTaskRunLog runLog = null;
            try {
                if (task.isRunning()) { // 如果正在执行，则直接返回，不允许并发运行同一个任务
                    log.error("Scheduled task is already running, id: {}, name: {}", task.getId(), task.getName());
                    return;
                }

                runLog = logStart(taskId, startTime);
                doRun(task);
                logEndSuccess(taskId, startTime, runLog);
            } catch (Throwable e) {
                log.error("Scheduled task running error, taskId: [{}]", taskId, e);
                logEndException(task, startTime, e, runLog);
            }
        };
    }

    private void logEndSuccess(Integer taskId, LocalDateTime startTime, VtaScheduledTaskRunLog runLog) {
        LocalDateTime endTime = LocalDateTime.now();
        VtaScheduledTask task = new VtaScheduledTask();
        task.setId(taskId);
        task.setStatus(ScheduledTaskStatusEnum.SUCCESS.code());
        task.setLastExecuteEndTime(endTime);
        long costTime = Duration.between(startTime, endTime).toMillis();
        task.setLastExecuteCostTime(costTime);
        task.setLastExecuteSuccessEndTime(endTime);
        updateTask(task);

        // 更新运行日志记录
        runLog.setStatus(ScheduledTaskStatusEnum.SUCCESS.code());
        runLog.setEndTime(endTime);
        runLog.setCostTime(costTime);
        vtaScheduledTaskRunLogMapper.updateByPrimaryKeySelective(runLog);
    }

    private void logEndException(VtaScheduledTask task, LocalDateTime startTime, Throwable throwable, VtaScheduledTaskRunLog runLog) {
        LocalDateTime endTime = LocalDateTime.now();
        VtaScheduledTask updateRecord = new VtaScheduledTask();
        updateRecord.setId(task.getId());
        updateRecord.setStatus(ScheduledTaskStatusEnum.EXCEPTION.code());
        updateRecord.setLastExecuteEndTime(endTime);
        long costTime = Duration.between(startTime, endTime).toMillis();
        updateRecord.setLastExecuteCostTime(costTime);
        updateRecord.setLastExecuteExceptionEndTime(endTime);
        String exceptionInfo = getExceptionInfo(throwable);
        updateRecord.setLastExecuteExceptionInfo(exceptionInfo);
        updateTask(updateRecord);

        // 更新运行日志记录
        if (runLog != null) {
            runLog.setStatus(ScheduledTaskStatusEnum.EXCEPTION.code());
            runLog.setEndTime(endTime);
            runLog.setCostTime(costTime);
            runLog.setExceptionInfo(exceptionInfo);
            vtaScheduledTaskRunLogMapper.updateByPrimaryKeySelective(runLog);
        }
    }

    private String getExceptionInfo(Throwable throwable) {
        String exceptionInfo = ExceptionUtils.getStackTrace(throwable);
        if (exceptionInfo.length() > 10000) {
            exceptionInfo = exceptionInfo.substring(0, 10000);
        }
        return exceptionInfo;
    }

    private VtaScheduledTaskRunLog logStart(Integer taskId, LocalDateTime startTime) {
        VtaScheduledTask task = new VtaScheduledTask();
        task.setId(taskId);
        task.setStatus(ScheduledTaskStatusEnum.RUNNING.code());
        task.setLastExecuteStartTime(startTime);
        updateTask(task);

        // 添加运行日志记录
        VtaScheduledTaskRunLog runLog = new VtaScheduledTaskRunLog();
        runLog.setStatus(ScheduledTaskStatusEnum.RUNNING.code());
        runLog.setStartTime(startTime);
        vtaScheduledTaskRunLogMapper.insert(runLog);
        return runLog;
    }

    private void doRun(VtaScheduledTask task) throws IllegalAccessException, InvocationTargetException {
        Object target = task.getTarget();
        Method method = task.getMethod(target);

        if (StringUtils.isEmpty(task.getParams())) {
            method.invoke(target);
        } else {
            method.invoke(target, task.getParams());
        }
    }
}

2 Spring定时任务调度器配置

@Configuration
public class TaskSchedulerConfig {
    @Bean
    public ThreadPoolTaskScheduler threadPoolTaskScheduler() {
        ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
        threadPoolTaskScheduler.setPoolSize(10);
        threadPoolTaskScheduler.setThreadNamePrefix("schedule-task-");
        threadPoolTaskScheduler.setRemoveOnCancelPolicy(true);
        return threadPoolTaskScheduler;
    }
}

3 定时任务创建-请求类

@Data
public class ScheduledTaskCreateRequest {
    @NotNull
    @Size(min = 2, max = 255)
    String name;

    @NotBlank
    String beanName;

    @NotBlank
    String methodName;

    String params;

    @NotBlank
    String cron;

    public void check() throws RequestInvalidException {
        if (!CronExpression.isValidExpression(cron)) {
            throw new RequestInvalidException("cron表达式不合法: " + cron);
        }

        Object target;
        try {
            target = SpringContextUtil.getBean(beanName);
        } catch (NoSuchBeanDefinitionException e) {
            throw new RequestInvalidException("找不到对应的Bean: " + beanName, e);
        }

        try {
            // 反射获取public方法
            if (StringUtils.isEmpty(params)) {
                target.getClass().getMethod(methodName);
            } else {
                target.getClass().getMethod(methodName, String.class);
            }
        } catch (NoSuchMethodException e) {
            String method = methodName + (StringUtils.isNotEmpty(params) ? "(String.class)" : "()");
            throw new RequestInvalidException("目标方法不存在: " + method, e);
        }

        VtaScheduledTaskMapper vtaScheduledTaskMapper = SpringContextUtil.getBean(VtaScheduledTaskMapper.class);
        if (vtaScheduledTaskMapper.selectByName(name) != null) {
            throw new RequestInvalidException("任务名称已存在，请使用其他名称");
        }
    }

    public VtaScheduledTask toRecord() {
        VtaScheduledTask task = new VtaScheduledTask();
        task.setName(name);
        task.setBeanName(beanName);
        task.setMethodName(methodName);
        task.setParams(params);
        task.setCron(cron);
        task.setIsStart(false); // 初始为未启动状态
        task.setStatus(ScheduledTaskStatusEnum.INIT.code()); // 初始为待运行状态
        return task;
    }
}

4 定时任务运行状态枚举

public enum ScheduledTaskStatusEnum implements IEnum {
    INIT(1, "待运行"),
    RUNNING(2, "运行中"),
    EXCEPTION(3, "异常"),
    SUCCESS(4, "成功"),
    ;

    private final int code;

    private final String value;

    ScheduledTaskStatusEnum(int code, String value) {
        this.code = code;
        this.value = value;
    }

    @Override
    public int code() {
        return code;
    }

    @Override
    public String value() {
        return value;
    }
}

5 表结构

5.1 定时任务表

建表SQL

CREATE TABLE `t_vta_scheduled_task` (
  `id` int(10) unsigned NOT NULL AUTO_INCREMENT COMMENT '自增id',
  `name` varchar(255) NOT NULL COMMENT '任务名称 唯一',
  `bean_name` varchar(500) NOT NULL COMMENT '执行任务的Spring Bean name',
  `method_name` varchar(255) NOT NULL COMMENT '执行任务的方法名',
  `params` varchar(1000) DEFAULT NULL COMMENT '方法入参（字符串类型）可为空',
  `cron` varchar(100) NOT NULL COMMENT 'cron表达式',
  `is_start` tinyint(1) NOT NULL DEFAULT '0' COMMENT '任务是否启动 1:是 0:否',
  `status` int(11) NOT NULL DEFAULT '1' COMMENT '任务运行状态 1待运行 2运行中 3异常 4成功',
  `last_execute_start_time` datetime DEFAULT NULL COMMENT '最近一次执行开始时间',
  `last_execute_end_time` datetime DEFAULT NULL COMMENT '最近一次执行结束时间',
  `last_execute_cost_time` bigint(20) DEFAULT NULL COMMENT '最近一次执行耗时（毫秒）',
  `last_execute_success_end_time` datetime DEFAULT NULL COMMENT '最近一次执行成功的结束时间',
  `last_execute_exception_end_time` datetime DEFAULT NULL COMMENT '最近一次执行异常的结束时间',
  `last_execute_exception_info` text COMMENT '最近一次执行异常的异常信息',
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk:name` (`name`) USING BTREE COMMENT '名称唯一索引'
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='定时任务表';

表实体类

@Slf4j
@Data
public class VtaScheduledTask {
    /**
     * 自增id，主键
     */
    private Integer id;

    /**
     * 任务名称 唯一
     */
    private String name;

    /**
     * 执行任务的Spring Bean name
     */
    private String beanName;

    /**
     * 执行任务的方法名
     */
    private String methodName;

    /**
     * 方法入参（字符串类型）可为空
     */
    private String params;

    /**
     * cron表达式
     */
    private String cron;

    /**
     * 任务是否启动
     */
    private Boolean isStart;

    /**
     * 任务运行状态 1待运行 2运行中 3异常 4成功
     * 
     * 对应枚举：{@link ScheduledTaskStatusEnum }
     */
    private Integer status;

    /**
     * 最近一次执行开始时间
     */
    private LocalDateTime lastExecuteStartTime;

    /**
     * 最近一次执行结束时间
     */
    private LocalDateTime lastExecuteEndTime;

    /**
     * 最近一次执行耗时（毫秒）
     */
    private Long lastExecuteCostTime;

    /**
     * 最近一次执行成功的结束时间
     */
    private LocalDateTime lastExecuteSuccessEndTime;

    /**
     * 最近一次执行异常的结束时间
     */
    private LocalDateTime lastExecuteExceptionEndTime;

    /**
     * 最近一次执行异常的异常信息
     */
    private String lastExecuteExceptionInfo;

    @JsonIgnore
    public boolean isRunning() {
        return Objects.equals(status, ScheduledTaskStatusEnum.RUNNING.code());
    }

    @JsonIgnore
    public Object getTarget() {
        try {
            return SpringContextUtil.getBean(beanName);
        } catch (NoSuchBeanDefinitionException e) {
            throw new BizException("找不到对应的Bean: " + beanName, e);
        }
    }

    @JsonIgnore
    public Method getMethod(Object target) {
        try {
            if (StringUtils.isEmpty(params)) {
                return target.getClass().getMethod(methodName);
            } else {
                return target.getClass().getMethod(methodName, String.class);
            }
        } catch (NoSuchMethodException e) {
            String method = methodName + (StringUtils.isNotEmpty(params) ? "(String.class)" : "()");
            throw new BizException("目标方法不存在: " + method, e);
        }
    }
}

Mybatis Mapper

/**
 * 针对表【t_vta_scheduled_task(定时任务表)】的数据库操作Mapper
 * 
 * 实体类 {@link  VtaScheduledTask}
 */
@Mapper
public interface VtaScheduledTaskMapper {
    int deleteByPrimaryKey(Integer id);

    int insert(VtaScheduledTask record);

    VtaScheduledTask selectByPrimaryKey(Integer id);

    VtaScheduledTask selectByName(String name);

    List selectAll();

    int updateByPrimaryKeySelective(VtaScheduledTask record);

    List findPage(ScheduledTaskPageRequest pageRequest);
}

5.2 定时任务运行日志表

建表SQL

CREATE TABLE `t_vta_scheduled_task_run_log` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '自增id',
  `task_code` int(11) NOT NULL COMMENT '任务编码',
  `task_name` varchar(500) NOT NULL COMMENT '任务名称',
  `params` varchar(1000) DEFAULT NULL COMMENT '任务参数',
  `status` tinyint(4) NOT NULL COMMENT '任务运行状态 2运行中 3异常 4成功',
  `start_time` datetime NOT NULL COMMENT '运行开始时间',
  `end_time` datetime DEFAULT NULL COMMENT '运行结束时间',
  `cost_time` bigint(20) DEFAULT NULL COMMENT '运行耗时（毫秒）',
  `exception_info` text COMMENT '运行异常时的异常信息',
  PRIMARY KEY (`id`),
  KEY `idx:task_code,start_time` (`task_code`,`start_time`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='定时任务运行日志表';

表实体类

@Data
public class VtaScheduledTaskRunLog {
    /**
     * 任务执行id
     */
    private Long id;

    /**
     * 所属任务code {@link VtaTaskNameEnum#getCode()}
     */
    private Integer taskCode;

    /**
     * 所属任务名称 {@link VtaTaskNameEnum#getTaskName()}
     */
    private String taskName;

    /**
     * 任务参数
     */
    private String params;

    /**
     * 任务运行状态 2运行中 3异常 4成功。
     */
    private Integer status;

    /**
     * 运行开始时间
     */
    private LocalDateTime startTime;

    /**
     * 运行结束时间
     */
    private LocalDateTime endTime;

    /**
     * 运行耗时（毫秒）
     */
    private Long costTime;

    /**
     * 运行异常时的异常信息
     */
    private String exceptionInfo;
}

Mybatis Mapper

public interface VtaScheduledTaskRunLogMapper {
    int deleteByPrimaryKey(Long id);

    int insert(VtaScheduledTaskRunLog record);

    int updateByPrimaryKeySelective(VtaScheduledTaskRunLog record);

    VtaScheduledTaskRunLog selectByPrimaryKey(Long id);
}

[Java]基于Spring的轻量级定时任务动态管理框架

0 介绍

0.1 背景

0.2 代码功能

0.3 代码依赖

0.4 代码结构

1 定时任务管理-服务类

2 Spring定时任务调度器配置

3 定时任务创建-请求类

4 定时任务运行状态枚举

5 表结构

5.1 定时任务表

建表SQL

表实体类

Mybatis Mapper

5.2 定时任务运行日志表

建表SQL

表实体类

Mybatis Mapper

相关文章

逆水寒手游导入装扮方法攻略

好用的知识管理软件排行榜

手机练打字软件26键推荐

燕云十六声不见山所有前尘旧物收集方法

推荐阅读

Springboot 常见面试题汇总

if-else 优化的折中思考：不是消灭分支，而是控制风险

JDK25中的StableValue

重写、重载、访问者模式