quartz 整合 postgresql 附带例子

Quartz是什么

Quartz是OpenSymphony开源组织在Job scheduling领域又一个开源项目，完全由Java开发，可以用来执行定时任务，类似于java.util.Timer。但是相较于Timer， Quartz增加了很多功能：

持久性作业 - 就是保持调度定时的状态;
作业管理 - 对调度作业进行有效的管理;

为什么用Quartz

项目需要对所有的定时需要进行统一的可视化的管理，之前都是用@Scheduled 进行定时任务的操作，维护比较麻烦

实现思路：

利用Quartz实现定时功能，自定义业务表，业务表跟Quartz关联，通过操作业务表来实现对Quartz的控制，也可进行可视化的管理，业务表存有方法与类的名字，通过反射的方法来实现方法的调用。
当然我们分布式的时候是需要Quartz持久化到数据库中的不能存在内存中

具体使用

本项目用的pgDB 整合可能跟mysql 有些许的不同,使用mybatisPlus进行db的操作

1.springboot引入pom

     <!--quartz依赖 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-quartz</artifactId></dependency>

2.springboot 添加 yml 配置

   quartz:#相关属性配置auto-startup: trueoverwrite-existing-jobs: trueproperties:org:quartz:scheduler:instanceName: IsimClusteredSchedulerinstanceId: AUTOjobStore:class: org.quartz.impl.jdbcjobstore.JobStoreTXdriverDelegateClass: org.quartz.impl.jdbcjobstore.PostgreSQLDelegatetablePrefix: qrtz_isClustered: trueclusterCheckinInterval: 50000useProperties: falsethreadPool:class: org.quartz.simpl.SimpleThreadPoolthreadCount: 10threadPriority: 5threadsInheritContextClassLoaderOfInitializingThread: true#数据库方式job-store-type: jdbc#初始化表结构，初始使用值为always,然后将值改为neverjdbc:initialize-schema: never

3.编写 QuartzHandler 工具类
对Quartz统一操作

Quartz API核心接口
Scheduler：(调度器)与scheduler交互的主要API；
Job：(作业)你通过scheduler执行任务，你的任务类需要实现的接口；
JobDetail：(作业实例)定义Job的实例；
Trigger：(触发器)触发Job的执行；
JobBuilder：定义和创建JobDetail实例的接口;
TriggerBuilder：定义和创建Trigger实例的接口；

public class QuartzHandler {private final static Logger log = LoggerFactory.getLogger(QuartzHandler.class);private final static String JOB_NAME_PREFIX = "JOB_";private final static String TRIGGER_GROUP_NAME = "ISIM_TRIGGER_GROUP";public final static String DEF_JOB_GROUP_NAME = "ISIM_JOB_GROUP";public final static int TASK_STATUS_STOPED = 0;// 定时任务状态-停止public final static int TASK_STATUS_RUNNING = 1;// 定时任务状态-启动/*** 获取CronTrigger* @param scheduler* @param jobId* @return*/public static CronTrigger getCronTrigger(Scheduler scheduler, Integer jobId) {try {return (CronTrigger)scheduler.getTrigger(getTriggerKey(jobId));} catch (SchedulerException e) {log.error("获取CronTrigger异常：", e);throw new QuartzJobException("获取CronTrigger异常：" + e.getMessage());}}/*** 添加定时Job* @param scheduler* @param job*/public static void addJob(Scheduler scheduler, ScheduleJob job) {try {// 构建job信息JobDetail jobDetail = JobBuilder.newJob(SysJob.class).withIdentity(getJobKey(job)).build();// 添加具体Job信息jobDetail.getJobDataMap().put(ScheduleJob.JOB_KEY, job);// Cron表达式调度构建器CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(job.getCronExpression()).withMisfireHandlingInstructionDoNothing();// 构建triggerCronTrigger trigger = TriggerBuilder.newTrigger().withIdentity(getTriggerKey(job.getId())).withSchedule(scheduleBuilder).build();// 交给scheduler调度scheduler.scheduleJob(jobDetail, trigger);// 暂停JOBif (job.getStatus() == TASK_STATUS_STOPED) {pauseJob(scheduler, job);}} catch (SchedulerException e) {log.error("添加定时Job失败：", e);throw new QuartzJobException("添加定时Job失败: " + e.getMessage());}}/*** 更新定时Job* @param scheduler* @param job*/public static void updateJob(Scheduler scheduler, ScheduleJob job) {try {TriggerKey triggerKey = getTriggerKey(job.getId());// 表达式调度构建器CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(job.getCronExpression()).withMisfireHandlingInstructionDoNothing();CronTrigger trigger = getCronTrigger(scheduler, job.getId());// Cron表达式调度构建器trigger = trigger.getTriggerBuilder().withIdentity(triggerKey).withSchedule(scheduleBuilder).build();// 添加具体Job信息trigger.getJobDataMap().put(ScheduleJob.JOB_KEY, job);// 重置jobscheduler.rescheduleJob(triggerKey, trigger);// 暂停Jobif (job.getStatus() == TASK_STATUS_STOPED) {pauseJob(scheduler, job);}} catch (SchedulerException e) {log.error("更新定时Job失败：", e);throw new QuartzJobException("更新定时Job失败：" + e.getMessage());}}/*** 立即执行定时Job* @param scheduler* @param sysJob*/public static void runJob(Scheduler scheduler, ScheduleJob job) {try {JobDataMap dataMap = new JobDataMap();dataMap.put(ScheduleJob.JOB_KEY, job);scheduler.triggerJob(getJobKey(job), dataMap);} catch (SchedulerException e) {log.error("立即执行定时Job失败：", e);throw new QuartzJobException("立即执行定时Job失败：" + e.getMessage());}}/*** 暂停定时Job*/public static void pauseJob(Scheduler scheduler, ScheduleJob job) {try {scheduler.pauseJob(getJobKey(job));} catch (SchedulerException e) {log.error("暂停定时Job失败：", e);throw new QuartzJobException("暂停定时Job失败：" + e.getMessage());}}/*** 恢复定时Job*/public static void resumeJob(Scheduler scheduler, ScheduleJob job) {try {scheduler.resumeJob(getJobKey(job));} catch (SchedulerException e) {log.error("恢复定时Job失败：", e);throw new QuartzJobException("恢复定时Job失败：" + e.getMessage());}}/*** 删除定时Job*/public static void deleteJob(Scheduler scheduler, ScheduleJob job) {try {scheduler.deleteJob(getJobKey(job));} catch (SchedulerException e) {log.error("删除定时Job失败：", e);throw new QuartzJobException("删除定时Job失败：" + e.getMessage());}}/*** 立即执行Job* @param job*/public static void execJob(ScheduleJob job) {Method method = null;try {Object jobClass = SpringContextHolder.getBean(job.getBeanName());String params = job.getParams();if (StringUtils.isNotEmpty(params)) {method = jobClass.getClass().getDeclaredMethod(job.getMethodName(), String.class);} else {method = jobClass.getClass().getDeclaredMethod(job.getMethodName());}ReflectionUtils.makeAccessible(method);// 利用反射执行Jobif (StringUtils.isNotBlank(params)) {method.invoke(jobClass, params);} else {method.invoke(jobClass);}} catch (Exception e) {e.printStackTrace();log.error("立即执行Job失败(" + job.getId() + ")", e);}}/*** 获取Trigger Key* @param jobId* @return*/private static TriggerKey getTriggerKey(Integer jobId) {return TriggerKey.triggerKey(JOB_NAME_PREFIX + jobId, TRIGGER_GROUP_NAME);}/*** 获取Job Key* @param jobId* @return*/private static JobKey getJobKey(ScheduleJob job) {String group = StringUtils.isEmpty(job.getGroupName()) ? DEF_JOB_GROUP_NAME : job.getGroupName();return JobKey.jobKey(JOB_NAME_PREFIX + job.getId(), group);}
}

4.编写 QuartzJobException 的异常类

public class QuartzJobException extends BaseRuntimeException {private static final long serialVersionUID = 1L;public QuartzJobException(String excpMsg) {super(excpMsg);}public QuartzJobException(String code, String excpMsg) {super(code, excpMsg);}
}

5.编写业务类

@Data
@EqualsAndHashCode(callSuper = false)
@Accessors(chain = true)
@TableName(schema = "admin", value = "schedule_job")
public class ScheduleJob extends BaseEntity<ScheduleJob> {private static final long serialVersionUID = 1L;@TableId(value = "id", type = IdType.AUTO)private Integer id;private String taskName;// 任务名称private String groupName;// Group名称@NotBlank(message = "Bean名称不能为空!")private String beanName;// SpringBean名称@NotBlank(message = "方法名称不能为空!")private String methodName;// 方法名称private String params;// 参数@NotBlank(message = "cron表达式不能为空!")private String cronExpression;// cron表达式private Integer status;// 任务状态private String remark;// 备注private Integer tenantId;// 租户Idpublic final static String BEAN_NAME = "bean_name";public final static String METHOD_NAME = "method_name";public final static String STATUS = "status";public static final String JOB_KEY = "JOB_KEY";// Job调度key@Overrideprotected Serializable pkVal() {return this.id;}}

6.编写controller

@RestController
@RequestMapping("/api/schedule")
@Api(tags = "Schedule")
public class ScheduleRestController extends BaseController {@Autowiredprivate ScheduleJobBiz scheduleJobBiz;/*** 定时任务-分页查询* @param scheduleJobDto* @return*/@ApiOperation("定时任务分页查询")@PostMapping("/page")public Result getJobPage(@RequestBody ScheduleJob scheduleJob) {return scheduleJobBiz.getJobPage(scheduleJob);}/*** 验证定时任务* @param ScheduleJobDto* @return*/@ApiOperation("验证定时任务")@PostMapping("/has")public Result hasJob(@RequestBody ScheduleJob scheduleJob) {if (StringUtils.isEmpty(scheduleJob.getBeanName())) {return Result.error("Bean名称不能为空");}boolean hasJob = scheduleJobBiz.hasJob(scheduleJob.getBeanName(), scheduleJob.getMethodName());return Result.success(hasJob);}/*** 定时任务-增加* @param scheduleJob* @return*/@ApiOperation("增加定时任务")@PostMapping("/add")public Result addJob(@RequestBody ScheduleJob scheduleJob) {scheduleJob.setTenantId(this.getTenantId());scheduleJob.setCreateBy(this.getUserId());Result ret;try {ValidatorUtils.validate(scheduleJob);ret = scheduleJobBiz.addJob(scheduleJob);} catch (InvalidDataException e) {ret = Result.error(e.getMessage());} catch (Exception e) {e.printStackTrace();log.error("Add job error: ", e);ret = Result.error("增加定时任务出错，请重试");}return ret;}/*** 定时任务-修改* @param scheduleJob* @return*/@ApiOperation("修改定时任务")@PostMapping("/upd")public Result updJob(@RequestBody ScheduleJob scheduleJob) {scheduleJob.setUpdateBy(this.getUserId());Result ret = null;try {ValidatorUtils.validate(scheduleJob);ret = scheduleJobBiz.updateJob(scheduleJob);} catch (InvalidDataException e) {ret = Result.error(e.getMessage());} catch (Exception e) {e.printStackTrace();log.error("Upd job error: ", e);ret = Result.error("修改定时任务出错，请重试");}return ret;}/*** 定时任务-删除* @param taskId* @return*/@ApiOperation("删除定时任务")@PostMapping("/del")public Result delJob(@RequestParam String id) {if (StringUtils.isEmpty(id)) {return Result.error("删除定时任务错，传入参数有误");}boolean result = scheduleJobBiz.deleteJob(id);return result ? Result.success("删除定时任务成功") : Result.error("删除定时任务出错，请重试");}/*** 定时任务-立即执行* @param id* @return*/@ApiOperation("立即执行定时任务")@PostMapping("/run")public Result runJob(@RequestParam String id) {if (StringUtils.isEmpty(id)) {return Result.error("执行定时任务错，传入参数有误");}scheduleJobBiz.runJob(id);return Result.success("执行定时任务成功");}/*** 定时任务-立即执行* @param id* @return*/@ApiOperation("立即执行定时任务")@PostMapping("/execute")public Result execJob(@RequestParam String id) {if (StringUtils.isEmpty(id)) {return Result.error("立即执行定时任务错，传入参数有误");}scheduleJobBiz.execJob(id);return Result.success("立即执行定时任务成功");}/*** 定时任务-暂停* @param id* @return*/@ApiOperation("暂停定时任务")@PostMapping("/pause")public Result pauseJob(@RequestParam String id) {if (StringUtils.isEmpty(id)) {return Result.error("暂停定时任务错，传入参数有误");}scheduleJobBiz.pauseJob(id, this.getUserId());return Result.success("暂停定时任务成功");}/*** 定时任务-恢复* @param id* @return*/@ApiOperation("恢复定时任务")@PostMapping("/resume")public Result resumeJob(@RequestParam String id) {if (StringUtils.isEmpty(id)) {return Result.error("恢复定时任务错，传入参数有误");}scheduleJobBiz.resumeJob(id, this.getUserId());return Result.success("恢复定时任务成功");}/*** 定时任务日志-分页查询* @param scheduleJobDto* @return*/@ApiOperation("定时任务日志分页查询")@PostMapping("/log/query")public Result queryJobLog(@RequestBody ScheduleJob scheduleJob) {return scheduleJobBiz.queryJobLogPage(scheduleJob);}
}

7.编写 biz

@Component
public class ScheduleJobBiz {@Autowiredprivate Scheduler scheduler;@Autowiredprivate ScheduleJobService scheduleJobService;@Autowiredprivate ScheduleJobLogService scheduleJobLogService;/*** 项目启动时，初始化定时器*/@PostConstructpublic void init() {List<ScheduleJob> jobList = scheduleJobService.list();for (ScheduleJob job : jobList) {CronTrigger cronTrigger = QuartzHandler.getCronTrigger(scheduler, job.getId());if (cronTrigger == null) {QuartzHandler.addJob(scheduler, job);} else {QuartzHandler.updateJob(scheduler, job);}}}/*** 定时任务分页查询* @param scheduleJob* @return*/public Result getJobPage(ScheduleJob scheduleJob) {QueryWrapper<ScheduleJob> queryWrapper = new QueryWrapper<>();if (StringUtils.isNotEmpty(scheduleJob.getBeanName())) {queryWrapper.like(ScheduleJob.BEAN_NAME, "%" + scheduleJob.getBeanName() + "%");}if (StringUtils.isNotEmpty(scheduleJob.getMethodName())) {queryWrapper.like(ScheduleJob.METHOD_NAME, "%" + scheduleJob.getMethodName() + "%");}if (null != scheduleJob.getStatus()) {queryWrapper.eq(ScheduleJob.STATUS, scheduleJob.getStatus());}Page<ScheduleJob> page = new Page<>(scheduleJob.getPage(), scheduleJob.getSize());IPage<ScheduleJob> result = scheduleJobService.page(page, queryWrapper);return Result.success(PageModel.getPageModel(result));}/*** 验证任务是否存在* @param beanName* @param methodName* @return*/public boolean hasJob(String beanName, String methodName) {return null != scheduleJobService.getScheduleJob(beanName, methodName);}/*** 增加定时任务* @param scheduleJob*/public Result addJob(ScheduleJob scheduleJob) {if (this.hasJob(scheduleJob.getBeanName(), scheduleJob.getMethodName())) {return Result.error("该任务已存在，请勿重复添加！");}if (!CronExpression.isValidExpression(scheduleJob.getCronExpression())) {return Result.error("Cron表达式不正确！");}if (StringUtils.isEmpty(scheduleJob.getGroupName())) {scheduleJob.setGroupName(QuartzHandler.DEF_JOB_GROUP_NAME);}if (StringUtils.isEmpty(scheduleJob.getParams())) {scheduleJob.setParams("{\"tenantId\":\"" + scheduleJob.getTenantId() + "\"}");}scheduleJob.setStatus(QuartzHandler.TASK_STATUS_RUNNING);scheduleJob.setCreateTime(new Date());scheduleJobService.addJob(scheduleJob);return Result.success("增加定时任务成功！");}/*** 任务修改* @param job*/public Result updateJob(ScheduleJob job) {if (null == job.getId() || StringUtils.isEmpty(job.getBeanName(), job.getMethodName(), job.getCronExpression())) {return Result.error("修改定时任务出错，传入数据有误！");}if (!CronExpression.isValidExpression(job.getCronExpression())) {return Result.error("Cron表达式不正确！");}scheduleJobService.updateJob(job);return Result.success("修改定时任务成功！");}/*** 任务删除* @param jobId* @return*/public boolean deleteJob(String jobId) {List<Integer> jobIds = StringUtils.getSplitInteger(jobId, ",");return scheduleJobService.deleteJob(jobIds);}/*** 停止定时任务* @param jobId* @param userId*/public void pauseJob(String jobId, Integer userId) {List<Integer> jobIds = StringUtils.getSplitInteger(jobId, ",");scheduleJobService.pauseJob(jobIds, userId);}/*** 恢复定时任务* @param jobId* @param userId*/public void resumeJob(String jobId, Integer userId) {List<Integer> jobIds = StringUtils.getSplitInteger(jobId, ",");scheduleJobService.resumeJob(jobIds, userId);}/*** 立即执行定时任务* @param jobId*/public void runJob(String jobId) {List<Integer> jobIds = StringUtils.getSplitInteger(jobId, ",");List<ScheduleJob> jobList = scheduleJobService.listByIds(jobIds);for (ScheduleJob scheduleJob : jobList) {QuartzHandler.runJob(scheduler, scheduleJob);}}/*** 立即执行定时任务* @param jobId*/@Asyncpublic void execJob(String jobId) {List<Integer> jobIds = StringUtils.getSplitInteger(jobId, ",");List<ScheduleJob> jobList = scheduleJobService.listByIds(jobIds);for (ScheduleJob scheduleJob : jobList) {QuartzHandler.execJob(scheduleJob);}}/*** 定时任务日志分页查询* @param scheduleJob* @return*/public Result queryJobLogPage(ScheduleJob scheduleJob) {QueryWrapper<ScheduleJobLog> queryWrapper = new QueryWrapper<>();if (StringUtils.isNotEmpty(scheduleJob.getId())) {queryWrapper.eq(ScheduleJobLog.JOB_ID, scheduleJob.getId());}if (StringUtils.isNotEmpty(scheduleJob.getBeanName())) {queryWrapper.like(ScheduleJobLog.BEAN_NAME, "%" + scheduleJob.getBeanName() + "%");}if (StringUtils.isNotEmpty(scheduleJob.getMethodName())) {queryWrapper.like(ScheduleJobLog.METHOD_NAME, "%" + scheduleJob.getMethodName() + "%");}if (null != scheduleJob.getStatus()) {queryWrapper.eq(ScheduleJobLog.STATUS, scheduleJob.getStatus());}Page<ScheduleJobLog> page = new Page<>(scheduleJob.getPage(), scheduleJob.getSize());IPage<ScheduleJobLog> result = scheduleJobLogService.page(page, queryWrapper);return Result.success(PageModel.getPageModel(result));}}

8.编写 service

@Service
public class ScheduleJobService extends ServiceImpl<ScheduleJobMapper, ScheduleJob> {@Autowiredprivate Scheduler scheduler;/*** 查询定时任务* @param beanName* @param methodName* @return*/public ScheduleJob getScheduleJob(String beanName, String methodName) {QueryWrapper<ScheduleJob> queryWrapper = new QueryWrapper<>();queryWrapper.eq(ScheduleJob.BEAN_NAME, beanName).eq(ScheduleJob.METHOD_NAME, methodName);return this.getOne(queryWrapper);}/*** 增加Job* @param scheduleJob*/@Transactional(rollbackFor = Exception.class, value = "transactionManager")public void addJob(ScheduleJob scheduleJob) {this.save(scheduleJob);QuartzHandler.addJob(scheduler, scheduleJob);}/*** 修改Job* @param scheduleJob*/@Transactional(rollbackFor = Exception.class, value = "transactionManager")public void updateJob(ScheduleJob scheduleJob) {ScheduleJob job = this.getById(scheduleJob.getId());if (null != job) {QuartzHandler.updateJob(scheduler, scheduleJob);scheduleJob.setUpdateTime(new Date());scheduleJob.updateById();}}/*** 删除Job* @param jobIds* @return*/public boolean deleteJob(List<Integer> jobIds) {List<ScheduleJob> jobList = this.listByIds(jobIds);jobList.forEach(job -> QuartzHandler.deleteJob(scheduler, job));return this.removeByIds(jobIds);}/*** 暂停运行* @param jobIds* @param userId*/@Transactional(rollbackFor = Exception.class, value = "transactionManager")public void pauseJob(List<Integer> jobIds, Integer userId) {List<ScheduleJob> jobList = this.listByIds(jobIds);for (ScheduleJob job : jobList) {QuartzHandler.pauseJob(scheduler, job);this.updScheduleJob(job.getId(), QuartzHandler.TASK_STATUS_STOPED, userId);}}/*** 恢复运行* @param jobIds* @param userId*/@Transactional(rollbackFor = Exception.class, value = "transactionManager")public void resumeJob(List<Integer> jobIds, Integer userId) {List<ScheduleJob> jobList = this.listByIds(jobIds);for (ScheduleJob job : jobList) {QuartzHandler.resumeJob(scheduler, job);this.updScheduleJob(job.getId(), QuartzHandler.TASK_STATUS_RUNNING, userId);}}/*** 更新定时任务数据库状态* @param jobId* @param status* @param userId* @return*/private ScheduleJob updScheduleJob(Integer jobId, int status, Integer userId) {ScheduleJob scheduleJob = this.getById(jobId);if (null != scheduleJob) {// 更新定时任务状态ScheduleJob upd = new ScheduleJob();upd.setId(scheduleJob.getId());upd.setStatus(status);upd.setUpdateBy(userId);upd.setUpdateTime(new Date());this.updateById(upd);}return scheduleJob;}}

9.反射调用类


@Slf4j
@Component("testTask")
public class TestTask {public void test1(String params) throws InterruptedException {log.info("TestTask test1 run start：" + params);Thread.sleep(3000);log.info("TestTask test1 run end.");}public void test2(String params) throws InterruptedException {log.info("TestTask test2 run start：" + params);Thread.sleep(6000);log.info("TestTask test2 run end.");}}

10.编写后开始测试
第一次启动项目将yml中修改一下改成always

      #数据库方式job-store-type: jdbc#初始化表结构，初始使用值为always,然后将值改为neverjdbc:initialize-schema: always

启动成功会在db 看到

掉用创建方法等待1min会看到

成功

ps：cron表达式不会写的可以参考这个 https://cron.qqe2.com/