Java定时任务(二)：quartz调度框架
2018-06-13 本文已影响0人
sawyerlsy
Quartz是OpenSymphony开源组织在任务调度领域的一个开源项目，完全基于Java实现。Quartz具备以下特点：
- 强大的调度功能，例如支持丰富多样的调度方法，可以满足各种常规及特殊需求
- 灵活的应用方式，例如支持任务和调度的多种组合方式，支持调度数据的多种存储方式；
- 集群能力
- 能够很轻易的就与spring集成在一起
一、基础配置
1. 引入jar包
<!-- quartz -->
<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. xml配置
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
<!-- 1：定义任务的bean ，这里使用JobDetailFactoryBean,也可以使用MethodInvokingJobDetailFactoryBean-->
<bean name="testJob" class="org.springframework.scheduling.quartz.JobDetailFactoryBean">
<property name="name" value="test_job"/>
<property name="group" value="test_group"/>
<property name="jobClass" value="com.sawyer.job.TestJob"/>
</bean>
<!--<bean name="simpleTrigger" class="org.springframework.scheduling.quartz.SimpleTriggerFactoryBean">
<property name="name" value="test_trigger"/>
<property name="group" value="test_trigger_group"/>
<property name="jobDetail" ref="testJob"/>
<property name="startDelay" value="1000"/>
<property name="repeatInterval" value="5000"/>
<property name="repeatCount" value="15"/>
</bean>-->
<!-- 定义触发器，并与job关联 -->
<bean id="cronTrigger" class="org.springframework.scheduling.quartz.CronTriggerFactoryBean">
<property name="name" value="test_trigger"/>
<property name="group" value="test_trigger_group"/>
<property name="jobDetail" ref="testJob"/>
<property name="startDelay" value="3000"/>
<property name="cronExpression" value="0/5 * * * * ?"/>
</bean>
<!-- 定义调度器，并将Trigger注册到调度器中 -->
<bean id="schedulers" class="org.springframework.scheduling.quartz.SchedulerFactoryBean">
<property name="jobFactory">
<bean class="com.sawyer.job.AutowiringSpringBeanJobFactory"></bean>
</property>
<property name="applicationContextSchedulerContextKey" value="applicationContext"/>
<property name="triggers">
<list>
<!--<ref bean="simpleTrigger"/>-->
<ref bean="cronTrigger"/>
</list>
</property>
<property name="autoStartup" value="true"/>
</bean>
</beans>
使用quartz需要配置job、trigger、scheduler,其中：
job：任务的执行类，需要在xml中指明名称、组别和类
trigger：任务的触发器，需要指明名称、组别以及关联的job。一个触发器只能对应一个job；触发器有两种，分别为CronTriggerFactoryBean和SimpleTriggerFactoryBean，前者支持cron表达式，后者只支持一些简单的配置。
scheduler：调度器，需要将trigger配置在scheduler的triggers中，可以配置多个。这里自定义实现了jobFactory，可以在job中自动注入spring bean；applicationContextSchedulerContextKey属性用于在job中获取spring 的上下文
3. 代码清单
3.1. 定义SpringBeanJobFactory
package com.sawyer.job;
import org.quartz.spi.TriggerFiredBundle;
import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.scheduling.quartz.SpringBeanJobFactory;
public class AutowiringSpringBeanJobFactory extends SpringBeanJobFactory implements ApplicationContextAware {
private transient AutowireCapableBeanFactory beanFactory;
public void setApplicationContext(final ApplicationContext context) {
beanFactory = context.getAutowireCapableBeanFactory();
}
@Override
public Object createJobInstance(final TriggerFiredBundle bundle) throws Exception {
final Object job = super.createJobInstance(bundle);
beanFactory.autowireBean(job);
return job;
}
}
自定义实现了JobFactory，就可以在job中自动注入bean
3.2 AbstractJob
package com.sawyer.job;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
/**
* 抽象类，job需要继承该类,通过这种方式可以做一些自定义处理
*/
public abstract class AbstractJob implements Job {
@Override
public final void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
try {
safeExecute(jobExecutionContext);
} catch (Exception e) {
JobExecutionException e2 = new JobExecutionException(e);
if (!ignoreException()) {
if (isRefireImmediatelyWhenException()) {
//立即重新运行当前job
e2.setRefireImmediately(true);
} else {
//立即停止与当前Job有关的所有触发器，当前job不会再运行
e2.setUnscheduleAllTriggers(true);
}
}
throw e2;
}
}
public abstract void safeExecute(JobExecutionContext context) throws Exception;
/**
* 是否忽略job运行时产生的异常
*/
public abstract boolean ignoreException();
/**
* 发生异常时是否立即重新执行JOB或将JOB挂起.
* <p>
*
* @return {@code true} Job运行产生异常时，立即重新执行JOB. <br>
* {@code false} Job运行产生异常时，挂起JOB等候管理员处理.
*/
public abstract boolean isRefireImmediatelyWhenException();
}
AbstractJob是一个抽象类，凡是job都应该继承该类，通过这种方式可以job做一些自定义处理。如图中代码所示，AbstractJob中有4个方法：
ignoreException：是否忽略异常，返回true时忽略异常，否则必须处理。
isRefireImmediatelyWhenException：当出现异常时，是否立即重新执行。返回true时，立即重新执行，否则将会挂起所有与该job有关的trigger，不再执行。
safeExecute：子类需要实现该方法，在方法中定义具体的实现逻辑。
execute：接口job的实现方法，job最终是在该方法中执行的。只有这个方法才是Job自带的，其他的都是自定义方法。
3.3 TestJob
package com.sawyer.job;
import org.quartz.DisallowConcurrentExecution;
import org.quartz.JobDataMap;
import org.quartz.JobExecutionContext;
import org.quartz.PersistJobDataAfterExecution;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.data.redis.core.RedisTemplate;
import java.util.Date;
@PersistJobDataAfterExecution
@DisallowConcurrentExecution
public class TestJob extends AbstractJob {
private static final String COUNT_KEY = "count";
private static final String APPLICATION_CONTEXT_KEY = "applicationContext";
@Autowired
private RedisTemplate redisTemplate;
@Override
public void safeExecute(JobExecutionContext context) throws Exception {
//统计执行次数
JobDataMap jobDataMap = context.getJobDetail().getJobDataMap();
int count = 0;
if (jobDataMap.containsKey(COUNT_KEY)) {
count = (int) jobDataMap.get(COUNT_KEY);
}
count++;
jobDataMap.put(COUNT_KEY, count);
//打印出redisTemplate，测试是否能自动注入
System.out.println("autowiring spring bean ：" + redisTemplate);
//获取spring上下文
ApplicationContext applicationContext = (ApplicationContext) context.getScheduler().getContext().get(APPLICATION_CONTEXT_KEY);
System.out.println("spring context :" + applicationContext);
if (10 == count) {
int m = 1 / 0;
}
System.out.println("********current date :" + new Date() + " and the thread is :" + Thread.currentThread().getName() + " and the count is :" + count + " **********");
}
@Override
public boolean isRefireImmediatelyWhenException() {
return false;
}
@Override
public boolean ignoreException() {
return false;
}
}
Quartz默认是支持并发的，即上一个任务未完成的时候就开始了下一个任务，并且每次都是生成一个新的job实例。图中例子由于需要共享count，统计次数，所在TestJob上面加了两个注解@PersistJobDataExecution和@DisallowConcurrentExecution，这是用于持久化JobData和禁用并发的。加了这两个注解就能够让testJob成为一个有状态的job，并且上一次任务执行完后，才开始下一次任务，这样一来就可以通过jobDataMap共享数据，反之，如果不加注解的话，job就是一个无状态的job，每次运行时都会产生一个新的job实例，数据无法共享。而禁用并发是为了防止并发产生数据紊乱的问题。
TestJob中isRefireImmediatelyWhenException和ignoreException都为false，那么当count等于10时，会抛出异常，系统就会将于当前job有关有的所有trigger挂起，不再执行job。
二、Quartz 存储与持久化
Quartz提供了两种作业存储类型，分别为RAMJobSore和JDBCJobStore。
RASMJobStore：将作业的调度信息存储到内存中，不需要配置外置数据库，配置简单，运行速度快；但是由于调度信息存储在内存中，当应用程序停止时，作业的调度信息将会丢失，此外一旦作业运行期间崩溃，将无法恢复事故现场，比如原定执行30次，执行到第15次是崩溃了，那么系统重启时，将会从0开始。
JDBCJobStore：将作业的调度信息存储到数据库中，该种方式支持集群，调度信息不会丢失，并且可以手动恢复意外停止的job；但是这种方式会较为复杂。
Quartz默认使用的就是RAMJobStore，下面开始介绍持久化配置
1. 数据表初始化
- 从官网中下载quartz包：http://www.quartz-scheduler.org/downloads
- 在quartz-2.2.3/docs/dbTables目录下找到与数据库对应的sql文件，我使用的是mysql数据库，所以这里选择tables.mysql.sql文件，在数据库中执行
2. 引入 quartz.properties ，并根据业务需要进行配置
org.quartz.jobStore.isClustered = true
org.quartz.jobStore.clusterCheckinInterval = 20000
org.quartz.scheduler.instanceId = AUTO
org.quartz.scheduler.skipUpdateCheck = true
# 集群配置
org.quartz.scheduler.instanceName = DefaultQuartzScheduler
org.quartz.scheduler.rmi.export = false
org.quartz.scheduler.rmi.proxy = false
org.quartz.scheduler.wrapJobExecutionInUserTransaction = false
# 线程池的实现类
org.quartz.threadPool.class = org.quartz.simpl.SimpleThreadPool
# 线程数量
org.quartz.threadPool.threadCount = 10
# 线程优先级，最大值为10，最小值为1
org.quartz.threadPool.threadPriority = 5
#
org.quartz.threadPool.threadsInheritContextClassLoaderOfInitializingThread = true
# 持久化配置
#org.quartz.jobStore.class = org.quartz.simpl.RAMJobStore
org.quartz.jobStore.class = org.quartz.impl.jdbcjobstore.JobStoreTX
#org.quartz.jobStore.driverDelegateClass=org.quartz.impl.jdbcjobstore.HSQLDBDelegate
org.quartz.jobStore.driverDelegateClass=org.quartz.impl.jdbcjobstore.StdJDBCDelegate
org.quartz.scheduler.classLoadHelper.class=org.quartz.simpl.CascadingClassLoadHelper
#org.quartz.jobStore.useProperties = true
org.quartz.jobStore.tablePrefix = QRTZ_
org.quartz.jobStore.maxMisfiresToHandleAtATime=1
org.quartz.jobStore.selectWithLockSQL=SELECT * FROM {0}LOCKS UPDLOCK WHERE LOCK_NAME = ?
org.quartz.jobStore.misfireThreshold = 60000
#============================================================================
# 配置插件
#============================================================================
#org.quartz.plugin.triggHistory.class=org.quartz.plugins.history.LoggingJobHistoryPlugin
org.quartz.plugin.runningListener.class=com.sawyer.job.RunningListenerPlugin
org.quartz.plugin.runningListener.LogRunningInfo=true
如上图所示,在quartz.properties中，配置了一个自定义的插件：com.sawyer.job.RunningListenerPlugin，可以在这里做一些比较有意义的事情。另外就是，我并没有将数据库信息配置在quart.properties中，而是选择另行配置，目的为了将数据库信息集中到一起，方便操作，实际项目中，也推荐使用这种方式
3. application-datasource.xml
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context" xmlns:tx="http://www.springframework.org/schema/tx"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd">
<context:property-placeholder location="classpath:/config.properties" local-override="true"/>
<tx:annotation-driven transaction-manager="transactionManager"/>
<context:component-scan base-package="**.*.service"/>
<context:component-scan base-package="**.*.dao"/>
<bean id="dataSource" class="com.alibaba.druid.pool.DruidDataSource" init-method="init" destroy-method="close">
<property name="url" value="${jdbc.url}"/>
<property name="username" value="${jdbc.user}"/>
<property name="password" value="${jdbc.password}"/>
<property name="filters" value="stat"/>
<property name="maxActive" value="20"/>
<property name="initialSize" value="1"/>
<property name="maxWait" value="60000"/>
<property name="minIdle" value="1"/>
<property name="timeBetweenEvictionRunsMillis" value="60000"/>
<property name="minEvictableIdleTimeMillis" value="300000"/>
<property name="testWhileIdle" value="true"/>
<property name="testOnBorrow" value="false"/>
<property name="testOnReturn" value="false"/>
<property name="poolPreparedStatements" value="true"/>
<property name="maxOpenPreparedStatements" value="20"/>
</bean>
<!-- config mysql statements -->
<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<property name="dataSource" ref="dataSource"/>
</bean>
</beans>
4. application-job.xml
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
<context:property-placeholder location="classpath:config.properties"/>
<bean name="quartzScheduler" class="org.springframework.scheduling.quartz.SchedulerFactoryBean">
<property name="dataSource" ref="dataSource"/>
<property name="autoStartup" value="${job.autoStartup}"/>
<property name="jobFactory">
<bean class="com.sawyer.job.AutowiringSpringBeanJobFactory"/>
</property>
<property name="applicationContextSchedulerContextKey" value="applicationContext"/>
<property name="configLocation" value="classpath:quartz.properties"/>
</bean>
</beans>
这一步的scheduler配置与上文的基本一致，只是配置了dataSource和引入了quartz.properties文件
5. 自定义插件 RunningListenerPlugin
package com.sawyer.job;
import org.quartz.ListenerManager;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.impl.matchers.EverythingMatcher;
import org.quartz.spi.ClassLoadHelper;
import org.quartz.spi.SchedulerPlugin;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
public class RunningListenerPlugin implements SchedulerPlugin {
private final Logger log = LoggerFactory.getLogger(getClass());
private Scheduler scheduler;
private boolean isLogRunningInfo;
@Override
public void initialize(String s, Scheduler scheduler, ClassLoadHelper classLoadHelper) throws SchedulerException {
this.scheduler = scheduler;
System.out.println("################## 调度器初始化 ##################");
}
@Override
public void start() {
System.out.println("################## 调度器启动 ###################");
try {
ApplicationContext applicationContext = (ApplicationContext) scheduler.getContext().get("applicationContext");
System.out.println("获取到的 ApplicationContext ：" + applicationContext);
ListenerManager listenerManager = scheduler.getListenerManager();
if (isLogRunningInfo()) {
listenerManager.addJobListener(new JobRunningListener(applicationContext), EverythingMatcher.allJobs());
listenerManager.addSchedulerListener(new SchedulerRunningListener(applicationContext));
}
} catch (SchedulerException e) {
throw new RuntimeException(e);
}
}
@Override
public void shutdown() {
System.out.println("############# 调度器关闭 ##################");
}
public boolean isLogRunningInfo() {
return isLogRunningInfo;
}
public void setLogRunningInfo(boolean logRunningInfo) {
isLogRunningInfo = logRunningInfo;
}
}
可以添加多个监听器，实现各种各样的业务需求，比如用于任务结束后发邮件，记录更为详细的运行信息等。只需要实现响应的接口即可，这里我选择了添加JobRunningListener和SchedulerRunningListener
6. JobRunningListener
package com.sawyer.job;
import org.quartz.*;
import org.springframework.context.ApplicationContext;
public class JobRunningListener implements JobListener {
private final static String LISTENER_NAME = "JobRunningListener";
private ApplicationContext applicationContext;
public JobRunningListener(ApplicationContext applicationContext) {
this.applicationContext = applicationContext;
}
@Override
public String getName() {
return LISTENER_NAME;
}
@Override
public void jobToBeExecuted(JobExecutionContext jobExecutionContext) {
System.out.println("######### job 准备开始执行 ##########");
}
@Override
public void jobExecutionVetoed(JobExecutionContext jobExecutionContext) {
System.out.println("######### job 被否决了 ##########");
}
@Override
public void jobWasExecuted(JobExecutionContext jobExecutionContext, JobExecutionException e) {
System.out.println("######### job 执行完毕 ##########");
}
}
7. SchdulerRunningListener
package com.sawyer.job;
import org.quartz.*;
import org.springframework.context.ApplicationContext;
public class SchedulerRunningListener implements SchedulerListener {
private final ApplicationContext applicationContext;
public SchedulerRunningListener(ApplicationContext applicationContext) {
this.applicationContext = applicationContext;
}
@Override
public void jobScheduled(Trigger trigger) {
}
@Override
public void jobUnscheduled(TriggerKey triggerKey) {
}
@Override
public void triggerFinalized(Trigger trigger) {
}
@Override
public void triggerPaused(TriggerKey triggerKey) {
}
@Override
public void triggersPaused(String s) {
}
@Override
public void triggerResumed(TriggerKey triggerKey) {
}
@Override
public void triggersResumed(String s) {
}
@Override
public void jobAdded(JobDetail jobDetail) {
}
@Override
public void jobDeleted(JobKey jobKey) {
System.out.println("############ " + jobKey.getGroup() + " 下的 " + jobKey.getName() + "已被删除 ###################");
}
@Override
public void jobPaused(JobKey jobKey) {
}
@Override
public void jobsPaused(String s) {
}
@Override
public void jobResumed(JobKey jobKey) {
}
@Override
public void jobsResumed(String s) {
}
@Override
public void schedulerError(String s, SchedulerException e) {
}
@Override
public void schedulerInStandbyMode() {
}
@Override
public void schedulerStarted() {
}
@Override
public void schedulerStarting() {
}
@Override
public void schedulerShutdown() {
}
@Override
public void schedulerShuttingdown() {
}
@Override
public void schedulingDataCleared() {
}
}
8. 执行job
package com.sawyer;
import com.sawyer.job.AbstractJob;
import org.quartz.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
public class Main {
private static final Logger logger = LoggerFactory.getLogger(Main.class);
public static void main(String[] args) {
String configLocation = "classpath:spring/applicationContext*.xml";
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(configLocation);
try {
createJob(context);
} catch (SchedulerException e) {
e.printStackTrace();
}
logger.info("Container has been startup.");
}
public static void createJob(ApplicationContext context) throws SchedulerException {
final String jobClassName = "com.sawyer.job.TestJob";
final String jobName = "test_job";
final String jobGroup = "test_group";
final String jobDescription = "这是TestJob的描述信息";
final String triggerName = "test_trigger";
final String triggerGroup = "test_trigger_group";
final String cron = "0/5 * * * * ?";
// 加载job类，并判断 job类的父类是否为AbstractJob
boolean assignableFrom = false;
Class forName = null;
try {
forName = Class.forName(jobClassName);
assignableFrom = AbstractJob.class.isAssignableFrom(forName);
} catch (ClassNotFoundException e) {
if (logger.isErrorEnabled()) {
logger.error(e.getMessage(), e);
}
}
JobBuilder jb = JobBuilder.newJob(forName).withIdentity(jobName, jobGroup)
.withDescription(jobDescription);
//放入数据
JobDataMap data = new JobDataMap();
data.put("count", 5);
jb = jb.usingJobData(data);
JobDetail jobDetail = jb.build();
TriggerBuilder<Trigger> triggerBuilder = TriggerBuilder.newTrigger()
.withIdentity(triggerName, triggerGroup).forJob(jobDetail);
ScheduleBuilder sche = CronScheduleBuilder.cronSchedule(cron);
Trigger trigger = triggerBuilder.withSchedule(sche).build();
Scheduler quartzScheduler = (Scheduler) context.getBean("quartzScheduler");
quartzScheduler.scheduleJob(jobDetail, trigger);
}
}
这里使用的job类仍然是上文中的TestJob，只是改用程序生成调用而已，另外这里的count我手动赋值成了5
到此，quartz持久化配置就完成了。
三、web管理
quartz是通过jobName和jobGroup来区分job的，所以job的name和group一定要填写，然后通过这个就可以对job进行添加、暂停、恢复和删除了。
添加：quartzScheduler.scheduleJob(jobDetail,trigger)
暂停：quartzScheduler.pauseJob(JobKey.jobKey(jobName,jobGroup))
恢复：quartzSchduler.resumeJob(JobKey.jobKey(jobName,jobGroup))
删除：quartScheduler.deleteJob(JobKey.jobKey(jobName,jobGroup))
可以根据需要定制相关的UI界面，通过界面去触发job操作
