Java定时任务(二)：quartz调度框架

作者: sawyerlsy | 来源:发表于2018-06-13 14:44 被阅读0次

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操作