在大多数的情况下，我们都希望任务能按照我们预期的时间进行执行。我最常接触的是spring自带的@Scheduled注解或是使用XXL-JOB完成对采数逻辑的调度。@Scheduled注解无法直接完成分布式的任务调度，需要配合关系型数据库的行级别锁互斥，这无疑增大了设计的复杂度。使用XXL-JOB则无疑增大了data-pipeline对第三方应用的依赖。基于以上的考虑，我选择了直接集成quartz，quartz提供了基于数据库行级别锁互斥的分布式调度方案，下面罗列集成的步骤：
- pom添加依赖：
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-quartz</artifactId>
</dependency>
- application-*.yml添加如下配置（可根据注释按照实际情况调整）：
quartz:
jobStore:
# 数据保存方式为数据库持久化
class: org.quartz.impl.jdbcjobstore.JobStoreTX
# JobDataMaps是否都为String类型，默认false
useProperties: false
# 表的前缀，默认QRTZ_
tablePrefix: QRTZ_
# 是否加入集群
isClustered: true
# 调度实例失效的检查时间间隔 ms
clusterCheckinInterval: 5000
# 数据库代理类
driverDelegateClass: org.quartz.impl.jdbcjobstore.StdJDBCDelegate
# 当设置为“true”时，此属性告诉Quartz 在非托管JDBC连接上调用setTransactionIsolation
txIsolationLevelReadCommitted: true
scheduler:
# 调度标识名 集群中每一个实例都必须使用相同的名称
instanceName: ClusterQuartz
# ID设置为自动获取 每一个必须不同
instanceId: AUTO
threadPool:
# 线程池的实现类（一般使用SimpleThreadPool即可满足几乎所有用户的需求）
class: org.quartz.simpl.SimpleThreadPool
# 指定线程数，一般设置为1-100直接的整数，根据系统资源配置
threadCount: 5
# 设置线程的优先级(可以是Thread.MIN_PRIORITY（即1）和Thread.MAX_PRIORITY（这是10）之间的任何int
threadPriority: 5
- 创建类SpringJobFactory.java，解决job中service注入为空的问题
package cn.juque.datapipeline.quartz;
import org.quartz.spi.TriggerFiredBundle;
import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
import org.springframework.scheduling.quartz.AdaptableJobFactory;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
/**
* @author juque
* @version 1.0.0
* <ul>
* <li>SpringJobFactory</li>
* <li>解决job中service注入为空的问题。</li>
* </ul>
* @date 2023-04-08 21:58:32
**/
@Component
public class SpringJobFactory extends AdaptableJobFactory {
@Resource
private AutowireCapableBeanFactory autowireCapableBeanFactory;
@Override
protected Object createJobInstance(TriggerFiredBundle bundle) throws Exception {
Object jobInstance = super.createJobInstance(bundle);
// 进行注入
this.autowireCapableBeanFactory.autowireBean(jobInstance);
return jobInstance;
}
}
- 创建配置类QuartzConfig.java
package cn.juque.datapipeline.config;
import cn.juque.datapipeline.quartz.SpringJobFactory;
import org.quartz.Scheduler;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.quartz.SchedulerFactoryBean;
import javax.annotation.Resource;
import javax.sql.DataSource;
import java.util.Properties;
/**
* @author juque
* @version 1.0.0
* <ul>
* <li>QuartzConfig</li>
* </ul>
* @date 2023-04-07 15:05:33
**/
@Configuration
public class QuartzConfig {
@Value("${quartz.jobStore.useProperties}")
private Boolean jobStoreUseProperties;
@Value("${quartz.jobStore.tablePrefix}")
private String jobStoreTablePrefix;
@Value("${quartz.jobStore.isClustered}")
private Boolean jobStoreIsClustered;
@Value("${quartz.jobStore.clusterCheckinInterval}")
private String jobStoreClusterCheckinInterval;
@Value("${quartz.jobStore.txIsolationLevelReadCommitted}")
private Boolean jobStoreTxIsolationLevelReadCommitted;
@Value("${quartz.jobStore.class}")
private String jobStoreClass;
@Value("${quartz.jobStore.driverDelegateClass}")
private String jobStoreDriverDelegateClass;
@Value("${quartz.scheduler.instanceName}")
private String schedulerInstanceName;
@Value("${quartz.scheduler.instanceId}")
private String schedulerInstanceId;
@Value("${quartz.threadPool.class}")
private String threadPoolClass;
@Value("${quartz.threadPool.threadCount}")
private String threadPoolThreadCount;
@Value("${quartz.threadPool.threadPriority}")
private String threadPoolThreadPriority;
@Resource
private SpringJobFactory springJobFactory;
@Bean
public SchedulerFactoryBean schedulerFactoryBean(DataSource dataSource) {
SchedulerFactoryBean schedulerFactoryBean = new SchedulerFactoryBean();
schedulerFactoryBean.setDataSource(dataSource);
Properties properties = new Properties();
properties.put("org.quartz.jobStore.useProperties", jobStoreUseProperties);
properties.put("org.quartz.jobStore.tablePrefix", jobStoreTablePrefix);
properties.put("org.quartz.jobStore.isClustered", jobStoreIsClustered);
properties.put("org.quartz.jobStore.clusterCheckinInterval", jobStoreClusterCheckinInterval);
properties.put("org.quartz.jobStore.txIsolationLevelReadCommitted", jobStoreTxIsolationLevelReadCommitted);
properties.put("org.quartz.jobStore.class", jobStoreClass);
properties.put("org.quartz.jobStore.driverDelegateClass", jobStoreDriverDelegateClass);
properties.put("org.quartz.scheduler.instanceName", schedulerInstanceName);
properties.put("org.quartz.scheduler.instanceId", schedulerInstanceId);
properties.put("org.quartz.threadPool.class", threadPoolClass);
properties.put("org.quartz.threadPool.threadCount", threadPoolThreadCount);
properties.put("org.quartz.threadPool.threadPriority", threadPoolThreadPriority);
schedulerFactoryBean.setQuartzProperties(properties);
schedulerFactoryBean.setSchedulerName("dpp-scheduler");
schedulerFactoryBean.setStartupDelay(1);
schedulerFactoryBean.setApplicationContextSchedulerContextKey("applicationContextKey");
// QuartzScheduler 启动时更新己存在的Job，这样就不用每次修改targetObject后删除qrtz_job_details表对应记录
schedulerFactoryBean.setOverwriteExistingJobs(true);
schedulerFactoryBean.setAutoStartup(true);
schedulerFactoryBean.setJobFactory(this.springJobFactory);
return schedulerFactoryBean;
}
@Bean
public Scheduler scheduler(DataSource dataSource) {
return this.schedulerFactoryBean(dataSource).getScheduler();
}
}
- 至此，基础配置已完成。
- 下面参考任务调度的实现，创建TaskGroupJobServiceImpl.java
package cn.juque.datapipeline.service.impl;
import cn.juque.datapipeline.constans.BusinessConstants;
import cn.juque.datapipeline.helper.TaskInfoHelper;
import org.quartz.Job;
import org.quartz.JobDataMap;
import org.quartz.JobExecutionContext;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
/**
* @author juque
* @version 1.0.0
* <ul>
* <li>TaskGroupJobjServiceImpl</li>
* </ul>
* @date 2023-04-07 15:12:39
**/
@Service("taskGroupJobService")
public class TaskGroupJobServiceImpl implements Job {
@Resource
private TaskInfoHelper taskInfoHelper;
@Override
public void execute(JobExecutionContext jobExecutionContext) {
JobDataMap jobDataMap = jobExecutionContext.getJobDetail().getJobDataMap();
String groupId = jobDataMap.getString(BusinessConstants.GROUP_ID_KEY);
taskInfoHelper.startTask(groupId);
}
}
- 添加任务的逻辑，具体参考：QuartzHelper.java
/**
* 添加任务调度
*
* @param groupInfo 任务组信息
*/
public void addJob(TaskGroupInfo groupInfo) {
JobDetail jobDetail = JobBuilder
.newJob(TaskGroupJobServiceImpl.class)
.withDescription(groupInfo.getGroupName())
.withIdentity(QUARTZ_JOB_NAME + groupInfo.getId(), QUARTZ_JOB_GROUP_NAME)
// 传递任务组ID
.usingJobData(BusinessConstants.GROUP_ID_KEY, groupInfo.getId()).build();
Trigger trigger;
if (GroupExecuteTypeEnum.CORN.getCode().equals(groupInfo.getExecuteType())) {
trigger = TriggerBuilder.newTrigger()
.withDescription(groupInfo.getGroupName())
.withIdentity(QUARTZ_TRIGGER_NAME + groupInfo.getId(), QUARTZ_TRIGGER_GROUP_NAME)
.startNow().withSchedule(CronScheduleBuilder.cronSchedule(groupInfo.getCron())).build();
} else {
trigger = TriggerBuilder.newTrigger()
.withDescription(groupInfo.getGroupName())
.withIdentity(QUARTZ_TRIGGER_NAME + groupInfo.getId(), QUARTZ_TRIGGER_GROUP_NAME)
.startAt(DateUtil.offsetSecond(new Date(), groupInfo.getDelaySeconds()))
.withSchedule(SimpleScheduleBuilder.repeatSecondlyForever(groupInfo.getDelaySeconds())).build();
}
try {
this.scheduler.scheduleJob(jobDetail, trigger);
log.info("完成任务组【{}】的调度初始化", groupInfo.getGroupName());
} catch (Exception e) {
log.error("任务组【{}】调度任务初始化失败", groupInfo.getGroupName(), e);
throw new AppException(MessageEnum.SYSTEM_ERROR);
}
}
- 至此完成对quartz的集成，并实现quartz对任务组的分布式调度。下面附加对Job的其他操作方法：
/**
* 删除任务
*
* @param groupId 任务组ID
*/
public void deleteJob(String groupId) {
JobKey jobKey = JobKey.jobKey(QUARTZ_JOB_NAME + groupId, QUARTZ_JOB_GROUP_NAME);
try {
this.scheduler.deleteJob(jobKey);
} catch (Exception e) {
log.error("任务ID【{}】调度任务删除失败", groupId, e);
throw new AppException(MessageEnum.SYSTEM_ERROR);
}
}
/**
* job是否存在
*
* @param groupId 任务组ID
* @return boolean
*/
public Boolean existsJob(String groupId) {
JobKey jobKey = JobKey.jobKey(QUARTZ_JOB_NAME + groupId, QUARTZ_JOB_GROUP_NAME);
try {
return this.scheduler.checkExists(jobKey);
} catch (Exception e) {
log.error("任务ID【{}】调度任务删除失败", groupId, e);
throw new AppException(MessageEnum.SYSTEM_ERROR);
}
}
/**
* 调度一次
*
* @param groupId 任务组ID
*/
public void runOnce(String groupId) {
JobKey jobKey = JobKey.jobKey(QUARTZ_JOB_NAME + groupId, QUARTZ_JOB_GROUP_NAME);
try {
this.scheduler.triggerJob(jobKey);
} catch (Exception e) {
log.error(e.getMessage(), e);
throw new AppException(MessageEnum.SYSTEM_ERROR);
}
}
/**
* 暂停任务
*
* @param groupId 任务组id
*/
public void pauseJob(String groupId) {
JobKey jobKey = JobKey.jobKey(QUARTZ_JOB_NAME + groupId, QUARTZ_JOB_GROUP_NAME);
try {
this.scheduler.pauseJob(jobKey);
} catch (Exception e) {
log.error(e.getMessage(), e);
throw new AppException(MessageEnum.SYSTEM_ERROR);
}
}
/**
* 恢复任务
*
* @param groupId 任务组ID
*/
public void resumeJob(String groupId) {
JobKey jobKey = JobKey.jobKey(QUARTZ_JOB_NAME + groupId, QUARTZ_JOB_GROUP_NAME);
try {
this.scheduler.resumeJob(jobKey);
} catch (Exception e) {
log.error(e.getMessage(), e);
throw new AppException(MessageEnum.SYSTEM_ERROR);
}
}
