基于springboot和quartz的任务调度系统

一、简述

项目地址： https://github.com/littlechare/job-manager

项目详细的信息请参看项目的 README

下载地址： https://download.csdn.net/download/w172087242/10333176

之前想写一个关于定时任务的组建，便于项目中快速集成；

写完之后，又继续在此基础上演变成系统，当然该系统还缺少很多东西；

如业务方的bid生成与验证、任务回调响应报文的规则、任务调度的监控、如何做HA等等；

因为是个人项目，所以用了部分下班时间和本次清明节的空闲时间实现，

所以项目可定也有不足的地方，但是如果开发者想使用，也可以下载下来快速

集成，也可在此基础上加以修改。

二、项目功能简述

2.1 任务类型支持说明

2.2 业务方说明

1.业务方需要自己维护业务方编号(bid)和任务编号;

2.支持任务的随时取消、随时更改、随时添加等，回调地址一http(s)打头

2.3 模块说明与类说明

三、任务系统代码示例；

3.1 任务配置类(ScheduleTaskConfig)

package com.littlehow.job.base.config;

import com.littlehow.job.base.BaseTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Configurable;
import org.springframework.context.annotation.Lazy;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.SchedulingConfigurer;
import org.springframework.scheduling.config.*;
import org.springframework.scheduling.support.CronTrigger;
import org.springframework.stereotype.Component;

import java.util.Date;

@Lazy(false)
@Configurable
@Component
@EnableScheduling
public class ScheduleTaskConfig implements SchedulingConfigurer {
    private Logger log = LoggerFactory.getLogger(ScheduleTaskConfig.class);
    private final static String TASK_NOT_EXISTS = "not exists";
    private final static String TASK_EXISTS = "exists";
    private final static String FAILURE = "failure";
    private final static String SUCCESS = "success";
    private ScheduledTaskRegistrar scheduledTaskRegistrar;
    @Override
    public void configureTasks(ScheduledTaskRegistrar scheduledTaskRegistrar) {
        this.scheduledTaskRegistrar = scheduledTaskRegistrar;
        initTask();
    }

    /**
     * 添加任务
     * @param task
     * @return
     */
    public String addTask(BaseTask task) {
        if (scheduledTaskRegistrar == null || task == null) {
            return FAILURE;
        }
        if (TaskConfig.containsTask(task.getId())) {
            return TASK_EXISTS;
        }
        try {
            addTask0(task);
            TaskConfig.addTask(task);
            return SUCCESS;
        } catch (Exception e) {
            log.error("新增定时任务失败:" + task, e);
            throw e;
        }
    }

    /**
     * 改变任务执行频率
     * @param taskId
     * @param expression
     * @return
     */
    public String changeTask(String taskId, String expression) {
        BaseTask baseTask = TaskConfig.getTask(taskId);
        if (baseTask == null || TaskType.TRIGGER != baseTask.taskType || expression == null) {
            return TASK_NOT_EXISTS;
        }
        log.info("change trigger expression:(id=" + taskId + ",expression=" + expression+")");
        baseTask.setExpression(expression);
        return SUCCESS;
    }

    /**
     * 取消定时任务
     * @param taskId
     * @return
     */
    public String cancelTask(String taskId) {
        if (!TaskConfig.containsTask(taskId)) {
            return TASK_NOT_EXISTS;
        }
        try {
            log.info("cancel task:" + taskId);
            TaskConfig.removeTask(taskId).getScheduledTask().cancel();
        } catch (Exception e) {
            log.error("取消任务失败:" + taskId, e);
            throw e;
        }
        return SUCCESS;
    }


    /**
     * 初始化已配置任务
     */
    private void initTask() {
        TaskConfig.getTasks().forEach(task -> addTask0(task));
    }

    private void addTask0(BaseTask task) {
        log.info("add task:" + task);
        switch (task.taskType) {
            case TRIGGER: task.setScheduledTask(addTriggerTask(task));
                break;
            case CRON: task.setScheduledTask(addCronTask(task, task.getExpression()));
                break;
            case FIXED_RATE: task.setScheduledTask(addFixedRateTask(task, task.interval()));
                break;
            case FIXED_DELAY: task.setScheduledTask(addFixedDelayTask(task, task.interval(), task.delay()));
                break;
            default:
        }
    }

    /**
     * 添加不可改变时间表的定时任务
     * @param task
     */
    private ScheduledTask addCronTask(Runnable task, String expression) {
        return scheduledTaskRegistrar.scheduleCronTask(new CronTask(task, expression));
    }

    /**
     * 添加可变时间task
     * @param task
     * @return
     */
    private ScheduledTask addTriggerTask(BaseTask task) {
        return scheduledTaskRegistrar.scheduleTriggerTask(new TriggerTask(task, triggerContext -> {
            CronTrigger trigger = new CronTrigger(task.getExpression());
            Date nextExec = trigger.nextExecutionTime(triggerContext);
            return nextExec;
        }));
    }

    /**
     * 设置固定频率的定时任务
     * @param task
     * @param interval
     */
    private ScheduledTask addFixedRateTask(Runnable task, long interval) {
        return scheduledTaskRegistrar.scheduleFixedRateTask(new IntervalTask(task, interval, 0L));
    }

    /**
     * 设置延迟以固定频率执行的定时任务
     * @param task
     * @param interval
     * @param delay
     */
    private ScheduledTask addFixedDelayTask(Runnable task, long interval, long delay) {
        return scheduledTaskRegistrar.scheduleFixedDelayTask(new IntervalTask(task, interval, delay));
    }
}

3.2 任务基本实体类(BaseTask)

package com.littlehow.job.base;


import com.littlehow.job.base.config.TaskConfig;
import com.littlehow.job.base.config.TaskType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.scheduling.config.ScheduledTask;

/**
 * 基本的任务配置类
 */
public abstract class BaseTask implements Runnable,InitializingBean {
    protected final Logger log = LoggerFactory.getLogger(this.getClass());
    public final TaskType taskType;
    private ScheduledTask scheduledTask;
    private final String id;
    public BaseTask(TaskType taskType, String id) {
        this.taskType = taskType;
        this.id = id;
    }

    /**
     *  获取任务表达式如：0 0 0/1 * * *? (每个整点执行)
     * @return
     */
    public abstract String getExpression();

    /**
     * 固定频率执行的时间间隔
     * @return
     */
    public abstract long interval();

    /**
     * 固定频率执行的延迟时间
     * @return
     */
    public abstract long delay();

    /**
     * 设置任务表达式
     * @param expression
     */
    public abstract void setExpression(String expression);

    /**
     * 获取任务唯一标识
     * @return
     */
    public String getId() {
        return id;
    }

    public final ScheduledTask getScheduledTask() {
        return scheduledTask;
    }

    public final void setScheduledTask(ScheduledTask scheduledTask) {
        this.scheduledTask = scheduledTask;
    }

    @Override
    public void afterPropertiesSet() {
        TaskConfig.addTask(this);
    }

    public String toString() {
        return this.getClass().getSimpleName() + "(id:" + id + ",expression:" + getExpression()
                + ",type:" + taskType + ",interval:" + interval()
                + ", delay:" + delay() + ")";
    }
}

3.3 任务执行类(TaskExecuteService)

package com.littlehow.job.service;

import com.littlehow.job.base.BaseCronTask;
import com.littlehow.job.base.BaseFixedTask;
import com.littlehow.job.base.ScheduleBaseService;
import com.littlehow.job.manager.pojo.TaskDto;
import com.littlehow.job.manager.pojo.TaskType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.util.regex.Matcher;
import java.util.regex.Pattern;

@Service
public class TaskExecuteService {
    @Autowired
    private ScheduleBaseService scheduleBaseService;
    @Autowired
    private CallbackExecuteService executeService;
    private static final Pattern fixed = Pattern.compile("fixed=(//d+)//s*(,//s*delay=(//d+))?");

    /**
     * 新增任务
     * 因表达式解析错误可能会抛出异常
     * @param taskDto
     */
    public void addTask(TaskDto taskDto) {
        if (taskDto == null) {
            return;
        }
        String taskId = getTaskId(taskDto);
        CallbackExecuteService.addOrUpdateTask(taskId, taskDto.getCallbackUrl());
        TaskType taskType = taskDto.getTaskType();
        switch (taskType) {
            case FIXED:
            case FIXED_DELAY:
                long[] params = getFixedParameters(taskDto.getExpression());
                BaseFixedTask baseFixedTask = new BaseFixedTask(taskId,
                        params[0], params[1]) {
                    @Override
                    public void run() {
                        executeService.execute(this.getId());
                    }
                };
                scheduleBaseService.addTask(baseFixedTask);
                break;
            case CRON:
            case TRIGGER:
                BaseCronTask baseCronTask = new BaseCronTask(taskType == TaskType.CRON ?
                com.littlehow.job.base.config.TaskType.CRON :
                com.littlehow.job.base.config.TaskType.TRIGGER, taskId) {
                    @Override
                    public void run() {
                        executeService.execute(this.getId());
                    }
                };
                baseCronTask.setExpression(taskDto.getExpression());
                scheduleBaseService.addTask(baseCronTask);
        }
    }

    /**
     * 修改任务
     * @param taskDto
     */
    public void updateTask(TaskDto taskDto) {
        if (!StringUtils.isEmpty(taskDto.getExpression())) {
            //修改表达式
            scheduleBaseService.changeTask(getTaskId(taskDto), taskDto.getExpression());
        }
        if (!StringUtils.isEmpty(taskDto.getCallbackUrl())) {
            //修改回调地址
            CallbackExecuteService.addOrUpdateTask(getTaskId(taskDto), taskDto.getCallbackUrl());
        }
    }

    /**
     * 删除任务
     * @param taskDto
     */
    public void removeTask(TaskDto taskDto) {
        String taskId = getTaskId(taskDto);
        scheduleBaseService.removeTask(taskId);
        CallbackExecuteService.removeTask(taskId);
    }

    /**
     * 获取fixed类型的实际参数信息
     * @param expression
     * @return
     */
    private static long[] getFixedParameters(String expression) {
        Matcher matcher = fixed.matcher(expression);
        if (matcher.find()) {
            long[] fixedParam = {0L, 0L};
            fixedParam[0] = Long.parseLong(matcher.group(1));
            String delay = matcher.group(3);
            if (delay != null) {
                fixedParam[1] = Long.parseLong(delay);
            }
            return fixedParam;
        }
        throw new IllegalArgumentException("invalid expression:" +expression);
    }

    /**
     * 获取任务编号
     * @param task
     * @return
     */
    private String getTaskId(TaskDto task) {
        return task.getBusinessId() + "-" + task.getTaskId();
    }
}

3.4 任务回调地址处理类(CallbackExecuteService)

package com.littlehow.job.service;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

import java.util.HashMap;
import java.util.Map;

/**
 * 执行调用定时任务的回调地址
 * 可以约定定时任务的回执报文，便于进一步解析
 * littlehow 2018/4/7
 */
@Service
public class CallbackExecuteService {
    private final static Logger log = LoggerFactory.getLogger(CallbackExecuteService.class);
    /**
     * 缓存任务与回调之间的关系
     */
    private static final Map<String, String> callbackInfo = new HashMap<>();

    /**
     * http任务处理接口
     */
    private RestTemplate restTemplate = new RestTemplate();

    /**
     * 执行远程任务
     * @param taskId
     */
    public void execute(String taskId) {
        String url = callbackInfo.get(taskId);
        if (url == null) {
            log.error("任务[" + taskId + "]的回调地址为空");
            //FIXME 可以在此处加上监控或通知，也可以已异常的形式抛出，用aop统一处理
            return;
        }
        try {
            //FIXME 如果制定了返回值规则，则可以详细解析，否则就简单解析状态码
            ResponseEntity<String> response = restTemplate.getForEntity(url, String.class);
            if (response.getStatusCodeValue() >= 200 && response.getStatusCodeValue() <300) {
                //成功
                log.info("定时任务[" + taskId + "]调用完成");
            } else {
                //失败
                log.error("任务[" + taskId + ":" + url + "]执行失败," + response.toString());
            }
        } catch (Throwable t) {
            log.error("任务[" + taskId + ":" + url + "]执行异常", t);
            //FIXME 可以在此处加上监控或通知，也可以抛出此部分异常，用aop统一处理
        }
    }

    /**
     * 新增或修改任务回调
     * @param taskId
     * @param callbackUrl
     */
    static String addOrUpdateTask(String taskId, String callbackUrl) {
        return callbackInfo.put(taskId, callbackUrl);
    }

    /**
     * 删除任务回调
     * @param taskId
     * @return
     */
    static String removeTask(String taskId) {
        return callbackInfo.remove(taskId);
    }
}