SpringBoot整合Quartz 实现分布式定时任务调度

爱被打了一巴掌 2023-09-30 22:22 92阅读 0赞

一、Quartz 集群架构
Quartz 是 Java 领域最著名的开源任务调度工具。

在上篇文章中，我们详细的介绍了 Quartz 的单体应用实践，如果只在单体环境中应用，Quartz 未必是最好的选择，例如Spring Scheduled一样也可以实现任务调度，并且与SpringBoot无缝集成，支持注解配置，非常简单，但是它有个缺点就是在集群环境下，会导致任务被重复调度！

而与之对应的 Quartz 提供了极为广用的特性，如任务持久化、集群部署和分布式调度任务等等，正因如此，基于 Quartz 任务调度功能在系统开发中应用极为广泛！

在集群环境下，Quartz 集群中的每个节点是一个独立的 Quartz 应用，没有负责集中管理的节点，而是通过数据库表来感知另一个应用，利用数据库锁的方式来实现集群环境下进行并发控制，每个任务当前运行的有效节点有且只有一个！

bd56cb604a5442408b6b300e137e6612.jpg

特别需要注意的是：分布式部署时需要保证各个节点的系统时间一致！

二、数据表初始化
数据库表结构官网已经提供，我们可以直接访问Quartz对应的官方网站，找到对应的版本，然后将其下载！

watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5Zu-5Zu-5bCP5reY5rCUX3JlYWw_size_20_color_FFFFFF_t_70_g_se_x_16

我选择的是quartz-2.3.0-distribution.tar.gz，下载完成之后将其解压，在文件中搜索sql，在里面选择适合当前环境的数据库脚本文件，然后将其初始化到数据库中即可！

watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5Zu-5Zu-5bCP5reY5rCUX3JlYWw_size_20_color_FFFFFF_t_70_g_se_x_16 1

例如，我使用的数据库是mysql-5.7，因此我选择的是tables_mysql_innodb.sql脚本，具体内容如下： 

  1. DROP TABLE IF EXISTS QRTZ_FIRED_TRIGGERS;
  2. DROP TABLE IF EXISTS QRTZ_PAUSED_TRIGGER_GRPS;
  3. DROP TABLE IF EXISTS QRTZ_SCHEDULER_STATE;
  4. DROP TABLE IF EXISTS QRTZ_LOCKS;
  5. DROP TABLE IF EXISTS QRTZ_SIMPLE_TRIGGERS;
  6. DROP TABLE IF EXISTS QRTZ_SIMPROP_TRIGGERS;
  7. DROP TABLE IF EXISTS QRTZ_CRON_TRIGGERS;
  8. DROP TABLE IF EXISTS QRTZ_BLOB_TRIGGERS;
  9. DROP TABLE IF EXISTS QRTZ_TRIGGERS;
  10. DROP TABLE IF EXISTS QRTZ_JOB_DETAILS;
  11. DROP TABLE IF EXISTS QRTZ_CALENDARS;
  13. CREATE TABLE QRTZ_JOB_DETAILS(
  14. SCHED_NAME VARCHAR(120) NOT NULL,
  15. JOB_NAME VARCHAR(190) NOT NULL,
  16. JOB_GROUP VARCHAR(190) NOT NULL,
  17. DESCRIPTION VARCHAR(250) NULL,
  18. JOB_CLASS_NAME VARCHAR(250) NOT NULL,
  19. IS_DURABLE VARCHAR(1) NOT NULL,
  20. IS_NONCONCURRENT VARCHAR(1) NOT NULL,
  21. IS_UPDATE_DATA VARCHAR(1) NOT NULL,
  22. REQUESTS_RECOVERY VARCHAR(1) NOT NULL,
  23. JOB_DATA BLOB NULL,
  24. PRIMARY KEY (SCHED_NAME,JOB_NAME,JOB_GROUP))
  25. ENGINE=InnoDB;
  27. CREATE TABLE QRTZ_TRIGGERS (
  28. SCHED_NAME VARCHAR(120) NOT NULL,
  29. TRIGGER_NAME VARCHAR(190) NOT NULL,
  30. TRIGGER_GROUP VARCHAR(190) NOT NULL,
  31. JOB_NAME VARCHAR(190) NOT NULL,
  32. JOB_GROUP VARCHAR(190) NOT NULL,
  33. DESCRIPTION VARCHAR(250) NULL,
  34. NEXT_FIRE_TIME BIGINT(13) NULL,
  35. PREV_FIRE_TIME BIGINT(13) NULL,
  36. PRIORITY INTEGER NULL,
  37. TRIGGER_STATE VARCHAR(16) NOT NULL,
  38. TRIGGER_TYPE VARCHAR(8) NOT NULL,
  39. START_TIME BIGINT(13) NOT NULL,
  40. END_TIME BIGINT(13) NULL,
  41. CALENDAR_NAME VARCHAR(190) NULL,
  42. MISFIRE_INSTR SMALLINT(2) NULL,
  43. JOB_DATA BLOB NULL,
  44. PRIMARY KEY (SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP),
  45. FOREIGN KEY (SCHED_NAME,JOB_NAME,JOB_GROUP)
  46. REFERENCES QRTZ_JOB_DETAILS(SCHED_NAME,JOB_NAME,JOB_GROUP))
  47. ENGINE=InnoDB;
  49. CREATE TABLE QRTZ_SIMPLE_TRIGGERS (
  50. SCHED_NAME VARCHAR(120) NOT NULL,
  51. TRIGGER_NAME VARCHAR(190) NOT NULL,
  52. TRIGGER_GROUP VARCHAR(190) NOT NULL,
  53. REPEAT_COUNT BIGINT(7) NOT NULL,
  54. REPEAT_INTERVAL BIGINT(12) NOT NULL,
  55. TIMES_TRIGGERED BIGINT(10) NOT NULL,
  56. PRIMARY KEY (SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP),
  57. FOREIGN KEY (SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP)
  58. REFERENCES QRTZ_TRIGGERS(SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP))
  59. ENGINE=InnoDB;
  61. CREATE TABLE QRTZ_CRON_TRIGGERS (
  62. SCHED_NAME VARCHAR(120) NOT NULL,
  63. TRIGGER_NAME VARCHAR(190) NOT NULL,
  64. TRIGGER_GROUP VARCHAR(190) NOT NULL,
  65. CRON_EXPRESSION VARCHAR(120) NOT NULL,
  66. TIME_ZONE_ID VARCHAR(80),
  67. PRIMARY KEY (SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP),
  68. FOREIGN KEY (SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP)
  69. REFERENCES QRTZ_TRIGGERS(SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP))
  70. ENGINE=InnoDB;
  72. CREATE TABLE QRTZ_SIMPROP_TRIGGERS
  73.   (
  74.     SCHED_NAME VARCHAR(120) NOT NULL,
  75.     TRIGGER_NAME VARCHAR(190) NOT NULL,
  76.     TRIGGER_GROUP VARCHAR(190) NOT NULL,
  77.     STR_PROP_1 VARCHAR(512) NULL,
  78.     STR_PROP_2 VARCHAR(512) NULL,
  79.     STR_PROP_3 VARCHAR(512) NULL,
  80.     INT_PROP_1 INT NULL,
  81.     INT_PROP_2 INT NULL,
  82.     LONG_PROP_1 BIGINT NULL,
  83.     LONG_PROP_2 BIGINT NULL,
  84.     DEC_PROP_1 NUMERIC(13,4) NULL,
  85.     DEC_PROP_2 NUMERIC(13,4) NULL,
  86.     BOOL_PROP_1 VARCHAR(1) NULL,
  87.     BOOL_PROP_2 VARCHAR(1) NULL,
  88.     PRIMARY KEY (SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP),
  89.     FOREIGN KEY (SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP)
  90.     REFERENCES QRTZ_TRIGGERS(SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP))
  91. ENGINE=InnoDB;
  93. CREATE TABLE QRTZ_BLOB_TRIGGERS (
  94. SCHED_NAME VARCHAR(120) NOT NULL,
  95. TRIGGER_NAME VARCHAR(190) NOT NULL,
  96. TRIGGER_GROUP VARCHAR(190) NOT NULL,
  97. BLOB_DATA BLOB NULL,
  98. PRIMARY KEY (SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP),
  99. INDEX (SCHED_NAME,TRIGGER_NAME, TRIGGER_GROUP),
  100. FOREIGN KEY (SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP)
  101. REFERENCES QRTZ_TRIGGERS(SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP))
  102. ENGINE=InnoDB;
  104. CREATE TABLE QRTZ_CALENDARS (
  105. SCHED_NAME VARCHAR(120) NOT NULL,
  106. CALENDAR_NAME VARCHAR(190) NOT NULL,
  107. CALENDAR BLOB NOT NULL,
  108. PRIMARY KEY (SCHED_NAME,CALENDAR_NAME))
  109. ENGINE=InnoDB;
  111. CREATE TABLE QRTZ_PAUSED_TRIGGER_GRPS (
  112. SCHED_NAME VARCHAR(120) NOT NULL,
  113. TRIGGER_GROUP VARCHAR(190) NOT NULL,
  114. PRIMARY KEY (SCHED_NAME,TRIGGER_GROUP))
  115. ENGINE=InnoDB;
  117. CREATE TABLE QRTZ_FIRED_TRIGGERS (
  118. SCHED_NAME VARCHAR(120) NOT NULL,
  119. ENTRY_ID VARCHAR(95) NOT NULL,
  120. TRIGGER_NAME VARCHAR(190) NOT NULL,
  121. TRIGGER_GROUP VARCHAR(190) NOT NULL,
  122. INSTANCE_NAME VARCHAR(190) NOT NULL,
  123. FIRED_TIME BIGINT(13) NOT NULL,
  124. SCHED_TIME BIGINT(13) NOT NULL,
  125. PRIORITY INTEGER NOT NULL,
  126. STATE VARCHAR(16) NOT NULL,
  127. JOB_NAME VARCHAR(190) NULL,
  128. JOB_GROUP VARCHAR(190) NULL,
  129. IS_NONCONCURRENT VARCHAR(1) NULL,
  130. REQUESTS_RECOVERY VARCHAR(1) NULL,
  131. PRIMARY KEY (SCHED_NAME,ENTRY_ID))
  132. ENGINE=InnoDB;
  134. CREATE TABLE QRTZ_SCHEDULER_STATE (
  135. SCHED_NAME VARCHAR(120) NOT NULL,
  136. INSTANCE_NAME VARCHAR(190) NOT NULL,
  137. LAST_CHECKIN_TIME BIGINT(13) NOT NULL,
  138. CHECKIN_INTERVAL BIGINT(13) NOT NULL,
  139. PRIMARY KEY (SCHED_NAME,INSTANCE_NAME))
  140. ENGINE=InnoDB;
  142. CREATE TABLE QRTZ_LOCKS (
  143. SCHED_NAME VARCHAR(120) NOT NULL,
  144. LOCK_NAME VARCHAR(40) NOT NULL,
  145. PRIMARY KEY (SCHED_NAME,LOCK_NAME))
  146. ENGINE=InnoDB;
  148. CREATE INDEX IDX_QRTZ_J_REQ_RECOVERY ON QRTZ_JOB_DETAILS(SCHED_NAME,REQUESTS_RECOVERY);
  149. CREATE INDEX IDX_QRTZ_J_GRP ON QRTZ_JOB_DETAILS(SCHED_NAME,JOB_GROUP);
  151. CREATE INDEX IDX_QRTZ_T_J ON QRTZ_TRIGGERS(SCHED_NAME,JOB_NAME,JOB_GROUP);
  152. CREATE INDEX IDX_QRTZ_T_JG ON QRTZ_TRIGGERS(SCHED_NAME,JOB_GROUP);
  153. CREATE INDEX IDX_QRTZ_T_C ON QRTZ_TRIGGERS(SCHED_NAME,CALENDAR_NAME);
  154. CREATE INDEX IDX_QRTZ_T_G ON QRTZ_TRIGGERS(SCHED_NAME,TRIGGER_GROUP);
  155. CREATE INDEX IDX_QRTZ_T_STATE ON QRTZ_TRIGGERS(SCHED_NAME,TRIGGER_STATE);
  156. CREATE INDEX IDX_QRTZ_T_N_STATE ON QRTZ_TRIGGERS(SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP,TRIGGER_STATE);
  157. CREATE INDEX IDX_QRTZ_T_N_G_STATE ON QRTZ_TRIGGERS(SCHED_NAME,TRIGGER_GROUP,TRIGGER_STATE);
  158. CREATE INDEX IDX_QRTZ_T_NEXT_FIRE_TIME ON QRTZ_TRIGGERS(SCHED_NAME,NEXT_FIRE_TIME);
  159. CREATE INDEX IDX_QRTZ_T_NFT_ST ON QRTZ_TRIGGERS(SCHED_NAME,TRIGGER_STATE,NEXT_FIRE_TIME);
  160. CREATE INDEX IDX_QRTZ_T_NFT_MISFIRE ON QRTZ_TRIGGERS(SCHED_NAME,MISFIRE_INSTR,NEXT_FIRE_TIME);
  161. CREATE INDEX IDX_QRTZ_T_NFT_ST_MISFIRE ON QRTZ_TRIGGERS(SCHED_NAME,MISFIRE_INSTR,NEXT_FIRE_TIME,TRIGGER_STATE);
  162. CREATE INDEX IDX_QRTZ_T_NFT_ST_MISFIRE_GRP ON QRTZ_TRIGGERS(SCHED_NAME,MISFIRE_INSTR,NEXT_FIRE_TIME,TRIGGER_GROUP,TRIGGER_STATE);
  164. CREATE INDEX IDX_QRTZ_FT_TRIG_INST_NAME ON QRTZ_FIRED_TRIGGERS(SCHED_NAME,INSTANCE_NAME);
  165. CREATE INDEX IDX_QRTZ_FT_INST_JOB_REQ_RCVRY ON QRTZ_FIRED_TRIGGERS(SCHED_NAME,INSTANCE_NAME,REQUESTS_RECOVERY);
  166. CREATE INDEX IDX_QRTZ_FT_J_G ON QRTZ_FIRED_TRIGGERS(SCHED_NAME,JOB_NAME,JOB_GROUP);
  167. CREATE INDEX IDX_QRTZ_FT_JG ON QRTZ_FIRED_TRIGGERS(SCHED_NAME,JOB_GROUP);
  168. CREATE INDEX IDX_QRTZ_FT_T_G ON QRTZ_FIRED_TRIGGERS(SCHED_NAME,TRIGGER_NAME,TRIGGER_GROUP);
  169. CREATE INDEX IDX_QRTZ_FT_TG ON QRTZ_FIRED_TRIGGERS(SCHED_NAME,TRIGGER_GROUP);
  171. commit;

watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5Zu-5Zu-5bCP5reY5rCUX3JlYWw_size_20_color_FFFFFF_t_70_g_se_x_16 2

其中，QRTZ_LOCKS 就是 Quartz 集群实现同步机制的行锁表！

三、Quartz 集群实践

1、引入quartz依赖

  1. <dependency>
  2.             <groupId>org.springframework.boot</groupId>
  3.             <artifactId>spring-boot-starter-quartz</artifactId>
  4.         </dependency>

2、创建 application.properties 配置文件，再配置文件中加入quartz配置

  1. #调度配置
  2. org:
  3.   quartz:
  4.     scheduler:
  5.       #调度器实例名称
  6.       instanceName: MaterialFromEamScheduler
  7.       #调度器实例编号自动生成
  8.       instanceId: AUTO
  9.       #是否在Quartz执行一个job前使用UserTransaction
  10.       wrapJobExecutionInUserTransaction: false
  11. #线程池配置
  12.     threadPool:
  13.       #线程池的实现类
  14.       class: org.quartz.simpl.SimpleThreadPool
  15.       #线程池中的线程数量
  16.       threadCount: 10
  17.       #线程优先级
  18.       threadPriority: 5
  19.       #配置是否启动自动加载数据库内的定时任务，默认true
  20.       threadsInheritContextClassLoaderOfInitializingThread: true
  21.     #持久化方式配置
  22.     jobStore:
  23.       #JobDataMaps是否都为String类型
  24.       useProperties: true
  25.       #数据表的前缀，默认QRTZ_
  26.       tablePrefix: QRTZ_
  27.       #最大能忍受的触发超时时间
  28.       misfireThreshold: 60000
  29.       #是否以集群方式运行
  30.       isClustered: true
  31.       #调度实例失效的检查时间间隔，单位毫秒
  32.       clusterCheckinInterval: 2000
  33.       maxMisfiresToHandleAtATime: 1
  34.       #数据保存方式为数据库持久化
  35.       class: org.quartz.impl.jdbcjobstore.JobStoreTX
  36.       #数据库代理类，一般org.quartz.impl.jdbcjobstore.StdJDBCDelegate可以满足大部分数据库
  37.       driverDelegateClass: org.quartz.impl.jdbcjobstore.PostgreSQLDelegate
  38.       # 是否开启悲观锁控制集群中trigger并发
  39.       acquireTriggersWithinLock: true
  40.       #数据库别名 随便取
  41.       dataSource: qzDS
  42.       #数据库连接池，将其设置为druid
  43.     dataSource:
  44.       qzDS:
  45.         connectionProvider:
  46.           class: com.mes.material.quartz.provider.DruidConnectionProvider
  47.         #数据库引擎
  48.         driver: org.postgresql.Driver
  49.         #数据库连接
  50.         URL: jdbc:postgresql://127.0.0.1:5432/mes_cloud_stag_yqs?serverTimezone=UTC&useUnicode=true&characterEncoding=utf-8&useSSL=true
  51.         #数据库用户
  52.         user: postgres
  53.         #数据库密码
  54.         password: postgres
  55.         #允许最大连接
  56.         maxConnection: 100

3、注册 Quartz 任务工厂

  1. import org.quartz.spi.TriggerFiredBundle;
  2. import org.springframework.beans.factory.annotation.Autowired;
  3. import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
  4. import org.springframework.scheduling.quartz.AdaptableJobFactory;
  5. import org.springframework.stereotype.Component;
  7. @Component
  8. public class QuartzJobFactory extends AdaptableJobFactory {
  10.     @Autowired
  11.     private AutowireCapableBeanFactory capableBeanFactory;
  13.     @Override
  14.     protected Object createJobInstance(TriggerFiredBundle bundle) throws Exception {
  15.         //调用父类的方法
  16.         Object jobInstance = super.createJobInstance(bundle);
  17.         //进行注入
  18.         capableBeanFactory.autowireBean(jobInstance);
  19.         return jobInstance;
  20.     }
  21. }

4、注册调度工厂

本文采用自定义数据源实现

  1. import com.mes.material.constants.MaterialConstants;
  2. import com.mes.material.quartz.factory.QuartzJobFactory;
  3. import com.mes.material.quartz.listener.SimpleJobListener;
  4. import com.mes.material.quartz.listener.SimpleSchedulerListener;
  5. import com.mes.material.quartz.listener.SimpleTriggerListener;
  6. import org.quartz.JobKey;
  7. import org.quartz.Scheduler;
  8. import org.quartz.SchedulerException;
  9. import org.quartz.impl.matchers.EverythingMatcher;
  10. import org.quartz.impl.matchers.KeyMatcher;
  11. import org.springframework.beans.factory.annotation.Autowired;
  12. import org.springframework.beans.factory.annotation.Value;
  13. import org.springframework.context.annotation.Bean;
  14. import org.springframework.context.annotation.Configuration;
  15. import org.springframework.scheduling.quartz.SchedulerFactoryBean;
  17. import javax.sql.DataSource;
  18. import java.io.IOException;
  19. import java.util.Properties;
  21. /** @Author: 
  22.  * @Description: 3.5、注册调度工厂
  23.  * @Date: 15:42 2022/8/10
  24.  * @Param
  25.  * @return
  26.  **/
  28. @Configuration
  29. public class QuartzConfig {
  31.     @Autowired
  32.     private QuartzJobFactory jobFactory;
  34.     @Autowired
  35.     private SimpleSchedulerListener simpleSchedulerListener;
  37.     @Autowired
  38.     private SimpleJobListener simpleJobListener;
  40.     @Autowired
  41.     private SimpleTriggerListener simpleTriggerListener;
  42.     @Autowired
  43.     private DataSource dataSource;
  45.     @Value("${org.quartz.scheduler.instanceName}")
  46.     private String instanceName;
  47.     @Value("${org.quartz.scheduler.instanceId}")
  48.     private String instanceId;
  49.     @Value("${org.quartz.threadPool.class}")
  50.     private String threadPoolClass;
  51.     @Value("${org.quartz.threadPool.threadCount}")
  52.     private String threadCount;
  53.     @Value("${org.quartz.threadPool.threadPriority}")
  54.     private String threadPriority;
  55.     @Value("${org.quartz.jobStore.class}")
  56.     private String jobStoreClass;
  57.     @Value("${org.quartz.jobStore.isClustered}")
  58.     private String isClustered;
  59.     @Value("${org.quartz.jobStore.clusterCheckinInterval}")
  60.     private String clusterCheckinInterval;
  61.     @Value("${org.quartz.jobStore.maxMisfiresToHandleAtATime}")
  62.     private String maxMisfiresToHandleAtATime;
  63.     @Value("${org.quartz.jobStore.misfireThreshold}")
  64.     private String misfireThreshold;
  65.     @Value("${org.quartz.jobStore.tablePrefix}")
  66.     private String tablePrefix;
  67.     @Value("${org.quartz.jobStore.acquireTriggersWithinLock}")
  68.     private String acquireTriggersWithinLock;
  69.     @Value("${org.quartz.jobStore.driverDelegateClass}")
  70.     private String driverDelegateClass;
  71.     @Value("${org.quartz.jobStore.dataSource}")
  72.     private String dataSourceName;
  73.     @Value("${org.quartz.dataSource.qzDS.connectionProvider.class}")
  74.     private String connectionProviderClass;
  75.     @Value("${org.quartz.dataSource.qzDS.driver}")
  76.     private String driver;
  77.     @Value("${org.quartz.dataSource.qzDS.URL}")
  78.     private String url;
  79.     @Value("${org.quartz.dataSource.qzDS.user}")
  80.     private String user;
  81.     @Value("${org.quartz.dataSource.qzDS.password}")
  82.     private String password;
  83.     @Value("${org.quartz.dataSource.qzDS.maxConnection}")
  84.     private String maxConnection;
  85.     @Value("${org.quartz.threadPool.threadsInheritContextClassLoaderOfInitializingThread}")
  86.     private String threadsInheritContextClassLoaderOfInitializingThread;
  89.     @Bean
  90.     public SchedulerFactoryBean schedulerFactoryBean() throws IOException {
  91.         //创建SchedulerFactoryBean
  92.         SchedulerFactoryBean factory = new SchedulerFactoryBean();
  94.         //quartz参数
  95.         Properties prop = new Properties();
  96.         //调度器实例名称
  97.         prop.put("org.quartz.scheduler.instanceName", instanceName);
  98.         //调度器实例编号自动生成
  99.         prop.put("org.quartz.scheduler.instanceId", instanceId);
  100.         //线程池配置--线程池的实现类
  101.         prop.put("org.quartz.threadPool.class", threadPoolClass);
  102.         //线程池中的线程数量
  103.         prop.put("org.quartz.threadPool.threadCount", threadCount);
  104.         //配置是否启动自动加载数据库内的定时任务，默认true
  105.         prop.put("org.quartz.threadPool.threadsInheritContextClassLoaderOfInitializingThread", threadsInheritContextClassLoaderOfInitializingThread);
  106.         //线程优先级
  107.         prop.put("org.quartz.threadPool.threadPriority", threadPriority);
  108.         //JobStore配置-数据保存方式为数据库持久化
  109.         prop.put("org.quartz.jobStore.class", jobStoreClass);
  110.         //集群配置--是否以集群方式运行
  111.         prop.put("org.quartz.jobStore.isClustered", isClustered);
  112.         //调度实例失效的检查时间间隔，单位毫秒
  113.         prop.put("org.quartz.jobStore.clusterCheckinInterval", clusterCheckinInterval);
  114.         prop.put("org.quartz.jobStore.maxMisfiresToHandleAtATime", maxMisfiresToHandleAtATime);
  115.         //最大能忍受的触发超时时间
  116.         prop.put("org.quartz.jobStore.misfireThreshold", misfireThreshold);
  117.         //数据表的前缀，默认QRTZ_
  118.         prop.put("org.quartz.jobStore.tablePrefix", tablePrefix);
  119.         //是否开启悲观锁控制集群中trigger并发
  120.         prop.put("org.quartz.jobStore.acquireTriggersWithinLock", acquireTriggersWithinLock);
  122.         //数据库代理类，一般org.quartz.impl.jdbcjobstore.StdJDBCDelegate可以满足大部分数据库
  123.         prop.put("org.quartz.jobStore.driverDelegateClass", driverDelegateClass);
  124.         //数据库别名 随便取
  125.         prop.put("org.quartz.jobStore.dataSource",dataSourceName);
  126.         //数据库连接池，将其设置为druid
  127.         prop.put("org.quartz.dataSource.qzDS.connectionProvider.class",connectionProviderClass);
  128.         //数据库引擎
  129.         prop.put("org.quartz.dataSource.qzDS.driver",driver);
  130.         //数据库连接
  131.         prop.put("org.quartz.dataSource.qzDS.URL",url);
  132.         //数据库用户
  133.         prop.put("org.quartz.dataSource.qzDS.user",user);
  134.         //数据库密码
  135.         prop.put("org.quartz.dataSource.qzDS.password",password);
  136.         //允许最大连接
  137.         prop.put("org.quartz.dataSource.qzDS.maxConnection",maxConnection);
  139.         factory.setQuartzProperties(prop);
  141.         //支持在JOB实例中注入其他的业务对象
  142.         factory.setJobFactory(jobFactory);
  143.         factory.setApplicationContextSchedulerContextKey("applicationContextKey");
  144.         //这样当spring关闭时，会等待所有已经启动的quartz job结束后spring才能完全shutdown。
  145.         factory.setWaitForJobsToCompleteOnShutdown(true);
  146.         //是否覆盖己存在的Job
  147.         factory.setOverwriteExistingJobs(false);
  148.         //QuartzScheduler 延时启动，应用启动完后 QuartzScheduler 再启动
  149.         factory.setStartupDelay(10);
  150.         //使用数据源，自定义数据源(采用项目数据源)
  151.         //在quartz.properties配置文件中，去掉org.quartz.jobStore.dataSource配置
  152. //        factory.setDataSource(dataSource);
  153.         return factory;
  154.     }
  156.     /**
  157.      * 通过SchedulerFactoryBean获取Scheduler的实例
  158.      * @return
  159.      * @throws IOException
  160.      * @throws SchedulerException
  161.      */
  162.     @Bean(name = "scheduler")
  163.     public Scheduler scheduler() throws IOException, SchedulerException {
  164.         Scheduler scheduler = schedulerFactoryBean().getScheduler();
  165.         //全局添加监听器
  166.         //添加SchedulerListener监听器
  167.         scheduler.getListenerManager().addSchedulerListener(simpleSchedulerListener);
  169.         // 添加JobListener, 支持带条件匹配监听器
  170.         scheduler.getListenerManager().addJobListener(simpleJobListener, KeyMatcher.keyEquals(JobKey.jobKey(MaterialConstants.QURATZ_NAME,MaterialConstants.QURATZ_GROUP_NAME)));
  172.         // 添加triggerListener，设置全局监听
  173.         scheduler.getListenerManager().addTriggerListener(simpleTriggerListener, EverythingMatcher.allTriggers());
  174.         return scheduler;
  175.     }
  176. }

5、重新设置 Quartz 数据连接池

  1. import com.alibaba.druid.pool.DruidDataSource;
  2. import org.quartz.SchedulerException;
  3. import org.quartz.utils.ConnectionProvider;
  5. import java.sql.Connection;
  6. import java.sql.SQLException;
  8. /** @Author: 
  9.  * @Description: 3.6、重新设置 Quartz 数据连接池
  10.  * @Date: 15:40 2022/8/10
  11.  * @Param
  12.  * @return
  13.  **/
  14. public class DruidConnectionProvider implements ConnectionProvider {
  16.     /**
  17.      * 常量配置，与quartz.properties文件的key保持一致(去掉前缀)，同时提供set方法，Quartz框架自动注入值。
  18.      * @return
  19.      * @throws SQLException
  20.      */
  22.     /**JDBC驱动*/
  23.     public String driver;
  24.     /**JDBC连接串*/
  25.     public String URL;
  26.     /**数据库用户名*/
  27.     public String user;
  28.     /**数据库用户密码*/
  29.     public String password;
  30.     /**数据库最大连接数*/
  31.     public int maxConnection;
  32.     /**数据库SQL查询每次连接返回执行到连接池，以确保它仍然是有效的。*/
  33.     public String validationQuery;
  35.     private boolean validateOnCheckout;
  37.     private int idleConnectionValidationSeconds;
  39.     public String maxCachedStatementsPerConnection;
  41.     private String discardIdleConnectionsSeconds;
  43.     public static final int DEFAULT_DB_MAX_CONNECTIONS = 10;
  45.     public static final int DEFAULT_DB_MAX_CACHED_STATEMENTS_PER_CONNECTION = 120;
  47.     //Druid连接池
  48.     private DruidDataSource datasource;
  50.     @Override
  51.     public Connection getConnection() throws SQLException {
  52.         return datasource.getConnection();
  53.     }
  55.     @Override
  56.     public void shutdown() throws SQLException {
  57.         datasource.close();
  58.     }
  60.     @Override
  61.     public void initialize() throws SQLException {
  62.         if (this.URL == null) {
  63.             throw new SQLException("DBPool could not be created: DB URL cannot be null");
  64.         }
  66.         if (this.driver == null) {
  67.             throw new SQLException("DBPool driver could not be created: DB driver class name cannot be null!");
  68.         }
  70.         if (this.maxConnection < 0) {
  71.             throw new SQLException("DBPool maxConnectins could not be created: Max connections must be greater than zero!");
  72.         }
  74.         datasource = new DruidDataSource();
  75.         try{
  76.             datasource.setDriverClassName(this.driver);
  77.         } catch (Exception e) {
  78.             try {
  79.                 throw new SchedulerException("Problem setting driver class name on datasource: " + e.getMessage(), e);
  80.             } catch (SchedulerException e1) {
  81.             }
  82.         }
  84.         datasource.setUrl(this.URL);
  85.         datasource.setUsername(this.user);
  86.         datasource.setPassword(this.password);
  87.         datasource.setMaxActive(this.maxConnection);
  88.         datasource.setMinIdle(1);
  89.         datasource.setMaxWait(0);
  90.         datasource.setMaxPoolPreparedStatementPerConnectionSize(DEFAULT_DB_MAX_CONNECTIONS);
  92.         if (this.validationQuery != null) {
  93.             datasource.setValidationQuery(this.validationQuery);
  94.             if(!this.validateOnCheckout){
  95.                 datasource.setTestOnReturn(true);
  96.             }else{
  97.                 datasource.setTestOnBorrow(true);
  98.             }
  99.             datasource.setValidationQueryTimeout(this.idleConnectionValidationSeconds);
  100.         }
  101.     }
  103.     public String getDriver() {
  104.         return driver;
  105.     }
  107.     public void setDriver(String driver) {
  108.         this.driver = driver;
  109.     }
  111.     public String getURL() {
  112.         return URL;
  113.     }
  115.     public void setURL(String URL) {
  116.         this.URL = URL;
  117.     }
  119.     public String getUser() {
  120.         return user;
  121.     }
  123.     public void setUser(String user) {
  124.         this.user = user;
  125.     }
  127.     public String getPassword() {
  128.         return password;
  129.     }
  131.     public void setPassword(String password) {
  132.         this.password = password;
  133.     }
  135.     public int getMaxConnection() {
  136.         return maxConnection;
  137.     }
  139.     public void setMaxConnection(int maxConnection) {
  140.         this.maxConnection = maxConnection;
  141.     }
  143.     public String getValidationQuery() {
  144.         return validationQuery;
  145.     }
  147.     public void setValidationQuery(String validationQuery) {
  148.         this.validationQuery = validationQuery;
  149.     }
  151.     public boolean isValidateOnCheckout() {
  152.         return validateOnCheckout;
  153.     }
  155.     public void setValidateOnCheckout(boolean validateOnCheckout) {
  156.         this.validateOnCheckout = validateOnCheckout;
  157.     }
  159.     public int getIdleConnectionValidationSeconds() {
  160.         return idleConnectionValidationSeconds;
  161.     }
  163.     public void setIdleConnectionValidationSeconds(int idleConnectionValidationSeconds) {
  164.         this.idleConnectionValidationSeconds = idleConnectionValidationSeconds;
  165.     }
  167.     public DruidDataSource getDatasource() {
  168.         return datasource;
  169.     }
  171.     public void setDatasource(DruidDataSource datasource) {
  172.         this.datasource = datasource;
  173.     }
  175.     public String getDiscardIdleConnectionsSeconds() {
  176.         return discardIdleConnectionsSeconds;
  177.     }
  179.     public void setDiscardIdleConnectionsSeconds(String discardIdleConnectionsSeconds) {
  180.         this.discardIdleConnectionsSeconds = discardIdleConnectionsSeconds;
  181.     }
  182. }

创建完成之后，还需要在配置文件中设置一下即可！

  1. org:
  2.   quartz:
  3.     dataSource:
  4.       qzDS:
  5.         connectionProvider:
  6.           class: com.mes.material.quartz.provider.DruidConnectionProvider

6、编写 Job 具体任务类

  1. package com.mes.material.quartz.tfjob;
  3. import com.mes.material.eam.domain.MaterialRequestVo;
  4. import com.mes.material.eam.enums.InterfaceTypeEnum;
  5. import com.mes.material.eam.service.StrategyService;
  6. import org.quartz.DisallowConcurrentExecution;
  7. import org.quartz.Job;
  8. import org.quartz.JobExecutionContext;
  9. import org.slf4j.Logger;
  10. import org.slf4j.LoggerFactory;
  11. import org.springframework.beans.factory.annotation.Autowired;
  12. import org.springframework.beans.factory.annotation.Value;
  13. import org.springframework.data.redis.core.RedisTemplate;
  15. import java.util.concurrent.TimeUnit;
  18. /**
  19.  * @Author: 
  20.  * @Description: 3.7、编写 Job 具体任务类
  21.  * @Date Create in 15:43 2022/8/10
  22.  * @Modified By:
  23.  */
  25. //此标记用在实现Job的类上面,意思是不允许并发执行,按照我之前的理解是不允许调度框架在同一时刻调用Job类，后来经过测试发现并不是这样，
  26. // 而是Job(任务)的执行时间[比如需要10秒]大于任务的时间间隔[Interval（5秒)],那么默认情况下,调度框架为了能让
  27. //任务按照我们预定的时间间隔执行,会马上启用新的线程执行任务。否则的话会等待任务执行完毕以后再重新执行！（这样会导致任务的执行不是按照我们预先定义的时间间隔执行）
  28. //测试代码，这是官方提供的例子。设定的时间间隔为3秒,但job执行时间是5秒,设置@DisallowConcurrentExecution以后程序会等任务执行完毕以后再去执行，
  29. // 否则会在3秒时再启用新的线程执行
  31. @DisallowConcurrentExecution
  32. public class TfCommandJob implements Job {
  34.     @Autowired
  35.     private StrategyService strategyService;
  37.     private static final Logger log = LoggerFactory.getLogger(TfCommandJob.class);
  39.     private static final String lock_key = "material_from_eam_info";
  41.     private static final String lock_value = "material_from_eam_info_zkaw";
  43.     @Value("${schedule.expire}")
  44.     private long timeOut;
  46.     @Autowired
  47.     private RedisTemplate redisTemplate;
  49.     @Override
  50.     public void execute(JobExecutionContext context) {
  51.         try {
  52.             //分布式锁
  53.             boolean lock = false;
  54.             try {
  55.                 //如果返回true，说明key不存在，获取到锁
  56.                 lock = redisTemplate.opsForValue().setIfAbsent(lock_key, lock_value);
  57.                 log.info("是否获取到锁:" + lock);
  58.                 if (lock) {
  59.                     log.info("获取到锁，开启定时任务！");
  60.                     //设置过期时间
  61.                     redisTemplate.expire(lock_key, timeOut, TimeUnit.SECONDS);
  62.                     //调用EAM接口，同步物资数据
  63.                     log.info("调用EAM接口，同步物资数据");
  64.                     String type = InterfaceTypeEnum.autoMaterial.getCode();
  65.                     MaterialRequestVo materialRequestVo = new MaterialRequestVo();
  66.                     materialRequestVo.setType(type);
  67.                     strategyService.render(type).materialInfoMesToRoma(materialRequestVo);
  68.                  } else {
  69.                     log.info("其他系统正在执行此项任务");
  70.                     return;
  71.                 }
  72.             } catch (Exception e) {
  73.                 e.printStackTrace();
  74.             } catch (Throwable throwable) {
  75.                 throw new RuntimeException("分布式锁执行发生异常" + throwable.getMessage(), throwable);
  76.             }
  78.         } catch (Exception e) {
  79.             log.error("任务执行失败",e);
  80.         }
  81.     }
  82. }

7、编写 Quartz 服务层接口

  1. package com.mes.material.quartz.service;
  3. import java.util.Map;
  5. /**
  6.  * @Author: 
  7.  * @Description: 3.8、编写 Quartz 服务层接口
  8.  * @Date Create in 14:14 2022/5/16
  9.  * @Modified By:
  10.  */
  11. public interface IQuartzJobService {
  12.     /**
  13.      * 添加任务可以传参数
  14.      * @param clazzName
  15.      * @param jobName
  16.      * @param groupName
  17.      * @param cronExp
  18.      * @param param
  19.      */
  20.     void addJob(String clazzName, String jobName, String groupName, String cronExp, Map<String, Object> param);
  22.     /**
  23.      * 暂停任务
  24.      * @param jobName
  25.      * @param groupName
  26.      */
  27.     void pauseJob(String jobName, String groupName);
  29.     /**
  30.      * 恢复任务
  31.      * @param jobName
  32.      * @param groupName
  33.      */
  34.     void resumeJob(String jobName, String groupName);
  36.     /**
  37.      * 立即运行一次定时任务
  38.      * @param jobName
  39.      * @param groupName
  40.      */
  41.     void runOnce(String jobName, String groupName);
  43.     /**
  44.      * 更新任务
  45.      * @param jobName
  46.      * @param groupName
  47.      * @param cronExp
  48.      * @param param
  49.      */
  50.     void updateJob(String jobName, String groupName, String cronExp, Map<String, Object> param);
  52.     /**
  53.      * 删除任务
  54.      * @param jobName
  55.      * @param groupName
  56.      */
  57.     void deleteJob(String jobName, String groupName);
  59.     /**
  60.      * 启动所有任务
  61.      */
  62.     void startAllJobs();
  64.     /**
  65.      * 暂停所有任务
  66.      */
  67.     void pauseAllJobs();
  69.     /**
  70.      * 恢复所有任务
  71.      */
  72.     void resumeAllJobs();
  74.     /**
  75.      * 关闭所有任务
  76.      */
  77.     void shutdownAllJobs();
  78. }

8、编写 Quartz 服务层接口实现类

  1. package com.mes.material.quartz.service.impl;
  3. import com.mes.material.quartz.service.IQuartzJobService;
  4. import org.quartz.*;
  5. import org.slf4j.Logger;
  6. import org.slf4j.LoggerFactory;
  7. import org.springframework.beans.factory.annotation.Autowired;
  8. import org.springframework.stereotype.Service;
  10. import java.util.Map;
  12. /**
  13.  * @Author: 
  14.  * @Description:
  15.  * @Date Create in 14:15 2022/5/16
  16.  * @Modified By:
  17.  */
  19. @Service
  20. public class IQuartzJobServiceImpl implements IQuartzJobService {
  22.     private static final Logger log = LoggerFactory.getLogger(IQuartzJobServiceImpl.class);
  24.     @Autowired
  25.     private Scheduler scheduler;
  27.     @Override
  28.     public void addJob(String clazzName, String jobName, String groupName, String cronExp, Map<String, Object> param) {
  29.         try {
  30.             // 启动调度器，默认初始化的时候已经启动
  31. //            scheduler.start();
  32.             //构建job信息
  33.             Class<? extends Job> jobClass = (Class<? extends Job>) Class.forName(clazzName);
  34.             JobDetail jobDetail = JobBuilder.newJob(jobClass).withIdentity(jobName, groupName).build();
  35.             //表达式调度构建器(即任务执行的时间)
  36.             CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(cronExp);
  37.             //按新的cronExpression表达式构建一个新的trigger
  38.             CronTrigger trigger = TriggerBuilder.newTrigger().withIdentity(jobName, groupName).withSchedule(scheduleBuilder).build();
  39.             //获得JobDataMap，写入数据
  40.             if (param != null) {
  41.                 trigger.getJobDataMap().putAll(param);
  42.             }
  43.             scheduler.scheduleJob(jobDetail, trigger);
  44.         } catch (Exception e) {
  45.             log.error("创建任务失败", e);
  46.         }
  47.     }
  49.     @Override
  50.     public void pauseJob(String jobName, String groupName) {
  51.         try {
  52.             scheduler.pauseJob(JobKey.jobKey(jobName, groupName));
  53.         } catch (SchedulerException e) {
  54.             log.error("暂停任务失败", e);
  55.         }
  56.     }
  58.     @Override
  59.     public void resumeJob(String jobName, String groupName) {
  60.         try {
  61.             scheduler.resumeJob(JobKey.jobKey(jobName, groupName));
  62.         } catch (SchedulerException e) {
  63.             log.error("恢复任务失败", e);
  64.         }
  65.     }
  67.     @Override
  68.     public void runOnce(String jobName, String groupName) {
  69.         try {
  70.             scheduler.triggerJob(JobKey.jobKey(jobName, groupName));
  71.         } catch (SchedulerException e) {
  72.             log.error("立即运行一次定时任务失败", e);
  73.         }
  74.     }
  76.     @Override
  77.     public void updateJob(String jobName, String groupName, String cronExp, Map<String, Object> param) {
  78.         try {
  79.             TriggerKey triggerKey = TriggerKey.triggerKey(jobName, groupName);
  80.             CronTrigger trigger = (CronTrigger) scheduler.getTrigger(triggerKey);
  81.             if (cronExp != null) {
  82.                 // 表达式调度构建器
  83.                 CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(cronExp);
  84.                 // 按新的cronExpression表达式重新构建trigger
  85.                 trigger = trigger.getTriggerBuilder().withIdentity(triggerKey).withSchedule(scheduleBuilder).build();
  86.             }
  87.             //修改map
  88.             if (param != null) {
  89.                 trigger.getJobDataMap().putAll(param);
  90.             }
  91.             // 按新的trigger重新设置job执行
  92.             scheduler.rescheduleJob(triggerKey, trigger);
  93.         } catch (Exception e) {
  94.             log.error("更新任务失败", e);
  95.         }
  96.     }
  98.     @Override
  99.     public void deleteJob(String jobName, String groupName) {
  100.         try {
  101.             //暂停、移除、删除
  102.             scheduler.pauseTrigger(TriggerKey.triggerKey(jobName, groupName));
  103.             scheduler.unscheduleJob(TriggerKey.triggerKey(jobName, groupName));
  104.             scheduler.deleteJob(JobKey.jobKey(jobName, groupName));
  105.         } catch (Exception e) {
  106.             log.error("删除任务失败", e);
  107.         }
  108.     }
  110.     @Override
  111.     public void startAllJobs() {
  112.         try {
  113.             scheduler.start();
  114.         } catch (Exception e) {
  115.             log.error("开启所有的任务失败", e);
  116.         }
  117.     }
  119.     @Override
  120.     public void pauseAllJobs() {
  121.         try {
  122.             scheduler.pauseAll();
  123.         } catch (Exception e) {
  124.             log.error("暂停所有任务失败", e);
  125.         }
  126.     }
  128.     @Override
  129.     public void resumeAllJobs() {
  130.         try {
  131.             scheduler.resumeAll();
  132.         } catch (Exception e) {
  133.             log.error("恢复所有任务失败", e);
  134.         }
  135.     }
  137.     @Override
  138.     public void shutdownAllJobs() {
  139.         try {
  141.             if (!scheduler.isShutdown()) {
  142.                 // 需谨慎操作关闭scheduler容器
  143.                 // scheduler生命周期结束，无法再 start() 启动scheduler
  144.                 scheduler.shutdown(true);
  145.             }
  146.         } catch (Exception e) {
  147.             log.error("关闭所有的任务失败", e);
  148.         }
  149.     }
  150. }

9、编写 contoller 服务

先编写一个调用实体类

  1. package com.mes.material.quartz.config;
  3. import lombok.Data;
  5. import java.io.Serializable;
  6. import java.util.Map;
  8. /**
  9.  * @Author: 
  10.  * @Description:
  11.  * @Date Create in 14:16 2022/5/16
  12.  * @Modified By:
  13.  */
  14. @Data
  15. public class QuartzConfigDTO implements Serializable {
  16.     private static final long serialVersionUID = 1L;
  17.     /**
  18.      * 任务名称
  19.      */
  20.     private String jobName;
  22.     /**
  23.      * 任务所属组
  24.      */
  25.     private String groupName;
  27.     /**
  28.      * 任务执行类
  29.      */
  30.     private String jobClass;
  32.     /**
  33.      * 任务调度时间表达式
  34.      */
  35.     private String cronExpression;
  37.     /**
  38.      * 附加参数
  39.      */
  40.     private Map<String, Object> param;
  41. }

web服务接口

  1. @RestController
  2. @RequestMapping("/test")
  3. public class TestController {
  5.     private static final Logger log = LoggerFactory.getLogger(TestController.class);
  7.     @Autowired
  8.     private QuartzJobService quartzJobService;
  10.     /**
  11.      * 添加新任务
  12.      * @param configDTO
  13.      * @return
  14.      */
  15.     @RequestMapping("/addJob")
  16.     public Object addJob(@RequestBody QuartzConfigDTO configDTO) {
  17.         quartzJobService.addJob(configDTO.getJobClass(), configDTO.getJobName(), configDTO.getGroupName(), configDTO.getCronExpression(), configDTO.getParam());
  18.         return HttpStatus.OK;
  19.     }
  21.     /**
  22.      * 暂停任务
  23.      * @param configDTO
  24.      * @return
  25.      */
  26.     @RequestMapping("/pauseJob")
  27.     public Object pauseJob(@RequestBody QuartzConfigDTO configDTO) {
  28.         quartzJobService.pauseJob(configDTO.getJobName(), configDTO.getGroupName());
  29.         return HttpStatus.OK;
  30.     }
  32.     /**
  33.      * 恢复任务
  34.      * @param configDTO
  35.      * @return
  36.      */
  37.     @RequestMapping("/resumeJob")
  38.     public Object resumeJob(@RequestBody QuartzConfigDTO configDTO) {
  39.         quartzJobService.resumeJob(configDTO.getJobName(), configDTO.getGroupName());
  40.         return HttpStatus.OK;
  41.     }
  43.     /**
  44.      * 立即运行一次定时任务
  45.      * @param configDTO
  46.      * @return
  47.      */
  48.     @RequestMapping("/runOnce")
  49.     public Object runOnce(@RequestBody QuartzConfigDTO configDTO) {
  50.         quartzJobService.runOnce(configDTO.getJobName(), configDTO.getGroupName());
  51.         return HttpStatus.OK;
  52.     }
  54.     /**
  55.      * 更新任务
  56.      * @param configDTO
  57.      * @return
  58.      */
  59.     @RequestMapping("/updateJob")
  60.     public Object updateJob(@RequestBody QuartzConfigDTO configDTO) {
  61.         quartzJobService.updateJob(configDTO.getJobName(), configDTO.getGroupName(), configDTO.getCronExpression(), configDTO.getParam());
  62.         return HttpStatus.OK;
  63.     }
  65.     /**
  66.      * 删除任务
  67.      * @param configDTO
  68.      * @return
  69.      */
  70.     @RequestMapping("/deleteJob")
  71.     public Object deleteJob(@RequestBody QuartzConfigDTO configDTO) {
  72.         quartzJobService.deleteJob(configDTO.getJobName(), configDTO.getGroupName());
  73.         return HttpStatus.OK;
  74.     }
  76.     /**
  77.      * 启动所有任务
  78.      * @return
  79.      */
  80.     @RequestMapping("/startAllJobs")
  81.     public Object startAllJobs() {
  82.         quartzJobService.startAllJobs();
  83.         return HttpStatus.OK;
  84.     }
  86.     /**
  87.      * 暂停所有任务
  88.      * @return
  89.      */
  90.     @RequestMapping("/pauseAllJobs")
  91.     public Object pauseAllJobs() {
  92.         quartzJobService.pauseAllJobs();
  93.         return HttpStatus.OK;
  94.     }
  96.     /**
  97.      * 恢复所有任务
  98.      * @return
  99.      */
  100.     @RequestMapping("/resumeAllJobs")
  101.     public Object resumeAllJobs() {
  102.         quartzJobService.resumeAllJobs();
  103.         return HttpStatus.OK;
  104.     }
  106.     /**
  107.      * 关闭所有任务
  108.      * @return
  109.      */
  110.     @RequestMapping("/shutdownAllJobs")
  111.     public Object shutdownAllJobs() {
  112.         quartzJobService.shutdownAllJobs();
  113.         return HttpStatus.OK;
  114.     }
  116. }

10、调用quartz

217a2642ac84463c98d9cb4f6675b9ea.png

常量值

e5d687a51df748e397b14ff55203536f.png

11、注册监听器（选用）

  • 创建任务调度监听器

    package com.mes.material.quartz.listener;

    import org.quartz.*;
    import org.quartz.listeners.SchedulerListenerSupport;
    import org.springframework.stereotype.Component;

    /**

    • @Author:
    • @Description: 注册监听器（选用）
    • @Date Create in 14:47 2022/5/16
    • @Modified By:
      */

    @Component
    public class SimpleSchedulerListener extends SchedulerListenerSupport {

    1. @Override
    2. public void jobScheduled(Trigger trigger) {
    3.     System.out.println("任务被部署时被执行");
    4. }
    6. @Override
    7. public void jobUnscheduled(TriggerKey triggerKey) {
    8.     System.out.println("任务被卸载时被执行");
    9. }
    11. @Override
    12. public void triggerFinalized(Trigger trigger) {
    13.     System.out.println("任务完成了它的使命，光荣退休时被执行");
    14. }
    16. @Override
    17. public void triggerPaused(TriggerKey triggerKey) {
    18.     System.out.println(triggerKey + "（一个触发器）被暂停时被执行");
    19. }
    21. @Override
    22. public void triggersPaused(String triggerGroup) {
    23.     System.out.println(triggerGroup + "所在组的全部触发器被停止时被执行");
    24. }
    26. @Override
    27. public void triggerResumed(TriggerKey triggerKey) {
    28.     System.out.println(triggerKey + "（一个触发器）被恢复时被执行");
    29. }
    31. @Override
    32. public void triggersResumed(String triggerGroup) {
    33.     System.out.println(triggerGroup + "所在组的全部触发器被回复时被执行");
    34. }
    36. @Override
    37. public void jobAdded(JobDetail jobDetail) {
    38.     System.out.println("一个JobDetail被动态添加进来");
    39. }
    41. @Override
    42. public void jobDeleted(JobKey jobKey) {
    43.     System.out.println(jobKey + "被删除时被执行");
    44. }
    46. @Override
    47. public void jobPaused(JobKey jobKey) {
    48.     System.out.println(jobKey + "被暂停时被执行");
    49. }
    51. @Override
    52. public void jobsPaused(String jobGroup) {
    53.     System.out.println(jobGroup + "(一组任务）被暂停时被执行");
    54. }
    56. @Override
    57. public void jobResumed(JobKey jobKey) {
    58.     System.out.println(jobKey + "被恢复时被执行");
    59. }
    61. @Override
    62. public void jobsResumed(String jobGroup) {
    63.     System.out.println(jobGroup + "(一组任务）被恢复时被执行");
    64. }
    66. @Override
    67. public void schedulerError(String msg, SchedulerException cause) {
    68.     System.out.println("出现异常" + msg + "时被执行");
    69.     cause.printStackTrace();
    70. }
    72. @Override
    73. public void schedulerInStandbyMode() {
    74.     System.out.println("scheduler被设为standBy等候模式时被执行");
    75. }
    77. @Override
    78. public void schedulerStarted() {
    79.     System.out.println("scheduler启动时被执行");
    80. }
    82. @Override
    83. public void schedulerStarting() {
    84.     System.out.println("scheduler正在启动时被执行");
    85. }
    87. @Override
    88. public void schedulerShutdown() {
    89.     System.out.println("scheduler关闭时被执行");
    90. }
    92. @Override
    93. public void schedulerShuttingdown() {
    94.     System.out.println("scheduler正在关闭时被执行");
    95. }
    97. @Override
    98. public void schedulingDataCleared() {
    99.     System.out.println("scheduler中所有数据包括jobs, triggers和calendars都被清空时被执行");
    100. }

    }

  • 创建任务触发监听器

    package com.mes.material.quartz.listener;

    import org.quartz.JobExecutionContext;
    import org.quartz.Trigger;
    import org.quartz.listeners.TriggerListenerSupport;
    import org.springframework.stereotype.Component;

    /**

    • @Author:
    • @Description: 任务触发监听器
    • @Date Create in 14:48 2022/5/16

    • @Modified By:
      */
      @Component
      public class SimpleTriggerListener extends TriggerListenerSupport {

      /**

      • Trigger监听器的名称

      • @return
        */
        @Override
        public String getName() {
        return “mySimpleTriggerListener”;
        }

        /**

      • Trigger被激发 它关联的job即将被运行
      • @param trigger

      • @param context
        */
        @Override
        public void triggerFired(Trigger trigger, JobExecutionContext context) {
        System.out.println(“myTriggerListener.triggerFired()”);
        }

        /**

      • Trigger被激发 它关联的job即将被运行, TriggerListener 给了一个选择去否决 Job 的执行,如果返回TRUE 那么任务job会被终止
      • @param trigger
      • @param context

      • @return
        */
        @Override
        public boolean vetoJobExecution(Trigger trigger, JobExecutionContext context) {
        System.out.println(“myTriggerListener.vetoJobExecution()”);
        return false;
        }

        /**

      • 当Trigger错过被激发时执行,比如当前时间有很多触发器都需要执行，但是线程池中的有效线程都在工作，
      • 那么有的触发器就有可能超时，错过这一轮的触发。

      • @param trigger
        */
        @Override
        public void triggerMisfired(Trigger trigger) {
        System.out.println(“myTriggerListener.triggerMisfired()”);
        }

        /**

      • 任务完成时触发
      • @param trigger
      • @param context
      • @param triggerInstructionCode
        */
        @Override
        public void triggerComplete(Trigger trigger, JobExecutionContext context, Trigger.CompletedExecutionInstruction triggerInstructionCode) {
        System.out.println(“myTriggerListener.triggerComplete()”);
        }
        }

  • 创建任务执行监听器

    package com.mes.material.quartz.listener;

    import org.quartz.JobExecutionContext;
    import org.quartz.JobExecutionException;
    import org.quartz.listeners.JobListenerSupport;
    import org.springframework.stereotype.Component;

    /**

    • @Author:
    • @Description: 任务执行监听器
    • @Date Create in 14:49 2022/5/16
    • @Modified By:
      */

    @Component
    public class SimpleJobListener extends JobListenerSupport {

  1.     /**
  2.      * job监听器名称
  3.      * @return
  4.      */
  5.     @Override
  6.     public String getName() {
  7.         return "mySimpleJobListener";
  8.     }
  10.     /**
  11.      * 任务被调度前
  12.      * @param context
  13.      */
  14.     @Override
  15.     public void jobToBeExecuted(JobExecutionContext context) {
  16.         System.out.println("simpleJobListener监听器，准备执行："+context.getJobDetail().getKey());
  17.     }
  19.     /**
  20.      * 任务调度被拒了
  21.      * @param context
  22.      */
  23.     @Override
  24.     public void jobExecutionVetoed(JobExecutionContext context) {
  25.         System.out.println("simpleJobListener监听器，取消执行："+context.getJobDetail().getKey());
  26.     }
  28.     /**
  29.      * 任务被调度后
  30.      * @param context
  31.      * @param jobException
  32.      */
  33.     @Override
  34.     public void jobWasExecuted(JobExecutionContext context, JobExecutionException jobException) {
  35.         System.out.println("simpleJobListener监听器，执行结束："+context.getJobDetail().getKey());
  36.     }
  37. }

  • 最后，将监听器注册到Scheduler

    @Autowired
    private SimpleSchedulerListener simpleSchedulerListener;

    @Autowired
    private SimpleJobListener simpleJobListener;

    @Autowired
    private SimpleTriggerListener simpleTriggerListener;

    @Bean(name = “scheduler”)
    public Scheduler scheduler() throws IOException, SchedulerException {

    1. Scheduler scheduler = schedulerFactoryBean().getScheduler();
    2. //全局添加监听器
    3. //添加SchedulerListener监听器
    4. scheduler.getListenerManager().addSchedulerListener(simpleSchedulerListener);
    6. // 添加JobListener, 支持带条件匹配监听器
    7. scheduler.getListenerManager().addJobListener(simpleJobListener, KeyMatcher.keyEquals(JobKey.jobKey("myJob", "myGroup")));
    9. // 添加triggerListener，设置全局监听
    10. scheduler.getListenerManager().addTriggerListener(simpleTriggerListener, EverythingMatcher.allTriggers());
    11. return scheduler;

    }

本文主要围绕springboot + quartz + postgresql实现持久化分布式调度进行介绍，所有的代码功能，都测试过。

发表评论

表情：
评论列表 （有 0 条评论，92人围观）

还没有评论，来说两句吧...

相关阅读