Spring Boot 应用系列 6 -- Spring Boot 2 整合Quartz

2022-09-22 12:38:41

Quartz是实现定时任务的利器，Quartz主要有四个组成部分，分别是：

1. Job(任务)：包含具体的任务逻辑；

2. JobDetail(任务详情）：是对Job的一种详情描述；

3. Trigger(触发器)：负责管理触发JobDetail的机制；

4. Scheduler(调度器)：负责Job的执行。

有两种方式可以实现Spring Boot与Quartz的整合：

一、使用Spring提供的工厂类

spring-context.jar和spring-context-support.jar类库提供了一些org.quartz包的扩展，使用这些扩展并通过注入bean的方式可以实现与Quartz的整合。

1. pom.xml

添加依赖：

<dependency>

    <groupId>org.quartz-scheduler</groupId>

    <artifactId>quartz</artifactId>

</dependency>

<dependency>

    <groupId>org.springframework</groupId>

    <artifactId>spring-tx</artifactId>

</dependency>

<dependency>

    <groupId>org.springframework</groupId>

    <artifactId>spring-context-support</artifactId>

</dependency>

Spring Boot 2.0.4指定的quartz.jar的版本是2.3.0

2. Job

首先定义一个执行器：

import java.util.Date;

import devutility.internal.text.format.DateFormatUtils;

public class Executor {

    public static void execute(String name, long costMillis) {

        Date startDate = new Date();

        Date endDate = new Date(startDate.getTime() + costMillis);

        StringBuilder stringBuilder = new StringBuilder();

        stringBuilder.append(String.format("[%s]: ", DateFormatUtils.format(startDate, "yyyy-MM-dd HH:mm:ss:SSS")));

        stringBuilder.append(String.format("%s executing on %s, ", name, Thread.currentThread().getName()));

        stringBuilder.append(String.format("will finish at %s.", DateFormatUtils.format(endDate, "yyyy-MM-dd HH:mm:ss:SSS")));

        System.out.println(stringBuilder.toString());

        try {

            Thread.sleep(costMillis);

        } catch (InterruptedException e) {

            e.printStackTrace();

        }

    }

}

定义Job：

import org.springframework.scheduling.annotation.EnableScheduling;

import org.springframework.stereotype.Component;

import devutility.test.app.quartz.common.Executor;

@Component

@EnableScheduling

public class SpringJobs {

    public void job1() {

        Executor.execute("Job1", 5000);

    }

    public void job2() {

        Executor.execute("Job2", 6000);

    }

}

@EnableScheduling注解是必须要有的。

3. JobDetail

 package devutility.test.app.quartz.config;

 import org.quartz.Trigger;

 import org.springframework.beans.factory.annotation.Qualifier;

 import org.springframework.context.annotation.Bean;

 import org.springframework.context.annotation.Configuration;

 import org.springframework.scheduling.quartz.CronTriggerFactoryBean;

 import org.springframework.scheduling.quartz.MethodInvokingJobDetailFactoryBean;

 import org.springframework.scheduling.quartz.SchedulerFactoryBean;

 import devutility.test.app.quartz.jobs.SpringJobs;

 @Configuration

 public class SpringJobConfiguration {

     @Bean

     public MethodInvokingJobDetailFactoryBean jobDetailFactory1(SpringJobs springJobs) {

         MethodInvokingJobDetailFactoryBean jobDetailFactory = new MethodInvokingJobDetailFactoryBean();

         jobDetailFactory.setName("Spring-Job1");

         jobDetailFactory.setGroup("Spring-Group");

         jobDetailFactory.setTargetObject(springJobs);

         jobDetailFactory.setTargetMethod("job1");

         jobDetailFactory.setConcurrent(false);

         return jobDetailFactory;

     }

     @Bean

     public MethodInvokingJobDetailFactoryBean jobDetailFactory2(SpringJobs springJobs) {

         MethodInvokingJobDetailFactoryBean jobDetailFactory = new MethodInvokingJobDetailFactoryBean();

         jobDetailFactory.setName("Spring-Job2");

         jobDetailFactory.setGroup("Spring-Group");

         jobDetailFactory.setTargetObject(springJobs);

         jobDetailFactory.setTargetMethod("job2");

         jobDetailFactory.setConcurrent(true);

         return jobDetailFactory;

     }

MethodInvokingJobDetailFactoryBean是来自spring-context-support.jar的一个工厂类，它实现了FactoryBean<JobDetail>接口，完成了对具体job的封装。

21行指定具体的任务是我们在2中定义的SpringJobs，22行指定我们使用SpringJobs中的job1方法作为定时任务的具体业务实现。

注意24和37行，如果一个任务每隔5秒执行一次，但是每次需要执行10秒，那么它有两种执行方式，串行(执行完上一个再开启下一个)和并行(间隔时间到了就开始并行执行下一个)，24和37就分别设置成了并行和串行执行。

4. Trigger

Quartz支持多种trigger，其中配置最为灵活的一种当属CronTrigger，它属于表达式类型的配置方式，有关cron表达式的配置请参考

 package devutility.test.app.quartz.config;

 import org.quartz.Trigger;

 import org.springframework.beans.factory.annotation.Qualifier;

 import org.springframework.context.annotation.Bean;

 import org.springframework.context.annotation.Configuration;

 import org.springframework.scheduling.quartz.CronTriggerFactoryBean;

 import org.springframework.scheduling.quartz.MethodInvokingJobDetailFactoryBean;

 import org.springframework.scheduling.quartz.SchedulerFactoryBean;

 import devutility.test.app.quartz.jobs.SpringJobs;

 @Configuration

 public class SpringJobConfiguration {

     @Bean

     public CronTriggerFactoryBean cronTriggerFactory1(@Qualifier("jobDetailFactory1") MethodInvokingJobDetailFactoryBean jobDetailFactory1) {

         CronTriggerFactoryBean cronTriggerFactoryBean = new CronTriggerFactoryBean();

         cronTriggerFactoryBean.setName("cronTriggerFactoryForJobDetailFactory1");

         cronTriggerFactoryBean.setJobDetail(jobDetailFactory1.getObject());

         cronTriggerFactoryBean.setCronExpression("0/3 * * * * ?");

         return cronTriggerFactoryBean;

     }

     @Bean

     public CronTriggerFactoryBean cronTriggerFactory2(@Qualifier("jobDetailFactory2") MethodInvokingJobDetailFactoryBean jobDetailFactory2) {

         CronTriggerFactoryBean cronTriggerFactoryBean = new CronTriggerFactoryBean();

         cronTriggerFactoryBean.setName("cronTriggerFactoryForJobDetailFactory2");

         cronTriggerFactoryBean.setJobDetail(jobDetailFactory2.getObject());

         cronTriggerFactoryBean.setCronExpression("0/4 * * * * ?");

         return cronTriggerFactoryBean;

     }

本例使用的Cron表达式非常简单，分别是每隔3秒和每隔4秒执行一次。

5. Scheduler

     @Bean

     public SchedulerFactoryBean schedulerFactory1(Trigger cronTriggerFactory1, Trigger cronTriggerFactory2) {

         SchedulerFactoryBean schedulerFactoryBean = new SchedulerFactoryBean();

         schedulerFactoryBean.setStartupDelay(2);

         schedulerFactoryBean.setTriggers(cronTriggerFactory1, cronTriggerFactory2);

         return schedulerFactoryBean;

     }

行4指明系统启动之后需要延迟2秒执行；

行5用于注册需要执行的触发器；

SchedulerFactoryBean还有一个叫autoStartup的属性，用于指明任务在系统启动之后是否立即执行，默认是true。

6. 测试

由此可见，Job1是按照我们的预期串行执行的。

Job2则是并行执行的。

二、使用org.quartz原生类和方法

1. pom.xml，只需要添加quartz

<dependency>

    <groupId>org.quartz-scheduler</groupId>

    <artifactId>quartz</artifactId>

</dependency>

2. 定义一个Scheduler类型的Bean

package devutility.test.app.quartz.config;

import org.quartz.Scheduler;

import org.quartz.SchedulerException;

import org.quartz.impl.StdSchedulerFactory;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

@Configuration

public class QuartzConfiguration {

    @Bean

    public Scheduler scheduler() throws SchedulerException {

        return StdSchedulerFactory.getDefaultScheduler();

    }

}

3. Job

package devutility.test.app.quartz.jobs;

import org.quartz.DisallowConcurrentExecution;

import org.quartz.Job;

import org.quartz.JobDetail;

import org.quartz.JobExecutionContext;

import org.quartz.JobExecutionException;

import devutility.test.app.quartz.common.Executor;

@DisallowConcurrentExecution

public class ScheduleJob1 implements Job {

    @Override

    public void execute(JobExecutionContext context) throws JobExecutionException {

        Executor.execute("ScheduleJob1", 5000);

        JobDetail jobDetail = context.getJobDetail();

        System.out.println(String.format("key: %s", jobDetail.getKey()));

    }

}

这种方式每一个Job都需要定义一个类实现org.quartz.Job接口，形式上要比第一种方式更条理。

3. 我们定义一个创建任意Job的公共方法，来实现Job类的定时执行：

 package devutility.test.app.quartz.services;

 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.TriggerBuilder;

 import org.quartz.TriggerKey;

 import org.springframework.beans.factory.annotation.Autowired;

 import org.springframework.stereotype.Service;

 import devutility.internal.models.OperationResult;

 @Service

 public class JobServiceImpl implements JobService {

     @Autowired

     private Scheduler schedulerFactory1;

     @Autowired

     private Scheduler scheduler;

     @Override

     public void start(String name, String group, String cronExpression, Class<? extends Job> clazz) throws SchedulerException {

         JobDetail jobDetail = JobBuilder.newJob(clazz).withIdentity(name, group).build();

         String triggerName = String.format("trigger_%s", name);

         CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(cronExpression);

         CronTrigger trigger = TriggerBuilder.newTrigger().withIdentity(triggerName, group).withSchedule(scheduleBuilder).build();

         scheduler.scheduleJob(jobDetail, trigger);

         if (!scheduler.isStarted()) {

             scheduler.start();

         }

     }

28行创建一个新的JobDetail，并将Job实现类的Class对象赋值给它；

32行创建了一个新的Trigger；

33-37行使用注入进来的scheduler来运行一个定时任务。

三、对Job的CRUD操作

创建和运行Job上面已经讲过了，下面说一下Job的暂停、中断、删除和更新操作。

1. 暂停

scheduler有一个方法public void pauseJob(JobKey jobKey)，该方法通过暂停trigger的触发来实现暂停job的功能，调用该方法之后正在运行的job会持续运行到业务逻辑处理完毕，等下一个触发条件满足时不再开启新的job。

    public OperationResult pause(String group, String name) {

        OperationResult result = new OperationResult();

        JobKey jobKey = JobKey.jobKey(name, group);

        try {

            scheduler.pauseJob(jobKey);

        } catch (SchedulerException e) {

            e.printStackTrace();

            result.setErrorMessage(String.format("Pause Job with name = \"%s\" group = \"%s\" failed, system error!", name, group));

        }

        return result;

    }

2. 中断

针对需要中断的Job，quartz专门为其定义了一个接口org.quartz.InterruptableJob：

public interface InterruptableJob extends Job {

    /*

     * ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

     *

     * Interface.

     *

     * ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

     */

    /**

     * <p>

     * Called by the <code>{@link Scheduler}</code> when a user

     * interrupts the <code>Job</code>.

     * </p>

     *

     * @throws UnableToInterruptJobException

     *           if there is an exception while interrupting the job.

     */

    void interrupt()

        throws UnableToInterruptJobException;

}

所有需要支持中断的Job都需要实现这个接口：

package devutility.test.app.quartz.jobs;

import org.quartz.DisallowConcurrentExecution;

import org.quartz.InterruptableJob;

import org.quartz.JobExecutionContext;

import org.quartz.JobExecutionException;

import org.quartz.UnableToInterruptJobException;

import devutility.test.app.quartz.common.Executor;

@DisallowConcurrentExecution

public class ScheduleJob2 implements InterruptableJob {

    @Override

    public void execute(JobExecutionContext context) throws JobExecutionException {

        Executor.execute(context, 5000);

    }

    @Override

    public void interrupt() throws UnableToInterruptJobException {

        System.out.println("ScheduleJob2 is interrupting now。。。");

    }

}

然后通过调用Scheduler实例的boolean interrupt(JobKey jobKey)方法，查看StdScheduler的源码，我们发现Scheduler的interrupt方法仅仅是调用了InterruptableJob中的interrupt()方法实现，然后设置了一下自己的interrupted属性就完了，并未做任何其他操作。

所以，如果要实现可以中断的Job，我们需要在InterruptableJob实现类中增加中断的逻辑：

package devutility.test.app.quartz.jobs;

import org.quartz.InterruptableJob;

import org.quartz.JobExecutionContext;

import org.quartz.JobExecutionException;

import org.quartz.UnableToInterruptJobException;

import devutility.test.app.quartz.common.Executor;

public class ScheduleJob3 implements InterruptableJob {

    private boolean interrupted = false;

    @Override

    public void execute(JobExecutionContext context) throws JobExecutionException {

        Executor.execute("ScheduleJob3 sub-task 1", 1000);

        if (interrupted) {

            return;

        }

        Executor.execute("ScheduleJob3 sub-task 2", 1000);

        if (interrupted) {

            return;

        }

        Executor.execute("ScheduleJob3 sub-task 3", 1000);

    }

    @Override

    public void interrupt() throws UnableToInterruptJobException {

        interrupted = true;

    }

}

3. 删除

    public OperationResult delete(String jobName, String jobGroup) {

        OperationResult result = new OperationResult();

        result.append(String.format("Removing Quartz job: %s", jobName));

        try {

            if (!schedulerFactory1.deleteJob(JobKey.jobKey(jobName, jobGroup))) {

                result.setErrorMessage(String.format("Removing job %s failed!", jobName));

            }

        } catch (SchedulerException e) {

            e.printStackTrace();

            result.setErrorMessage(String.format("Removing job %s failed with error!", jobName));

        }

        return result;

    }

4. 更新

    public OperationResult update(CronTrigger cronTrigger, String cronExpression) {

        OperationResult result = new OperationResult();

        TriggerKey triggerKey = cronTrigger.getKey();

        CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(cronExpression);

        CronTrigger newTrigger = cronTrigger.getTriggerBuilder().withSchedule(scheduleBuilder).build();

        try {

            schedulerFactory1.rescheduleJob(triggerKey, newTrigger);

            result.append(String.format("Update cron trigger %s succeeded!", triggerKey.getName()));

        } catch (SchedulerException e) {

            e.printStackTrace();

            result.setErrorMessage(String.format("Update cron trigger %s failed!", triggerKey.getName()));

        }

        return result;

    }

Demo代码

码农公寓

相关文章