简介
Quartz是一个完全由java编写的开源作业调度框架。不要让作业调度这个术语吓着你。尽管Quartz框架整合了许多额外功能, 但就其简易形式看,你会发现它易用得简直让人受不了!。简单地创建一个实现org.quartz.Job接口的java类。Job接口包含唯一的方法:
public void execute(JobExecutionContext context) throws JobExecutionException;
在你的Job接口实现类里面,添加一些逻辑到execute()方法。一旦你配置好Job实现类并设定好调度时间表,Quartz将密切注意剩余时间。当调度程序确定该是通知你的作业的时候,Quartz框架将调用你Job实现类(作业类)上的execute()方法并允许做它该做的事情。无需报告任何东西给调度器或调用任何特定的东西。仅仅执行任务和结束任务即可。如果配置你的作业在随后再次被调用,Quartz框架将在恰当的时间再次调用它。
单机部署问题
单机模式下的定时任务调用很简单,有很多可实现的方案,这里不多说了,例如spring schedule,java timer等。
这里说一下集群部署的情况下,定时任务的使用。这种情况下,quartz是一个比较好的选择。简单,稳定。
想象一下,现在有 A , B , C 3 台机器同时作为集群服务器对外统一提供 SERVICE :
A , B , C 3 台机器上各有一个 QUARTZ Job,它们会按照即定的 SCHEDULE 自动执行各自的任务。
先不说实现什么功能,这样的架构有点像多线程。由于三台SERVER 里都有 QUARTZ ,因此会存在重复处理 TASK 的现象。
一般外面的解决方案是只在一台 服务器上装 QUARTZ ,其它两台不装,这样的话其实就是单机了,quartz存在单点问题,一旦装有quartz的服务器宕了。服务就无法提供了。
当然还有其他一些解决方案,无非就是改 QUARTZ JOB 的代码了,这对程序开发人员来说比较痛苦;
而quartz本身提供了很好的集群方案。下面我们来说一下在spring boot下的集成:
quartz集群需要数据库的支持(JobStore TX或者JobStoreCMT),从本质上来说,是使集群上的每一个节点通过共享同一个数据库来工作的
1 准备quartz基本环境
到quartz官网下载最新的包:http://www.quartz-scheduler.org/downloads/
解压后,可以看到结构目录。在\docs\dbTables下选择合适你数据库的SQL执行文件,创建quartz集群需要的表(共11张表)
找到自己使用的数据库脚本文件执行
数据库中对应表,注意:默认情况,在windows环境下,mysql表名不区分大小写,linux下区分大小写
2 集成springboot(这里是1.5.3版本)
2.1引入依赖jar包
<dependency>
<groupId>org.quartz-scheduler</groupId> <artifactId>quartz</artifactId> <version>2.2.3</version> </dependency> <dependency> <groupId>org.quartz-scheduler</groupId> <artifactId>quartz-jobs</artifactId> <version>2.2.3</version> </dependency>2.2 quartz配置
- #quartz集群配置
- # ===========================================================================
- # Configure Main Scheduler Properties 调度器属性
- # ===========================================================================
- #调度标识名 集群中每一个实例都必须使用相同的名称
- org.quartz.scheduler.instanceName=DefaultQuartzScheduler
- #ID设置为自动获取 每一个必须不同
- org.quartz.scheduler.instanceid=AUTO
- #============================================================================
- # Configure ThreadPool
- #============================================================================
- #线程池的实现类(一般使用SimpleThreadPool即可满足几乎所有用户的需求)
- org.quartz.threadPool.class = org.quartz.simpl.SimpleThreadPool
- #指定线程数,至少为1(无默认值)(一般设置为1-100直接的整数合适)
- org.quartz.threadPool.threadCount = 25
- #设置线程的优先级(最大为java.lang.Thread.MAX_PRIORITY 10,最小为Thread.MIN_PRIORITY 1,默认为5)
- org.quartz.threadPool.threadPriority = 5
- #============================================================================
- # Configure JobStore
- #============================================================================
- # 信息保存时间 默认值60秒
- org.quartz.jobStore.misfireThreshold = 60000
- #数据保存方式为数据库持久化
- org.quartz.jobStore.class = org.quartz.impl.jdbcjobstore.JobStoreTX
- #数据库代理类,一般org.quartz.impl.jdbcjobstore.StdJDBCDelegate可以满足大部分数据库
- org.quartz.jobStore.driverDelegateClass = org.quartz.impl.jdbcjobstore.StdJDBCDelegate
- #JobDataMaps是否都为String类型
- org.quartz.jobStore.useProperties = false
- #数据库别名 随便取
- org.quartz.jobStore.dataSource = myDS
- #表的前缀,默认QRTZ_
- org.quartz.jobStore.tablePrefix = QRTZ_
- #是否加入集群
- org.quartz.jobStore.isClustered = true
- #调度实例失效的检查时间间隔
- org.quartz.jobStore.clusterCheckinInterval = 20000
- #============================================================================
- # Configure Datasources
- #============================================================================
- #数据库引擎
- org.quartz.dataSource.myDS.driver = com.mysql.jdbc.Driver
- #数据库连接
- org.quartz.dataSource.myDS.URL = jdbc:mysql://172.30.12.14:7001/rbl_test?useUnicode=true&characterEncoding=utf-8&useSSL=false&allowMultiQueries=true
- #数据库用户
- org.quartz.dataSource.myDS.user = root
- #数据库密码
- org.quartz.dataSource.myDS.password = 123456
- #允许最大连接
- org.quartz.dataSource.myDS.maxConnections = 5
- #验证查询sql,可以不设置
- org.quartz.dataSource.myDS.validationQuery=select 0 from dual
注:如果嫌需要额外配置quart数据源很烦,也可以共用你项目配置的数据库链接,这样每次更换数据库连接,就不需要额外在修改。
2.3 springboot configuration 配置
直接使用springboot注入的的datasource的内容配置quartz的数据库连接
quartz属性配置可以读取配置文件读取,我这里没在配置文件读取,直接写在代码里测试的,直接新建一个配置文件里面写quartz的配置内容,然后通过springboot注入属性进来。
package com.kerry.config;import java.io.IOException;import java.util.Properties;import javax.sql.DataSource;import org.quartz.Scheduler;import org.quartz.SchedulerException;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.scheduling.quartz.SchedulerFactoryBean;/** * 分布式定时任务管理配置 * @author kerry * @date 2018-05-09 11:36:21 */@Configuration//@ConditionalOnProperty(prefix = "qybd", name = "quartz-open", havingValue = "true")public class QuartzConfig{ @Autowired DataSource dataSource; @Bean public SchedulerFactoryBean schedulerFactoryBean(QuartzJobFactory myJobFactory) throws Exception { SchedulerFactoryBean schedulerFactoryBean = new SchedulerFactoryBean(); schedulerFactoryBean.setDataSource(dataSource); //使job实例支持spring 容器管理 schedulerFactoryBean.setOverwriteExistingJobs(true); schedulerFactoryBean.setJobFactory(myJobFactory); schedulerFactoryBean.setQuartzProperties(quartzProperties()); // 延迟10s启动quartz schedulerFactoryBean.setStartupDelay(10); return schedulerFactoryBean; } @Bean public Scheduler scheduler(SchedulerFactoryBean schedulerFactoryBean) throws IOException, SchedulerException {// SchedulerFactory schedulerFactory = new StdSchedulerFactory(quartzProperties());// Scheduler scheduler = schedulerFactory.getScheduler();// scheduler.start();//初始化bean并启动scheduler Scheduler scheduler = schedulerFactoryBean.getScheduler(); scheduler.start(); return scheduler; } /** * 设置quartz属性 */ public Properties quartzProperties() throws IOException { Properties prop = new Properties(); prop.put("quartz.scheduler.instanceName", "ServerScheduler"); prop.put("org.quartz.scheduler.instanceId", "AUTO"); prop.put("org.quartz.scheduler.skipUpdateCheck", "true"); prop.put("org.quartz.scheduler.instanceId", "NON_CLUSTERED"); prop.put("org.quartz.scheduler.jobFactory.class", "org.quartz.simpl.SimpleJobFactory"); prop.put("org.quartz.jobStore.class", "org.quartz.impl.jdbcjobstore.JobStoreTX"); prop.put("org.quartz.jobStore.driverDelegateClass", "org.quartz.impl.jdbcjobstore.StdJDBCDelegate"); prop.put("org.quartz.jobStore.dataSource", "quartzDataSource"); prop.put("org.quartz.jobStore.tablePrefix", "QRTZ_"); prop.put("org.quartz.jobStore.isClustered", "true"); prop.put("org.quartz.threadPool.class", "org.quartz.simpl.SimpleThreadPool"); prop.put("org.quartz.threadPool.threadCount", "5");// prop.put("org.quartz.dataSource.quartzDataSource.driver", druidProperties.getDriverClassName());// prop.put("org.quartz.dataSource.quartzDataSource.URL", druidProperties.getUrl());// prop.put("org.quartz.dataSource.quartzDataSource.user", druidProperties.getUsername());// prop.put("org.quartz.dataSource.quartzDataSource.password", druidProperties.getPassword());// prop.put("org.quartz.dataSource.quartzDataSource.maxConnections", druidProperties.getMaxActive()); return prop; } }
注意上面的schedulerFactoryBean.setJobFactory(myJobFactory); //这个myJobFactory是自定义配置的一个类,如果这里不配置这个jobFactory,下面的那个CtripScenicTask会为空,获取不了注入对象
@Component
public class CtripScenicJob implements Job{ private Logger logger = LoggerFactory.getLogger(CtripScenicJob.class);@Autowired
private CtripScenicTask ctripScenicTask;这个类主要解决spring管理的Quartz job里面注入不了其他bean
package com.kerry.config;import org.quartz.spi.TriggerFiredBundle;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.beans.factory.config.AutowireCapableBeanFactory;import org.springframework.scheduling.quartz.AdaptableJobFactory;import org.springframework.stereotype.Component;@Componentpublic class QuartzJobFactory extends AdaptableJobFactory { //这个对象Spring会帮我们自动注入进来,也属于Spring技术范畴. @Autowired private AutowireCapableBeanFactory capableBeanFactory; protected Object createJobInstance(TriggerFiredBundle bundle) throws Exception { //调用父类的方法 Object jobInstance = super.createJobInstance(bundle); //进行注入,这属于Spring的技术,不清楚的可以查看Spring的API. capableBeanFactory.autowireBean(jobInstance); return jobInstance; }}
2.4 动态配置管理quartz
接口类
package com.kerry.modular.biz.service;import java.util.List;import org.quartz.SchedulerException;import com.kerry.modular.biz.model.TaskInfo;public interface TaskService { Listlist(); void addJob(TaskInfo info); void edit(TaskInfo info); void delete(String jobName, String jobGroup); void pause(String jobName, String jobGroup); void resume(String jobName, String jobGroup); boolean checkExists(String jobName, String jobGroup)throws SchedulerException;}
实现类
package com.kerry.modular.biz.service;import java.util.ArrayList;import java.util.Date;import java.util.HashSet;import java.util.List;import org.apache.commons.lang3.time.DateFormatUtils;import org.quartz.CronScheduleBuilder;import org.quartz.CronTrigger;import org.quartz.Job;import org.quartz.JobBuilder;import org.quartz.JobDetail;import org.quartz.JobKey;import org.quartz.Scheduler;import org.quartz.SchedulerException;import org.quartz.Trigger;import org.quartz.TriggerBuilder;import org.quartz.TriggerKey;import org.quartz.impl.matchers.GroupMatcher;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Service;import com.kerry.modular.biz.model.TaskInfo;@Servicepublic class TaskServiceImpl implements TaskService { private Logger logger = LoggerFactory.getLogger(TaskServiceImpl.class); @Autowired(required=false) private Scheduler scheduler; /** * 所有任务列表 */ public Listlist(){ List list = new ArrayList<>(); try { for(String groupJob: scheduler.getJobGroupNames()){ for(JobKey jobKey: scheduler.getJobKeys(GroupMatcher. groupEquals(groupJob))){ List triggers = scheduler.getTriggersOfJob(jobKey); for (Trigger trigger: triggers) { Trigger.TriggerState triggerState = scheduler.getTriggerState(trigger.getKey()); JobDetail jobDetail = scheduler.getJobDetail(jobKey); String cronExpression = "", createTime = ""; if (trigger instanceof CronTrigger) { CronTrigger cronTrigger = (CronTrigger) trigger; cronExpression = cronTrigger.getCronExpression(); createTime = cronTrigger.getDescription(); } TaskInfo info = new TaskInfo(); info.setJobName(jobKey.getName()); info.setJobGroup(jobKey.getGroup()); info.setJobDescription(jobDetail.getDescription()); info.setJobStatus(triggerState.name()); info.setCronExpression(cronExpression); info.setCreateTime(createTime); list.add(info); } } } } catch (SchedulerException e) { e.printStackTrace(); } return list; } /** * 保存定时任务 * @param info */ @SuppressWarnings("unchecked") public void addJob(TaskInfo info) { String jobName = info.getJobName(), jobGroup = info.getJobGroup(), cronExpression = info.getCronExpression(), jobDescription = info.getJobDescription(), createTime = DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss"); try { if (checkExists(jobName, jobGroup)) { logger.info("add job fail, job already exist, jobGroup:{}, jobName:{}", jobGroup, jobName); } TriggerKey triggerKey = TriggerKey.triggerKey(jobName, jobGroup); JobKey jobKey = JobKey.jobKey(jobName, jobGroup); CronScheduleBuilder schedBuilder = CronScheduleBuilder.cronSchedule(cronExpression).withMisfireHandlingInstructionDoNothing(); CronTrigger trigger = TriggerBuilder.newTrigger().withIdentity(triggerKey).withDescription(createTime).withSchedule(schedBuilder).build(); Class clazz = (Class )Class.forName(jobName); JobDetail jobDetail = JobBuilder.newJob(clazz).withIdentity(jobKey).withDescription(jobDescription).build(); scheduler.scheduleJob(jobDetail, trigger); } catch (SchedulerException | ClassNotFoundException e) { logger.error("类名不存在或执行表达式错误,exception:{}",e.getMessage()); } } /** * 修改定时任务 * @param info */ public void edit(TaskInfo info) { String jobName = info.getJobName(), jobGroup = info.getJobGroup(), cronExpression = info.getCronExpression(), jobDescription = info.getJobDescription(), createTime = DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss"); try { if (!checkExists(jobName, jobGroup)) { logger.info("edit job fail, job is not exist, jobGroup:{}, jobName:{}", jobGroup, jobName); } TriggerKey triggerKey = TriggerKey.triggerKey(jobName, jobGroup); JobKey jobKey = new JobKey(jobName, jobGroup); CronScheduleBuilder cronScheduleBuilder = CronScheduleBuilder.cronSchedule(cronExpression).withMisfireHandlingInstructionDoNothing(); CronTrigger cronTrigger = TriggerBuilder.newTrigger().withIdentity(triggerKey).withDescription(createTime).withSchedule(cronScheduleBuilder).build(); JobDetail jobDetail = scheduler.getJobDetail(jobKey); jobDetail.getJobBuilder().withDescription(jobDescription); HashSet triggerSet = new HashSet<>(); triggerSet.add(cronTrigger); scheduler.scheduleJob(jobDetail, triggerSet, true); } catch (SchedulerException e) { logger.error("类名不存在或执行表达式错误,exception:{}",e.getMessage()); } } /** * 删除定时任务 * @param jobName * @param jobGroup */ public void delete(String jobName, String jobGroup){ TriggerKey triggerKey = TriggerKey.triggerKey(jobName, jobGroup); try { if (checkExists(jobName, jobGroup)) { scheduler.pauseTrigger(triggerKey); scheduler.unscheduleJob(triggerKey); logger.info("delete job, triggerKey:{},jobGroup:{}, jobName:{}", triggerKey ,jobGroup, jobName); } } catch (SchedulerException e) { logger.error(e.getMessage()); } } /** * 暂停定时任务 * @param jobName * @param jobGroup */ public void pause(String jobName, String jobGroup){ TriggerKey triggerKey = TriggerKey.triggerKey(jobName, jobGroup); try { if (checkExists(jobName, jobGroup)) { scheduler.pauseTrigger(triggerKey); logger.info("pause job success, triggerKey:{},jobGroup:{}, jobName:{}", triggerKey ,jobGroup, jobName); } } catch (SchedulerException e) { logger.error(e.getMessage()); } } /** * 重新开始任务 * @param jobName * @param jobGroup */ public void resume(String jobName, String jobGroup){ TriggerKey triggerKey = TriggerKey.triggerKey(jobName, jobGroup); try { if (checkExists(jobName, jobGroup)) { scheduler.resumeTrigger(triggerKey); logger.info("resume job success,triggerKey:{},jobGroup:{}, jobName:{}", triggerKey ,jobGroup, jobName); } } catch (SchedulerException e) { logger.error(e.getMessage()); } } /** * 验证是否存在 * @param jobName * @param jobGroup * @throws SchedulerException */ public boolean checkExists(String jobName, String jobGroup) throws SchedulerException{ TriggerKey triggerKey = TriggerKey.triggerKey(jobName, jobGroup); return scheduler.checkExists(triggerKey); }}
taskinfo实体类
package com.kerry.modular.biz.model;import java.io.Serializable;public class TaskInfo implements Serializable{ private static final long serialVersionUID = -8054692082716173379L; private int id = 0; /**任务名称*/ private String jobName; /**任务分组*/ private String jobGroup; /**任务描述*/ private String jobDescription; /**任务状态*/ private String jobStatus; /**任务表达式*/ private String cronExpression; private String createTime; public String getJobName() { return jobName; } public void setJobName(String jobName) { this.jobName = jobName; } public String getJobGroup() { return jobGroup; } public void setJobGroup(String jobGroup) { this.jobGroup = jobGroup; } public String getJobDescription() { return jobDescription; } public void setJobDescription(String jobDescription) { this.jobDescription = jobDescription; } public String getJobStatus() { return jobStatus; } public void setJobStatus(String jobStatus) { this.jobStatus = jobStatus; } public String getCronExpression() { return cronExpression; } public void setCronExpression(String cronExpression) { this.cronExpression = cronExpression; } public String getCreateTime() { return createTime; } public void setCreateTime(String createTime) { this.createTime = createTime; } public int getId() { return id; } public void setId(int id) { this.id = id; }}
任务管理Controller类
package com.kerry.modular.biz.controller;import java.util.HashMap;import java.util.List;import java.util.Map;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Controller;import org.springframework.web.bind.annotation.PathVariable;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.ResponseBody;import com.alibaba.fastjson.JSON;import com.kerry.modular.biz.model.TaskInfo;import com.kerry.modular.biz.service.TaskService;/** * 任务管理 */@Controller@RequestMapping("/qy/api/task/")public class TaskManageController { @Autowired(required=false) private TaskService taskService; /** * Index.jsp */ @RequestMapping(value={"", "/", "index"}) public String info(){ return "index.jsp"; } /** * 任务列表 * @return */ @ResponseBody @RequestMapping(value="list") public String list(){ Mapmap = new HashMap<>(); List infos = taskService.list(); map.put("rows", infos); map.put("total", infos.size()); return JSON.toJSONString(map); } /** * 保存定时任务 * @param info */ @ResponseBody @RequestMapping(value="save", produces = "application/json; charset=UTF-8") public String save(TaskInfo info){ try { if(info.getId() == 0) { taskService.addJob(info); }else{ taskService.edit(info); } } catch (Exception e) { return e.getMessage(); } return "成功"; } /** * 删除定时任务 * @param jobName * @param jobGroup */ @ResponseBody @RequestMapping(value="delete/{jobName}/{jobGroup}", produces = "application/json; charset=UTF-8") public String delete(@PathVariable String jobName, @PathVariable String jobGroup){ try { taskService.delete(jobName, jobGroup); } catch (Exception e) { return e.getMessage(); } return "成功"; } /** * 暂停定时任务 * @param jobName * @param jobGroup */ @ResponseBody @RequestMapping(value="pause/{jobName}/{jobGroup}", produces = "application/json; charset=UTF-8") public String pause(@PathVariable String jobName, @PathVariable String jobGroup){ try { taskService.pause(jobName, jobGroup); } catch (Exception e) { return e.getMessage(); } return "成功"; } /** * 重新开始定时任务 * @param jobName * @param jobGroup */ @ResponseBody @RequestMapping(value="resume/{jobName}/{jobGroup}", produces = "application/json; charset=UTF-8") public String resume(@PathVariable String jobName, @PathVariable String jobGroup){ try { taskService.resume(jobName, jobGroup); } catch (Exception e) { return e.getMessage(); } return "成功"; }}
任务实现类实现job接口,重写execute方法
package com.kerry.modular.biz.task.quartz;import org.quartz.Job;import org.quartz.JobExecutionContext;import org.quartz.JobExecutionException;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;import com.kerry.modular.biz.task.schedule.CtripScenicTask;@Componentpublic class CtripScenicJob implements Job{ private Logger logger = LoggerFactory.getLogger(CtripScenicJob.class); @Autowired private CtripScenicTask ctripScenicTask; @Override public void execute(JobExecutionContext context) throws JobExecutionException { logger.info("JobName: {}", context.getJobDetail().getKey().getName()); ctripScenicTask.loadComment(); } }
此时可以通过调用TaskManageController时间动态控制定时任务
3 测试,启动springboot项目
输入添加任务的url:
http://localhost:8080/qy/api/task/save?jobName=com.stylefeng.guns.modular.biz.task.quartz.CtripHotelJob&jobGroup=group1&jobDescription=job描述&cronExpression=0/10 * * * * ?
jobName为job类的包名类名,jobGroup该任务所属组,jobDescription 描述,cronExpression :core表达式
上面的请求会添加一个定时任务,每10秒执行一次 CtripHotelJob里面的execute方法。
保存的定时任务会在quartz相关表里保存数据
如: