Springboot2.X集成Quartz集群

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

为什么要使用Quzrtz集群

在项目进行集群部署时，如果业务在执行中存在互斥关系，没有对定时任务进行统一管理，就会引起业务的多次执行，不能满足业务要求。这时就需要对任务进行管理，要保证一笔业务在所有的集群环境中，有且只有一台机器能执行该任务。

如果不适用Quartz集群，要如何实现这种业务逻辑？

在这里只列出两种简单的思路：

1. 利用单线程机制。可以在redis中设置一个属性为空，每次任务执行时去设置这个全局变量，进入任务中需要对值进行校验，值不为空则跳过本次执行任务，值为空时进行设置，方法执行完毕后再将该值置空。
2. 指定该定时任务在某台固定机器IP上执行，其他Ip则跳过该任务。

这两种方法都有其弊端，第一种是对代码侵入较大，第二种则是无法做到高可用。

准备工作

首先在http://www.quartz-scheduler.org/downloads/ 上下载quzrtz包，本文以quartz-2.3.0-distribution.tar.gz 版本为例。下载后解压，选择你需要的数据库进行表创建，本文以oracle为例，顾使用tables_oracle.sql即可。tips:可以直接在解压后的文件夹里搜索table，就能找到支持的所有类型的数据库脚本。

核心代码实现

quartz.properties 基本配置属性

###############内存版配置####################################
#org.quartz.scheduler.instanceName=spring-boot-quartz-demo
#org.quartz.scheduler.instanceId=AUTO
#org.quartz.threadPool.threadCount=5
#org.quartz.jobStore.class = org.quartz.simpl.RAMJobStore
###############集群版配置####################################
org.quartz.scheduler.instanceName=spring-boot-quartz-demo
org.quartz.scheduler.instanceId=AUTO
org.quartz.scheduler.skipUpdateCheck=true
org.quartz.threadPool.class=org.quartz.simpl.SimpleThreadPool
org.quartz.threadPool.threadCount=5
org.quartz.threadPool.threadPriority=5
org.quartz.jobStore.misfireThreshold=60000
org.quartz.jobStore.class=org.quartz.impl.jdbcjobstore.JobStoreTX
#org.quartz.jobStore.dataSource=dataSource
org.quartz.jobStore.tablePrefix=QRTZ_
org.quartz.jobStore.isClustered=true
org.quartz.jobStore.driverDelegateClass=org.quartz.impl.jdbcjobstore.oracle.OracleDelegate
org.quartz.jobStore.useProperties=false

QuartzJobFactory.java 注入到Spring 容器

package top.enjoyitlife.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.SpringBeanJobFactory;
import org.springframework.stereotype.Component;@Component
public class QuartzJobFactory extends SpringBeanJobFactory{
@Autowired
private AutowireCapableBeanFactory capableBeanFactory;
@Override
protected Object createJobInstance(TriggerFiredBundle bundle) throws Exception {
Object jobInstance = super.createJobInstance(bundle);
capableBeanFactory.autowireBean(jobInstance);
return jobInstance;
}
}

QuartzConfig.java 配置类加载配置文件对quartz进行实例化和属性设置注意使用集群方案需要连接数据库，而内存版则不用配置数据库

package top.enjoyitlife.config;import java.io.IOException;
import java.util.Properties;
import javax.sql.DataSource;
import org.quartz.spi.JobFactory;
import org.springframework.beans.factory.config.PropertiesFactoryBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.scheduling.quartz.SchedulerFactoryBean;
import org.springframework.transaction.PlatformTransactionManager;
@Configuration
public class QuartzConfig { public static final String QUARTZ_PROPERTIES_PATH = "/quartz.properties"; @Bean
public JobFactory jobFactory(ApplicationContext applicationContext) {
QuartzJobFactory jobFactory = new QuartzJobFactory();
jobFactory.setApplicationContext(applicationContext);
return jobFactory;
} @Bean
public SchedulerFactoryBean schedulerFactoryBean(JobFactory jobFactory,DataSource dataSource, PlatformTransactionManager transactionManager) throws IOException {
SchedulerFactoryBean factory = new SchedulerFactoryBean();
factory.setAutoStartup(true);
factory.setJobFactory(jobFactory);
factory.setQuartzProperties(quartzProperties());
//集群版配置
factory.setDataSource(dataSource);
factory.setTransactionManager(transactionManager);
return factory;
} @Bean
public Properties quartzProperties() throws IOException {
PropertiesFactoryBean propertiesFactoryBean = new PropertiesFactoryBean();
propertiesFactoryBean.setLocation(new ClassPathResource(QUARTZ_PROPERTIES_PATH));
propertiesFactoryBean.afterPropertiesSet();
return propertiesFactoryBean.getObject();
}}

QuartzBindOperationConfig.java 定时任务的绑定包括触发器 CRON表达式备注任务名任务所在组

package top.enjoyitlife.config;import org.quartz.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import top.enjoyitlife.schedule.TestSchedule;@Configuration
public class QuartzBindOperationConfig {
private Logger logger = LoggerFactory.getLogger(getClass());
@Autowired
private QuartzJobFactory quartzJobFactory;
@Autowired
private Scheduler scheduler;
public void scheduleBind() {
try {
scheduler.setJobFactory(quartzJobFactory);
JobDetail tesJobDetail = JobBuilder.newJob(TestSchedule.class)
.withIdentity("tesJob", "tesJob").withDescription("定时任务demo")
.build();
CronScheduleBuilder tesJobCronScheduleBuilder = CronScheduleBuilder.cronSchedule("0/10 * * * * ?");
CronTrigger tesJobCronTrigger = TriggerBuilder.newTrigger()
.withIdentity("tesJobTrigger", "tesJob")
.withSchedule(tesJobCronScheduleBuilder).build();
scheduler.scheduleJob(tesJobDetail, tesJobCronTrigger);
scheduler.start();
} catch (SchedulerException e) {
// e.printStackTrace();
} }}

ApplicationListenerConfig.java 项目启动初始化即加载任务

package top.enjoyitlife.config;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.event.ContextRefreshedEvent;@Configuration
public class ApplicationListenerConfig implements ApplicationListener<ContextRefreshedEvent> { private Logger logger = LoggerFactory.getLogger(getClass());
@Autowired
private QuartzBindOperationConfig quartzBindOperationConfig;
@Override
public void onApplicationEvent(ContextRefreshedEvent event) {
quartzBindOperationConfig.scheduleBind();
}}

TestSchedule.java 测试任务演示类

package top.enjoyitlife.schedule;
import java.util.Date;
import org.quartz.DisallowConcurrentExecution;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;@DisallowConcurrentExecution
public class TestSchedule implements Job{ private Logger logger = LoggerFactory.getLogger(getClass()); @Override
public void execute(JobExecutionContext context) throws JobExecutionException {
logger.info("*****"+new Date());
}
}

POM.xml 引用如下

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.1.4.RELEASE</version>
<relativePath/> 
</parent>
<groupId>top.enjoyitlife</groupId>
<artifactId>enjoyitlife</artifactId>
<version>0.0.1-SNAPSHOT</version>
<packaging>war</packaging>
<name>quartzDistribute</name>
<description>quartz集群demo</description> <properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
<java.version>1.8</java.version>
</properties> <repositories>
<repository>
<id>maven-ali</id>
<url>http://maven.aliyun.com/nexus/content/groups/public//</url>
<releases>
<enabled>true</enabled>
</releases>
<snapshots>
<enabled>true</enabled>
<updatePolicy>always</updatePolicy>
<checksumPolicy>fail</checksumPolicy>
</snapshots>
</repository>
</repositories> <dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency> <dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-tomcat</artifactId>
<scope>provided</scope>
</dependency> <dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-quartz</artifactId>
</dependency> <dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context-support</artifactId>
</dependency> <dependency>
<groupId>org.codehaus.janino</groupId>
<artifactId>janino</artifactId>
</dependency> 
<dependency>
<groupId>com.oracle</groupId>
<artifactId>ojdbc6</artifactId>
<version>11.2.0.4.0</version>
</dependency> <dependency>
<groupId>com.alibaba</groupId>
<artifactId>druid-spring-boot-starter</artifactId>
<version>1.1.10</version>
</dependency> <dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-jpa</artifactId>
</dependency> <dependency>
<groupId>org.mybatis.generator</groupId>
<artifactId>mybatis-generator-core</artifactId>
<version>1.3.5</version>
</dependency> <dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency> </dependencies> <build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build></project>

运行日志：

2019-05-11 16:11:20.076 INFO 9624 --- [Quartz_Worker-1] top.enjoyitlife.schedule.TestSchedule : *****Sat May 11 16:11:20 CST 2019
2019-05-11 16:11:30.037 INFO 9624 --- [Quartz_Worker-2] top.enjoyitlife.schedule.TestSchedule : *****Sat May 11 16:11:30 CST 2019
2019-05-11 16:11:40.033 INFO 9624 --- [Quartz_Worker-3] top.enjoyitlife.schedule.TestSchedule : *****Sat May 11 16:11:40 CST 2019

任务执行后数据库中就可以看到我对应的任务记录

集群效果验证

修改application.properties 中的server.port端口号，然后重新运行QuartzDistributeApplication的main 方法。以下截图为双服务同时启动时的任务，目前是只有任务1有日志记录，服务2没有任务记录。