- 1、引入quartz关联pom
- 2、yml配置
- 3、建立数据表
- 4、entity
- 5、mapper(接口)
- 6、mappring(xml)
- 7、工具类
- 8、BaseQuartzJobBean
- 9、业务代码执行类
- 10、JobManager
- 11、TaskController
- 12、测试
2、yml配置org.springframework.boot spring-boot-starter-quartz 1 org.mybatis.spring.boot mybatis-spring-boot-starter 2.1.2 mysql mysql-connector-java 8.0.20 org.projectlombok lombok true
只展示quartz相关的配置
spring:
quartz:
scheduler-name: quartz-single-scheduler
properties:
org:
quartz:
threadPool:
class: org.quartz.simpl.SimpleThreadPool
# 线程数量
threadCount: 10
# 线程优先级
threadPriority: 5
# 线程池中线程名称的前缀
threadNamePrefix: quartz-single
# 加载任务代码的ClassLoader是否从外部继承
threadsInheritContextClassLoaderOfInitializingThread: true
3、建立数据表
此处是单机版本的quartz，没有使用quartz原生的表结构
CREATE TABLE `sys_job` ( `id` varchar(32) NOT NULL COMMENT '任务ID', `name` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL DEFAULT '' COMMENT '任务名称', `invoke_target` varchar(500) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL COMMENT '目标字符串', `cron_expression` varchar(500) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL DEFAULT '' COMMENT 'cron执行表达式', `run_time` timestamp NULL DEFAULT NULL COMMENT '上次执行时间', `next_run_time` timestamp NULL DEFAULT NULL COMMENT '下次执行时间', `status` int NOT NULL COMMENT '状态：1-执行中，2-暂停', PRIMARY KEY (`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='定时任务调度表'4、entity
@Data
public class JobEntity {
private String id;
private String name;
private String invokeTarget;
private String cronExpression;
private Date runTime;
private Date nextRunTime;
private Integer status;
}
5、mapper(接口)
@Mapper
public interface TaskMapper {
List jobList();
JobEntity selectById(String id);
int insert(JobEntity job);
int update(JobEntity job);
int delete(String id);
}
6、mappring(xml)
7、工具类id,name,invoke_target,cron_expression,run_time,next_run_time,status
import com.example.quartz.entity.JobEntity;
import org.quartz.CronExpression;
import org.quartz.JobDataMap;
import org.quartz.TriggerUtils;
import org.quartz.impl.triggers.CronTriggerImpl;
import org.springframework.util.StringUtils;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
public class QuartzUtils {
public static Map generatorJobDataMap(JobEntity jobEntity) {
String jobId = jobEntity.getId();
String jobName = jobEntity.getName();
String cron = jobEntity.getCronExpression();
String invokeTarget = jobEntity.getInvokeTarget();
String beanName = invokeTarget.substring(0, invokeTarget.indexOf("."));
String methodName = invokeTarget.substring(invokeTarget.indexOf(".") + 1, invokeTarget.indexOf("("));
// 获取参数
List methodParams = new ArrayList<>();
// 默认第一个参数 加上 id 参数, 第二个参数未name
methodParams.add(new Object[]{jobId, String.class});
methodParams.add(new Object[]{jobName, String.class});
// 解析参数
getMethodParams(methodParams, invokeTarget);
Map map = new HashMap<>();
map.put("beanName", beanName);
map.put("methodName", methodName);
map.put("jobId", jobId);
map.put("jobName", jobName);
map.put("cron", cron);
map.put("methodParams", methodParams);
return map;
}
public static void isValid(String cronExpression) {
boolean expression = CronExpression.isValidExpression(cronExpression);
if (!expression) {
throw new RuntimeException("cron表达式错误！");
}
}
public static Object getNextExecution(String cronExpression, int num, boolean format) {
List list = new ArrayList<>();
try {
CronTriggerImpl cronTriggerImpl = new CronTriggerImpl();
cronTriggerImpl.setCronExpression(cronExpression);
// 近n次时间
List dates = TriggerUtils.computeFireTimes(cronTriggerImpl, null, num);
//不进行格式化
if (!format) {
return dates;
}
SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
for (Date date : dates) {
list.add(dateFormat.format(date));
}
} catch (ParseException e) {
e.printStackTrace();
}
return list;
}
public static Date nextCurrentTime(String cron) {
//如果存在获取的新的下次执行时间和当前已执行的时间一致，则返回第二个执行时间
List execution = (List) getNextExecution(cron, 2, false);
long timeMillis = System.currentTimeMillis();
if (Math.abs(timeMillis - execution.get(0).getTime()) > 1000) {
return execution.get(0);
}
//因为如果还在当前任务执行时间节点上，获取到的下次时间和当前是一致的，所有要获取下下次的时间节点
return execution.get(1);
}
public static Class[] getMethodParamsType(List methodParams) {
Class[] classs = new Class[methodParams.size()];
int index = 0;
for (Object[] os : methodParams) {
classs[index] = (Class) os[1];
index++;
}
return classs;
}
public static Object[] getMethodParamsValue(List methodParams) {
Object[] objects = new Object[methodParams.size()];
int index = 0;
for (Object[] os : methodParams) {
objects[index] = os[0];
index++;
}
return objects;
}
private static void getMethodParams(List methodParams, String invokeTarget) {
String paramStr = invokeTarget.substring(invokeTarget.indexOf("(") + 1, invokeTarget.indexOf(")"));
String[] params = paramStr.split(",");
for (String param : params) {
String str = param.trim();
// String字符串类型，包含`'`；
if (str.contains("'")) {
methodParams.add(new Object[]{StringUtils.replace(str, "'", ""), String.class});
}
// boolean布尔类型，等于true或者false
else if ("true".equals(str) || "false".equals(str)) {
methodParams.add(new Object[]{Boolean.valueOf(str), Boolean.class});
}
// long长整形，包含L
else if (str.toUpperCase().contains("L")) {
methodParams.add(new Object[]{Long.valueOf(StringUtils.replace(str.toUpperCase(), "L", "")), Long.class});
}
// double浮点类型，包含D
else if (str.toUpperCase().contains("D")) {
methodParams.add(new Object[]{Double.valueOf(StringUtils.replace(str.toUpperCase(), "D", "")), Double.class});
}
// 其他类型归类为整形
else {
methodParams.add(new Object[]{Integer.valueOf(str), Integer.class});
}
}
}
}
8、BaseQuartzJobBean
BaseQuartzJobBean主要就是根据beanName、methodName、methodParams通过反射执行具体的业务代码；
import com.example.quartz.entity.JobEntity;
import com.example.quartz.mapper.TaskMapper;
import com.example.quartz.utils.QuartzUtils;
import lombok.extern.slf4j.Slf4j;
import org.quartz.JobDataMap;
import org.quartz.JobDetail;
import org.quartz.JobExecutionContext;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.quartz.QuartzJobBean;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.lang.reflect.Method;
import java.util.Date;
import java.util.List;
import java.util.Map;
@Service
@Slf4j
public class BaseQuartzJobBean extends QuartzJobBean {
@Resource
private ApplicationContext applicationContext;
@Resource
private JobManager jobManager;
@Resource
private TaskMapper taskMapper;
@Override
protected void executeInternal(JobExecutionContext jobExecutionContext) {
JobDetail jobDetail = jobExecutionContext.getJobDetail();
String jobId = jobDetail.getKey().getName();
log.info("执行任务，任务ID：" + jobId);
// 获取参数
JobDataMap jobDataMap = jobDetail.getJobDataMap();
Map paramMap = (Map) jobDataMap.get("paramMap");
try {
// 通过反射执行方法
String invokeTarget = paramMap.get("beanName").toString();
log.info("beanName：" + invokeTarget);
String methodStr = paramMap.get("methodName").toString();
log.info("methodName：" + methodStr);
// 执行具体类的方法
List methodParams = (List) paramMap.get("methodParams");
Object beanName = applicationContext.getBean(invokeTarget);
Method method = beanName.getClass().getMethod(methodStr, QuartzUtils.getMethodParamsType(methodParams));
// 执行任务
method.invoke(beanName, QuartzUtils.getMethodParamsValue(methodParams));
} catch (Exception e) {
e.printStackTrace();
log.error("没有找到可执行方法！");
// 需要关闭任务
jobManager.deleteJob(jobId);
} finally {
// 更新执行时间
String cron = paramMap.get("cron").toString();
Date nextCurrentTime = QuartzUtils.nextCurrentTime(cron);
JobEntity job = new JobEntity();
job.setId(jobId);
job.setRunTime(new Date());
job.setNextRunTime(nextCurrentTime);
taskMapper.update(job);
}
}
}
9、业务代码执行类
SysServiceJob 为一个业务代码的执行类，类中可以存在多个方法，也可以有多个类，这些都没有限制可以任意定义；具体执行那个类的那个方法由实体中的invokeTarget 参数解析而来，并且通过BaseQuartzJobBean类利用反射机制进行执行；
@Service
@Slf4j
public class SysServiceJob {
public void test(String jobId, String jobName, String str, Long l, Double d, Boolean b) {
log.info("ID：{}，Name：{}", jobId, jobName);
// 打印参数
log.info("参数：str={},l={},d={},b={}", str, l, d, b);
}
}
10、JobManager
JobManager封装了对任务的修改、暂停、恢复、删除、新增等操作处理；
import com.example.quartz.entity.JobEntity;
import com.example.quartz.mapper.TaskMapper;
import com.example.quartz.utils.QuartzUtils;
import lombok.extern.slf4j.Slf4j;
import org.quartz.*;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.Map;
@Component
@Slf4j
public class JobManager {
@Resource
private Scheduler scheduler;
@Resource
private TaskMapper taskMapper;
public void startJob(JobEntity jobEntity) {
try {
String jobId = jobEntity.getId();
// 判断是否存在
JobKey jobKey = JobKey.jobKey(jobId);
boolean exists = scheduler.checkExists(jobKey);
if (exists) {
throw new RuntimeException("任务已经存在!");
}
log.info("======== 创建任务：{} ========", jobId);
// 构建参数
JobDataMap jobDataMap = new JobDataMap();
Map map = QuartzUtils.generatorJobDataMap(jobEntity);
jobDataMap.put("paramMap", map);
// 以任务的ID来构建定时任务
Class aClass = BaseQuartzJobBean.class;
JobBuilder jobBuilder = JobBuilder.newJob(aClass).withIdentity(jobId);
jobBuilder.setJobData(jobDataMap);
JobDetail jobDetail = jobBuilder.build();
// 设置cron
String cron = jobEntity.getCronExpression();
// 验证cron
QuartzUtils.isValid(cron);
CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(cron);
// 任务触发器
CronTrigger trigger = TriggerBuilder.newTrigger().withIdentity(jobId).withSchedule(scheduleBuilder).build();
scheduler.scheduleJob(jobDetail, trigger);
} catch (Exception e) {
e.printStackTrace();
throw new RuntimeException("任务启动失败!");
}
}
public void updateJob(JobEntity jobEntity) {
try {
String jobId = jobEntity.getId();
log.info("======== 更新任务：{} ========", jobId);
JobKey jobKey = JobKey.jobKey(jobEntity.getId());
boolean exists = scheduler.checkExists(jobKey);
if (!exists) {
// 任务原有没有执行，则不继续执行
return;
}
log.info("======== 创建任务：{} ========", jobId);
// 重新构建参数
Map map = (Map) scheduler.getJobDetail(jobKey).getJobDataMap().get("paramMap");
map.clear();
map.putAll(QuartzUtils.generatorJobDataMap(jobEntity));
// 根据Id获取触发器
TriggerKey triggerKey = TriggerKey.triggerKey(jobId);
// 设置cron
String cron = jobEntity.getCronExpression();
// 验证cron
QuartzUtils.isValid(cron);
CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(cron);
// 重新构建任务触发器
CronTrigger trigger = (CronTrigger) scheduler.getTrigger(triggerKey);
TriggerBuilder builder = trigger.getTriggerBuilder().withIdentity(triggerKey);
trigger = builder.withSchedule(scheduleBuilder).build();
// 重新执行任务
scheduler.rescheduleJob(triggerKey, trigger);
} catch (SchedulerException e) {
e.printStackTrace();
throw new RuntimeException("任务更新失败！");
}
}
public void deleteJob(String jobId) {
try {
JobKey jobKey = JobKey.jobKey(jobId);
scheduler.deleteJob(jobKey);
} catch (SchedulerException e) {
e.printStackTrace();
throw new RuntimeException("任务删除失败！ ");
}
}
public void pauseJob(String jobId) {
try {
JobKey jobKey = JobKey.jobKey(jobId);
boolean exists = scheduler.checkExists(jobKey);
if (!exists) {
throw new RuntimeException("任务未运行!");
}
scheduler.pauseJob(jobKey);
} catch (SchedulerException e) {
e.printStackTrace();
throw new RuntimeException("任务暂停失败：" + e.getMessage());
}
}
public void resumeJob(String jobId) {
// 恢复任务时会立即执行一次任务
log.info("恢复任务：" + jobId);
try {
JobKey jobKey = JobKey.jobKey(jobId);
boolean exists = scheduler.checkExists(jobKey);
if (exists) {
scheduler.resumeJob(jobKey);
} else {
// 任务暂未运行
// 从数据库中查询，并且执行任务
JobEntity jobEntity = taskMapper.selectById(jobId);
startJob(jobEntity);
if (jobEntity == null) {
throw new RuntimeException("任务不存在!");
}
}
} catch (SchedulerException e) {
e.printStackTrace();
throw new RuntimeException("任务恢复失败：" + e.getMessage());
}
}
public void runOnceJob(String jobId) {
try {
JobKey jobKey = JobKey.jobKey(jobId);
boolean exists = scheduler.checkExists(jobKey);
if (!exists) {
throw new RuntimeException("任务未运行!");
}
log.info("执行任务：" + jobId);
scheduler.triggerJob(jobKey);
} catch (SchedulerException e) {
e.printStackTrace();
throw new RuntimeException("任务执行失败：" + e.getMessage());
}
}
public Object getTaskInfo(String cronExpression, int num) {
return QuartzUtils.getNextExecution(cronExpression, num, true);
}
}
11、TaskController
TaskController测试入口类，此处方便测试没有service层代码
@RequestMapping("/task")
@Slf4j
public class TaskController {
@Resource
private JobManager jobManager;
@Resource
private TaskMapper taskMapper;
@ApiOperation(value = "新增任务", notes = "新增任务")
@ApiOperationSupport(order = 5)
@PostMapping("/insert")
public String check(@RequestBody JobEntity job) {
job.setId(UUIDUtils.getUuId());
// 插入数据库
taskMapper.insert(job);
// 启动任务
jobManager.startJob(job);
return "操作成功";
}
@ApiOperation(value = "更新任务", notes = "更新任务")
@ApiOperationSupport(order = 10)
@PutMapping("/update")
public String update(@RequestBody JobEntity job) {
// 更新数据库
int update = taskMapper.update(job);
if (update > 0) {
// 启动任务
jobManager.updateJob(job);
}
return "操作成功";
}
@ApiOperation(value = "删除任务", notes = "删除任务")
@ApiOperationSupport(order = 10)
@DeleteMapping("/delete/{id}")
public String delete(@PathVariable String id) {
// 删除数据库
taskMapper.delete(id);
// 启动任务
jobManager.deleteJob(id);
return "操作成功";
}
@ApiOperation(value = "暂停任务", notes = "暂停任务")
@ApiOperationSupport(order = 10)
@PostMapping("/pause/{id}")
public String pause(@PathVariable String id) {
// 更新数据库
JobEntity job = taskMapper.selectById(id);
job.setStatus(2);
taskMapper.update(job);
// 启动任务
jobManager.pauseJob(id);
return "操作成功";
}
@ApiOperation(value = "恢复任务", notes = "恢复任务")
@ApiOperationSupport(order = 10)
@PostMapping("/resume/{id}")
public String resume(@PathVariable String id) {
// 更新数据库
JobEntity job = taskMapper.selectById(id);
job.setStatus(1);
taskMapper.update(job);
// 启动任务
jobManager.resumeJob(id);
return "操作成功";
}
@ApiOperation(value = "执行一次任务", notes = "执行一次任务")
@ApiOperationSupport(order = 20)
@PostMapping("/once/{id}")
public String runOnceJob(@PathVariable String id) {
jobManager.runOnceJob(id);
return "操作成功";
}
@GetMapping("info/{id}")
@ApiOperation(value = "查询执行时间", notes = "查询执行时间")
@ApiImplicitParam(name = "id", value = "任务id", paramType = "path", required = true, dataType = "String")
public Object getTaskInfo(@PathVariable("id") String id) {
// 查询数据库
JobEntity entity = taskMapper.selectById(id);
return jobManager.getTaskInfo(entity.getCronExpression(), 5);
}
}
12、测试
此处使用knife4j进行测试，也可以结合postmant或者前端页面进行测试，新增测试如下：
在请求参数中的invokeTarget属性包含了具体需要执行的bean名称、方法名称、参数，内容为:sysServiceJob.test('aaa',10L,20.22D,false)，刚好与代码中SysServiceJob类的test方法对应；
执行效果：
