spring + quartz 之動態任務調度

時間 2019-11-30

原文原文鏈接

一、任務調度：根據特定的時間規則自動執行相應的業務邏輯，quartz算是一種比較簡單靈活的實現方式，可是不支持分佈式.java

二、maven依賴：　web

1 <dependency>
2             <groupId>org.quartz-scheduler</groupId>
3             <artifactId>quartz</artifactId>
4             <version>2.2.1</version>
5 </dependency>

View Code

三、實現：spring

　　一、建立任務常量類：json

 1 package com.quartz;
 2 
 3 /**
 4  * Created by xiao on 2015/9/28.
 5  */
 6 public final class JobConstants {
 7 
 8 
 9     /**
10      * Job狀態
11      */
12     public static class JobStatus{
13         /**
14          * 未處理
15          */
16         public static final String JOB_UNPROCESSED = "0";
17 
18         /**
19          * 執行成功
20          */
21         public static final String JOB_SUCCESSED = "1";
22 
23         /**
24          * 執行失敗
25          */
26         public static final String JOB_FAILED = "-1";
27 
28         /**
29          * 過時
30          */
31         public static final String JOB_EXPIRED = "-2";
32     }
33 
34     /**
35      * 任務組
36      */
37     public static final class JobGruop{
38 
39         /**
40          * 默認的任務組
41          */
42         public static final String JOB_DEFAULT_GROUP_ = "job_default_group_id";
43 
44         /**
45          * 默認的任務組
46          */
47         public static final String JOB_DEFAULT_GROUP_NAME = "job_default_group_name";
48     }
49 
50     public static final class JobName{
51 
52         /**
53          * 默認的任務名
54          */
55         public static final String JOB_DEFAULT_ID = "job_default_id";
56 
57         /**
58          * 默認的任務名
59          */
60         public static final String JOB_DEFAULT_NAME = "job_default_name";
61     }
62 
63 }

View Code

　　二、建立任務實體，封裝任務參數：　併發

  1 package com.quartz;
  2 
  3 import java.util.Date;
  4 
  5 /**
  6  * 任務 實體
  7  * Created by xiao on 2015/9/28.
  8  */
  9 public final class Job implements java.io.Serializable {
 10 
 11     /**
 12      * 任務ID
 13      */
 14     private String jobId;
 15 
 16     /**
 17      * 任務名稱
 18      */
 19     private String jobName;
 20 
 21     /**
 22      * 任務組id
 23      */
 24     private String jobGroupId;
 25     /**
 26      * 任務分組
 27      */
 28     private String jobGroup;
 29 
 30     /**
 31      * 建立時間
 32      */
 33     private Date createTime;
 34 
 35     /**
 36      * 更新時間
 37      */
 38     private Date updateTime;
 39 
 40     /**
 41      * 任務狀態
 42      */
 43     private String jobStatus;
 44     /**
 45      * cron表達式
 46      */
 47     private String cronExpression;
 48 
 49     /**
 50      * 簡單時間表達式
 51      */
 52     private Date simpleExpression;
 53 
 54     /**
 55      * 任務執行時調用哪一個類的方法 包名+類名
 56      */
 57     private String jobClassName;
 58 
 59     /**
 60      * 任務調用的方法名
 61      */
 62     private String jobMethodName;
 63 
 64     /**
 65      * 任務是否有狀態(即併發執行同一個任務)
 66      */
 67     private boolean isConcurrent = true;
 68 
 69     /**
 70      * 描述
 71      */
 72     private String description;
 73 
 74 
 75     public String getJobId() {
 76         return jobId;
 77     }
 78 
 79     public void setJobId(String jobId) {
 80         this.jobId = jobId;
 81     }
 82 
 83     public String getJobName() {
 84         return jobName;
 85     }
 86 
 87     public void setJobName(String jobName) {
 88         this.jobName = jobName;
 89     }
 90 
 91     public String getJobGroupId() {
 92         return jobGroupId;
 93     }
 94 
 95     public void setJobGroupId(String jobGroupId) {
 96         this.jobGroupId = jobGroupId;
 97     }
 98 
 99     public String getJobGroup() {
100         return jobGroup;
101     }
102 
103     public void setJobGroup(String jobGroup) {
104         this.jobGroup = jobGroup;
105     }
106 
107     public Date getCreateTime() {
108         return createTime;
109     }
110 
111     public void setCreateTime(Date createTime) {
112         this.createTime = createTime;
113     }
114 
115     public Date getUpdateTime() {
116         return updateTime;
117     }
118 
119     public void setUpdateTime(Date updateTime) {
120         this.updateTime = updateTime;
121     }
122 
123     public String getJobStatus() {
124         return jobStatus;
125     }
126 
127     public void setJobStatus(String jobStatus) {
128         this.jobStatus = jobStatus;
129     }
130 
131     public String getCronExpression() {
132         return cronExpression;
133     }
134 
135     public void setCronExpression(String cronExpression) {
136         this.cronExpression = cronExpression;
137     }
138 
139     public Date getSimpleExpression() {
140         return simpleExpression;
141     }
142 
143     public void setSimpleExpression(Date simpleExpression) {
144         this.simpleExpression = simpleExpression;
145     }
146 
147     public String getJobClassName() {
148         return jobClassName;
149     }
150 
151     public void setJobClassName(String jobClassName) {
152         this.jobClassName = jobClassName;
153     }
154 
155     public String getJobMethodName() {
156         return jobMethodName;
157     }
158 
159     public void setJobMethodName(String jobMethodName) {
160         this.jobMethodName = jobMethodName;
161     }
162 
163     public boolean isConcurrent() {
164         return isConcurrent;
165     }
166 
167     public void setConcurrent(boolean concurrent) {
168         isConcurrent = concurrent;
169     }
170 
171     public String getDescription() {
172         return description;
173     }
174 
175     public void setDescription(String description) {
176         this.description = description;
177     }
178 }

View Code

　　三、建立任務管理器，實現了任務動態的增、刪、更新、暫停等操做　　　maven

  1 package com.quartz;
  2 
  3 import com.quartz.job.DefaultJobFactory;
  4 import com.quartz.job.DisallowConcurrentJobFactory;
  5 import org.quartz.*;
  6 import org.quartz.impl.StdScheduler;
  7 import org.quartz.impl.matchers.GroupMatcher;
  8 import util.FastJsonUtil;
  9 import util.StringUtil;
 10 
 11 import java.util.ArrayList;
 12 import java.util.List;
 13 import java.util.Set;
 14 
 15 /**
 16  * 任務調度管理器， 實現任務的動態操做
 17  * Created by xiao on 2015/9/28.
 18  */
 19 public class JobManager {
 20 
 21     //爲調度管理器注入工廠bean
 22     private StdScheduler scheduler;
 23 
 24     //調度名稱
 25     private static final String SCHEDULER_NAME = "scheduler";
 26 
 27     public StdScheduler getScheduler() {
 28         return scheduler;
 29     }
 30 
 31     public void setScheduler(StdScheduler scheduler) {
 32         this.scheduler = scheduler;
 33     }
 34 
 35     /**
 36      * 添加任務
 37      * @param job
 38      * @throws SchedulerException
 39      */
 40     public void addJob(Job job) throws SchedulerException, ClassNotFoundException {
 41         if (job == null || StringUtil.isEmptyString(job.getJobId())) return;
 42         if(StringUtil.isEmptyString(job.getCronExpression())
 43                 && null == job.getSimpleExpression()) return;
 44         if (StringUtil.isEmptyString(job.getJobClassName())) return;
 45         if(StringUtil.isEmptyString(job.getJobName()))
 46             job.setJobName(JobConstants.JobName.JOB_DEFAULT_NAME);
 47         if(null == job.getSimpleExpression()) {
 48             addCronJob(job);
 49         }else{
 50             addSimpleJob(job);
 51         }
 52     }
 53 
 54     /**
 55      * 添加 cron 表達式任務
 56      * @param job
 57      * @throws SchedulerException
 58      * @throws ClassNotFoundException
 59      */
 60     private void addCronJob(Job job) throws SchedulerException, ClassNotFoundException {
 61         //根據任務id和任務組Id建立觸發器key
 62         TriggerKey triggerKey = TriggerKey.triggerKey(job.getJobId(), job.getJobGroup());
 63         //獲取觸發器對象
 64         CronTrigger trigger = (CronTrigger) scheduler.getTrigger(triggerKey);
 65         // 不存在，建立一個
 66         if (null == trigger) {
 67             JobDetail jobDetail = JobBuilder.newJob(job.isConcurrent() ? DefaultJobFactory.class : DisallowConcurrentJobFactory.class)
 68                     .withIdentity(job.getJobId(), job.getJobGroup()).build();
 69             jobDetail.getJobDataMap().put(SCHEDULER_NAME, FastJsonUtil.objToJson(job));
 70             trigger = TriggerBuilder.newTrigger()
 71                     .withIdentity(triggerKey)
 72                     .withSchedule(CronScheduleBuilder.cronSchedule(
 73                             job.getCronExpression())).build();
 74             scheduler.scheduleJob(jobDetail, trigger);
 75         } else {
 76             updateJobCron(job);
 77         }
 78     }
 79 
 80     /**
 81      * 添加 簡單時間 表達式任務
 82      * @param job
 83      * @throws SchedulerException
 84      * @throws ClassNotFoundException
 85      */
 86     private void addSimpleJob(Job job) throws SchedulerException {
 87         //根據任務id和任務組Id建立觸發器key
 88         TriggerKey triggerKey = TriggerKey.triggerKey(job.getJobId(), job.getJobGroup());
 89         //獲取觸發器對象
 90         SimpleTrigger trigger = (SimpleTrigger) scheduler.getTrigger(triggerKey);
 91         // 不存在，建立一個
 92         if (null == trigger) {
 93             JobDetail jobDetail = JobBuilder.newJob(job.isConcurrent() ? DefaultJobFactory.class : DisallowConcurrentJobFactory.class)
 94                     .withIdentity(job.getJobId(), job.getJobGroup()).build();
 95             jobDetail.getJobDataMap().put(SCHEDULER_NAME, FastJsonUtil.objToJson(job));
 96             trigger = TriggerBuilder.newTrigger()
 97                     .withIdentity(triggerKey)
 98                     .withSchedule(SimpleScheduleBuilder.simpleSchedule())
 99                     .startAt(job.getSimpleExpression()).build();
100             scheduler.scheduleJob(jobDetail, trigger);
101         } else {
102             updateJobSimple(job);
103         }
104     }
105 
106     /**
107      * 更新job時間表達式
108      *
109      * @param job
110      * @throws SchedulerException
111      */
112     public void updateJobCron(Job job) throws SchedulerException {
113         TriggerKey triggerKey = TriggerKey.triggerKey(
114                 job.getJobId(), job.getJobGroup());
115         CronTrigger trigger = (CronTrigger) scheduler.getTrigger(triggerKey);
116         trigger = trigger.getTriggerBuilder().withIdentity(triggerKey)
117                 .withSchedule(CronScheduleBuilder.cronSchedule(job.getCronExpression())).build();
118         scheduler.rescheduleJob(triggerKey, trigger);
119     }
120 
121     /**
122      * 更新job時間表達式
123      *
124      * @param job
125      * @throws SchedulerException
126      */
127     public void updateJobSimple(Job job) throws SchedulerException {
128         TriggerKey triggerKey = TriggerKey.triggerKey(
129                 job.getJobId(), job.getJobGroup());
130         SimpleTrigger trigger = (SimpleTrigger) scheduler.getTrigger(triggerKey);
131         trigger = trigger.getTriggerBuilder().withIdentity(triggerKey)
132                 .withSchedule(SimpleScheduleBuilder.simpleSchedule())
133                 .startAt(job.getSimpleExpression()).build();
134         scheduler.rescheduleJob(triggerKey, trigger);
135     }
136 
137     /** 獲取全部計劃中的任務列表
138      * @return
139      * @throws SchedulerException
140      **/
141     public List<Job> getAllJob() throws SchedulerException {
142         GroupMatcher<JobKey> matcher = GroupMatcher.anyJobGroup();
143         Set<JobKey> jobKeys = scheduler.getJobKeys(matcher);
144         List<Job> jobList = new ArrayList<>();
145         for (JobKey jobKey : jobKeys) {
146             List<? extends Trigger> triggers = scheduler.getTriggersOfJob(jobKey);
147             for (Trigger trigger : triggers) {
148                 Job job = new Job();
149                 job.setJobId(jobKey.getName());
150                 job.setJobGroup(jobKey.getGroup());
151                 Trigger.TriggerState triggerState = scheduler.getTriggerState(trigger.getKey());
152                 job.setJobStatus(triggerState.name());
153                 if (trigger instanceof CronTrigger) {
154                     CronTrigger cronTrigger = (CronTrigger) trigger;
155                     String cronExpression = cronTrigger.getCronExpression();
156                     job.setCronExpression(cronExpression);
157                 }
158                 jobList.add(job);
159             }
160         }
161         return jobList;
162     }
163 
164     /**
165      * 全部正在運行的job
166      *
167      * @return
168      * @throws SchedulerException
169      */
170     public List<Job> getRunningJob() throws SchedulerException {
171         List<JobExecutionContext> executingJobs = scheduler.getCurrentlyExecutingJobs();
172         List<Job> jobList = new ArrayList<>(executingJobs.size());
173         for (JobExecutionContext executingJob : executingJobs) {
174             Job job = new Job();
175             JobDetail jobDetail = executingJob.getJobDetail();
176             JobKey jobKey = jobDetail.getKey();
177             Trigger trigger = executingJob.getTrigger();
178             job.setJobName(jobKey.getName());
179             job.setJobGroup(jobKey.getGroup());
180             Trigger.TriggerState triggerState = scheduler.getTriggerState(trigger.getKey());
181             job.setJobStatus(triggerState.name());
182             if (trigger instanceof CronTrigger) {
183                 CronTrigger cronTrigger = (CronTrigger) trigger;
184                 String cronExpression = cronTrigger.getCronExpression();
185                 job.setCronExpression(cronExpression);
186             }
187             jobList.add(job);
188         }
189         return jobList;
190     }
191 
192     /**
193      * 暫停一個job
194      *
195      * @param scheduleJob
196      * @throws SchedulerException
197      */
198     public void pauseJob(Job scheduleJob) throws SchedulerException {
199         JobKey jobKey = JobKey.jobKey(scheduleJob.getJobId(), scheduleJob.getJobGroup());
200         scheduler.pauseJob(jobKey);
201     }
202 
203     /**
204      * 恢復一個job
205      *
206      * @param job
207      * @throws SchedulerException
208      */
209     public void resumeJob(Job job) throws SchedulerException {
210         JobKey jobKey = JobKey.jobKey(job.getJobId(), job.getJobGroup());
211         scheduler.resumeJob(jobKey);
212     }
213 
214     /**
215      * 刪除一個job
216      *
217      * @param scheduleJob
218      * @throws SchedulerException
219      */
220     public void deleteJob(Job scheduleJob) throws SchedulerException {
221         JobKey jobKey = JobKey.jobKey(scheduleJob.getJobId(), scheduleJob.getJobGroup());
222         scheduler.deleteJob(jobKey);
223 
224     }
225 
226     /**
227      * 當即執行job
228      *
229      * @param job
230      * @throws SchedulerException
231      */
232     public void triggerJob(Job job) throws SchedulerException {
233         JobKey jobKey = JobKey.jobKey(job.getJobId(), job.getJobGroup());
234         scheduler.triggerJob(jobKey);
235     }
236 
237 
238 }

View Code

　　四、建立任務執行類，這個類須要實現 org.quarzt.Job 接口, Job接口有一個excute(JobExecutionContext context) 方法, 調度被處罰之後會自動執行, 再次咱們實現這個方法, 經過反射來轉發到咱們建立任務時指定執行的方法, 代碼以下：分佈式

 1 package com.quartz.job;
 2 
 3 import com.quartz.Job;
 4 import org.quartz.JobExecutionContext;
 5 import org.quartz.JobExecutionException;
 6 import util.FastJsonUtil;
 7 import util.StringUtil;
 8 
 9 /**
10  * Created by xiao on 2015/9/29.
11  */
12 public class DefaultJobFactory extends  AbstractJobFactory implements org.quartz.Job {
13 
14     public void execute(JobExecutionContext context) throws JobExecutionException {
15         String scheduleJob = (String) context.getMergedJobDataMap().get("scheduler");
16         if(!StringUtil.isEmptyString(scheduleJob)){
17             Job job = FastJsonUtil.jsonToObj(scheduleJob, Job.class);
18             invoke(job);
19         }
20 
21     }
22 }

View Code

 1 package com.quartz.job;
 2 
 3 import org.quartz.DisallowConcurrentExecution;
 4 import org.quartz.Job;
 5 import org.quartz.JobExecutionContext;
 6 import org.quartz.JobExecutionException;
 7 import util.FastJsonUtil;
 8 import util.StringUtil;
 9 
10 /**
11  * Created by xiao on 2015/9/29.
12  */
13 @DisallowConcurrentExecution
14 public class DisallowConcurrentJobFactory extends  AbstractJobFactory implements Job {
15 
16     public void execute(JobExecutionContext context) throws JobExecutionException {
17 
18         String scheduleJob = (String) context.getMergedJobDataMap().get("scheduler");
19         if(!StringUtil.isEmptyString(scheduleJob)){
20             com.quartz.Job job = FastJsonUtil.jsonToObj(scheduleJob, com.quartz.Job.class);
21             invoke(job);
22         }
23     }
24 }

View Code

 1 package com.quartz.job;
 2 
 3 import com.quartz.Job;
 4 import util.SpringBeanUtil;
 5 import util.StringUtil;
 6 
 7 import java.lang.reflect.InvocationTargetException;
 8 import java.lang.reflect.Method;
 9 
10 /**
11  * Created by xiao on 2016/4/29.
12  */
13 public class AbstractJobFactory {
14 
15     public void invoke(Job job){
16         if(!StringUtil.isEmptyString(job.getJobClassName())){
17             try {
18                 Class<?> clazz = Class.forName(job.getJobClassName());
19                 //從spring容器獲取bean,不然沒法注入
20                 Object obj = SpringBeanUtil.getBeanByType(clazz);
21                 //反射方法
22                 Method method = obj.getClass().getDeclaredMethod(job.getJobMethodName());
23                 method.invoke(obj, job);
24             } catch (ClassNotFoundException e) {
25                 e.printStackTrace();
26             } catch (NoSuchMethodException e) {
27                 e.printStackTrace();
28             } catch (IllegalAccessException e) {
29                 e.printStackTrace();
30             } catch (InvocationTargetException e) {
31                 e.printStackTrace();
32             }
33         }
34     }
35 }

View Code

　　獲取bean 工具類：ide

 1 package util;
 2 
 3 import org.springframework.web.context.ContextLoader;
 4 import org.springframework.web.context.WebApplicationContext;
 5 
 6 public final class SpringBeanUtil {
 7 
 8     //獲取web容器上下文
 9     private static WebApplicationContext wac = ContextLoader
10             .getCurrentWebApplicationContext();
11 
12     /**
13      * 違背了 Spring 依賴注入思想
14      *
15      * @param beanId
16      * @return
17      */
18     public static Object getBeanByName(String beanId) throws Exception {
19         if (StringUtil.isEmptyString(beanId)) {
20             throw new Exception("beanId is null");
21         }
22         return wac.getBean(beanId);
23 
24     }
25 
26     /**
27      *  違背spring的ioc解耦思想。
28      */
29     public static  <T> T getBeanByType(Class clazz) {
30         if (clazz == null) {
31             return null;
32         }
33         return (T) wac.getBean(clazz);
34     }
35 }

View Code

　　FastJson 序列化工具類, FastJson是阿里開源的號稱最快的序列化工具工具

 1 package util;
 2 
 3 import com.alibaba.fastjson.JSON;
 4 import com.alibaba.fastjson.serializer.SerializerFeature;
 5 
 6 /**
 7  * FastJson 序列化工具
 8  *
 9  * @author xiao
10  *
11  */
12 public final class FastJsonUtil {
13 
14     /**
15      * 序列化參數
16      */
17     private static final SerializerFeature[] features = {
18             SerializerFeature.WriteMapNullValue,
19             SerializerFeature.WriteNullBooleanAsFalse,
20             SerializerFeature.WriteNullStringAsEmpty,
21             SerializerFeature.WriteNullListAsEmpty,
22             SerializerFeature.WriteNullNumberAsZero };
23 
24     /**
25      * 對象轉換成json 支持list,map,array
26      *
27      * @param obj
28      * @return
29      */
30     public static String objToJson(Object obj) {
31         return JSON.toJSONString(obj, features);
32     }
33 
34 
35     /**
36      * json 轉換成對象
37      *
38      * @param json
39      * @param clazz
40      * @return
41      */
42     public static <T> T jsonToObj(String json, Class<?> clazz) {
43         return (T) JSON.parseObject(json, clazz);
44     }
45 }

View Code

　　五、最後一步, 在spring中集成,配置以下：ui

1     <!-- 任務管理器 -->
2     <bean class="com.quartz.JobManager">
3         <property name="scheduler">
4         <!-- 將觸發器注入任務工程 -->
5             <bean id="scheduler" lazy-init="false" scope="singleton"
6                   class="org.springframework.scheduling.quartz.SchedulerFactoryBean"/>
7         </property>
8     </bean>

View Code