Quartz实现分布式集群schedule调度(定时任务执行)

一时失言乱红尘 2022-04-16 06:39 350阅读 0赞

**项目需求：**  
需要执行一些定时任务，比如每间隔几小时同步其他系统中信息，或者定期监控一些服务是否正常。如果定时任务程序有多个实例在运行，就是运行多次，因此需要集群分布式的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程序，完整代码在[这里][Link 1]， 欢迎加星，fork。 我使用了本机的数据库，用户名和密码也配置了，运行时需要先创建数据库，创建表，然后修改application.properties中的配置信息。  
因此具体步骤如下

1.  初始数据库  
    初始脚本在 quartz-2.2.3\\docs\\dbTables下面。我用的是mysql数据innodb引擎，因此运行tables\_mysql\_innodb.sql脚本。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1c3NsZQ_size_16_color_FFFFFF_t_70]
2.  引入依赖包

这里主要列出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>

1.  创建数据源配置

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

完整的代码在[这里][Link 1]，运行多个实例后可以观察日志，只有一个实例执行schedule，多个实例之间随机获得机会执行schedule任务。

[Link 1]: https://github.com/yqbjtu/springboot/tree/master/DistScheduler
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3J1c3NsZQ_size_16_color_FFFFFF_t_70]: /images/20220416/39dec7b516c14be0a6a8cabab44efd5f.png