Quartz實現分散式叢集schedule排程(定時任務執行)
專案需求:
需要執行一些定時任務,比如每間隔幾小時同步其他系統中資訊,或者定期監控一些服務是否正常。如果定時任務程式有多個例項在執行,就是執行多次,因此需要叢集分散式的schedule。保證定時任務有且只有一個執行。
調研各種schedule框架後,選擇quartz,原因如下:1, 專案已經在使用quartz,大家比較熟悉,2,quartz提供的clustering功能完備,滿足專案的需求。
說明,
本文主要參考quartz的官方文件,從http://www.quartz-scheduler.org/downloads下載, 解壓後參考example13 - Clustering Quartz and JDBC Job Stores。 我下載的是quartz-2.2.3-distribution.tar.gz
具體步驟,
我們使用springboot程式,完整程式碼在這裡, 歡迎加星,fork。 我使用了本機的資料庫,使用者名稱和密碼也配置了,執行時需要先建立資料庫,建立表,然後修改application.properties中的配置資訊。
因此具體步驟如下
-
初始資料庫
初始指令碼在 quartz-2.2.3\docs\dbTables下面。我用的是mysql資料innodb引擎,因此執行tables_mysql_innodb.sql指令碼。
-
引入依賴包
這裡主要列出quartz的依賴包,springboot的可以參考完整的程式碼
<quartz.version >2.3.0</quartz.version>
...
<dependency>
<groupId>org.quartz-scheduler</groupId>
<artifactId>quartz</artifactId>
<version>${quartz.version}</version>
</dependency>
<dependency>
< groupId>org.quartz-scheduler</groupId>
<artifactId>quartz-jobs</artifactId>
<version>${quartz.version}</version>
</dependency>
- 建立資料來源配置
package com.yq.config;
import javax.sql.DataSource;
import com.alibaba.druid.pool.DruidDataSource;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class DataSourceConfig {
@Value("${spring.datasource.url:jdbc:mysql://127.0.0.1:3306/myscheduler?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&useSSL=false}")
private String url;
@Value("${spring.datasource.driver-class-name:com.mysql.jdbc.Driver}")
private String driverClassName;
@Value("${spring.datasource.username:user1}")
private String userName;
@Value("${spring.datasource.password:password}")
private String password;
@Value("${spring.datasource.initialSize:6}")
private int initialSize;
@Value("${spring.datasource.minIdle:6}")
private int minIdle;
@Value("${pring.datasource.maxActive:24}")
private int maxActive;
@Value("${spring.datasource.maxWait:60000}")
private int maxWait;
@Value("${spring.datasource.timeBetweenEvictionRunsMilli:50000}")
private int timeBetweenEvictionRunsMillis;
@Value("${spring.datasource.minEvictableIdleTimeMillis:200000}")
private int minEvictableIdleTimeMillis;
@Value("${spring.datasource.validationQuery:SELECT 1 FROM DUAL}")
private String validationQuery;
@Value("${spring.datasource.testWhileIdle:true}")
private boolean testWhileIdle;
@Value("${spring.datasource.testOnBorrow:false}")
private boolean testOnBorrow;
@Value("${spring.datasource.testOnReturn:false}")
private boolean testOnReturn;
@Value("${spring.datasource.poolPreparedStatements:true}")
private boolean poolPreparedStatements;
@Value("${spring.datasource.connectionProperties:druid.stat.mergeSql=true;druid.stat.slowSqlMillis=5000}")
private String properties;
@Bean
public DataSource dataSource(){
DruidDataSource configDataSource = new DruidDataSource();
configDataSource.setUrl(url);
configDataSource.setDriverClassName(driverClassName);
configDataSource.setUsername(userName);
configDataSource.setPassword(password);
configDataSource.setInitialSize(initialSize);
configDataSource.setDefaultAutoCommit(true);
configDataSource.setMinIdle(minIdle);
configDataSource.setMaxActive(maxActive);
configDataSource.setMaxWait(maxWait);
configDataSource.setTimeBetweenEvictionRunsMillis(timeBetweenEvictionRunsMillis);
configDataSource.setMinEvictableIdleTimeMillis(minEvictableIdleTimeMillis);
configDataSource.setValidationQuery(validationQuery);
configDataSource.setTestWhileIdle(testWhileIdle);
configDataSource.setTestOnBorrow(testOnBorrow);
configDataSource.setTestOnReturn(testOnReturn);
configDataSource.setConnectionProperties(properties);
return configDataSource;
}
}
3 ,建立Quartz排程配置類
package com.yq.config;
import com.yq.job.HelloWorldJob;
import com.yq.job.UpdateRunningDaysJob;
import javax.sql.DataSource;
import lombok.extern.slf4j.Slf4j;
import org.quartz.CronTrigger;
import org.quartz.JobDetail;
import org.quartz.spi.JobFactory;
import org.quartz.spi.TriggerFiredBundle;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
import org.springframework.beans.factory.config.PropertiesFactoryBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.scheduling.quartz.CronTriggerFactoryBean;
import org.springframework.scheduling.quartz.JobDetailFactoryBean;
import org.springframework.scheduling.quartz.SchedulerFactoryBean;
import org.springframework.scheduling.quartz.SpringBeanJobFactory;
import java.io.IOException;
import java.util.Properties;
@Configuration
@Slf4j
public class QuartzSchedulerConfig {
@Autowired
private DataSource dataSource;
private static final String QUARTZ_PROPERTIES_NAME = "/quartz.properties";
@Bean
public JobFactory jobFactory(ApplicationContext applicationContext) {
AutowiringSpringBeanJobFactory jobFactory = new AutowiringSpringBeanJobFactory();
jobFactory.setApplicationContext(applicationContext);
return jobFactory;
}
@Bean
public SchedulerFactoryBean schedulerFactoryBean(JobFactory jobFactory, CronTrigger[] cronTrigger,
JobDetail[] jobDetails) {
SchedulerFactoryBean factoryBean = new SchedulerFactoryBean();
try {
factoryBean.setQuartzProperties(quartzProperties());
factoryBean.setDataSource(dataSource);
factoryBean.setJobFactory(jobFactory);
factoryBean.setTriggers(cronTrigger);
factoryBean.setJobDetails(jobDetails);
factoryBean.setOverwriteExistingJobs(true);
} catch (Exception e) {
log.error("Failed to load config file {}.", QUARTZ_PROPERTIES_NAME, e);
throw new RuntimeException("LoadingConfigFileError", e);
}
return factoryBean;
}
@Bean(name = "job1Trigger")
public CronTriggerFactoryBean job1Trigger(@Qualifier("jobHelloWorldDetail") JobDetail jobDetail) {
CronTriggerFactoryBean cronTriggerFactoryBean = new CronTriggerFactoryBean();
cronTriggerFactoryBean.setJobDetail(jobDetail);
//這裡為了演示效果,直接hardcode為"0/15 * * * * ?", 實際專案中可以配置在application.propertis中
cronTriggerFactoryBean.setCronExpression("0/15 * * * * ?");
return cronTriggerFactoryBean;
}
@Bean(name = "jobHelloWorldDetail")
public JobDetailFactoryBean job1Detail() {
JobDetailFactoryBean jobDetailFactoryBean = new JobDetailFactoryBean();
//HelloWorldJob.class是具體的執行任務的job類
jobDetailFactoryBean.setJobClass(HelloWorldJob.class);
jobDetailFactoryBean.setDurability(true);
return jobDetailFactoryBean;
}
@Bean(name = "job2Trigger")
public CronTriggerFactoryBean job2Trigger(@Qualifier("jobUpdateRunningDaysDetail") JobDetail jobDetail) {
CronTriggerFactoryBean cronTriggerFactoryBean = new CronTriggerFactoryBean();
cronTriggerFactoryBean.setJobDetail(jobDetail);
//這裡為了演示效果,直接hardcode為"0/15 * * * * ?", 實際專案中可以配置在application.propertis中
cronTriggerFactoryBean.setCronExpression("0/5 * * * * ?");
return cronTriggerFactoryBean;
}
@Bean(name = "jobUpdateRunningDaysDetail")
public JobDetailFactoryBean job2Detail() {
JobDetailFactoryBean jobDetailFactoryBean = new JobDetailFactoryBean();
//UpdateRunningDaysJob.class是具體的執行任務的job類
jobDetailFactoryBean.setJobClass(UpdateRunningDaysJob.class);
jobDetailFactoryBean.setDurability(true);
return jobDetailFactoryBean;
}
@Bean
public Properties quartzProperties() throws IOException {
PropertiesFactoryBean propertiesFactoryBean = new PropertiesFactoryBean();
propertiesFactoryBean.setLocation(new ClassPathResource(QUARTZ_PROPERTIES_NAME));
propertiesFactoryBean.afterPropertiesSet();
return propertiesFactoryBean.getObject();
}
class AutowiringSpringBeanJobFactory extends SpringBeanJobFactory implements ApplicationContextAware {
private transient AutowireCapableBeanFactory beanFactory;
@Override
public void setApplicationContext(final ApplicationContext context) {
beanFactory = context.getAutowireCapableBeanFactory();
}
@Override
protected Object createJobInstance(final TriggerFiredBundle bundle) throws Exception {
final Object job = super.createJobInstance(bundle);
beanFactory.autowireBean(job);
return job;
}
}
}
4, 建立job
job是schedule具體執行的任務,我們可以在job中引入serviceImpl類,這裡是示例程式碼因此只打印日誌。job繼承QuartzJobBean ,只需要實現executeInternal方法即可。
package com.yq.job;
import lombok.extern.slf4j.Slf4j;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.springframework.scheduling.quartz.QuartzJobBean;
import org.springframework.stereotype.Component;
import java.text.SimpleDateFormat;
import java.util.Date;
@Component
@Slf4j
public class HelloWorldJob extends QuartzJobBean {
@Override
protected void executeInternal(JobExecutionContext jobExecutionContext) throws JobExecutionException {
log.info("HelloWorldJob at {}", (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS")).format(new Date()));
}
}
5, 配置quartz.properties
org.quartz.scheduler.instanceName = quartzScheduler
org.quartz.scheduler.instanceId = AUTO
org.quartz.jobStore.class = org.quartz.impl.jdbcjobstore.JobStoreTX
org.quartz.jobStore.driverDelegateClass = org.quartz.impl.jdbcjobstore.StdJDBCDelegate
org.quartz.jobStore.tablePrefix = QRTZ_
org.quartz.jobStore.isClustered = true
org.quartz.jobStore.useProperties = false
org.quartz.jobStore.clusterCheckinInterval = 20000
org.quartz.threadPool.class = org.quartz.simpl.SimpleThreadPool
org.quartz.threadPool.threadCount = 10
org.quartz.threadPool.threadPriority = 5
org.quartz.threadPool.threadsInheritContextClassLoaderOfInitializingThread = true
完整的程式碼在這裡,執行多個例項後可以觀察日誌,只有一個例項執行schedule,多個例項之間隨機獲得機會執行schedule任務。