xxl-job任务定时触发流程

xxl-job任务触发流程

xxl-job老版本是依赖quartz的定时任务触发，在v2.1.0版本开始移除quartz依赖：一方面是为了精简系统降低冗余依赖，另一方面是为了提供系统的可控度与稳定性。（本文相应代码版本 2.2.0-SNAPSHOT）

以下是本文的目录大纲：

一.任务触发执行总体流程

　　二.任务定时触发流程

　　三.关于这么设计的感悟

请尊重作者劳动成果，转载请标明原文链接：

https://www.cnblogs.com/wanghongsen/p/12510533.html

一任务触发执行总体流程

先来看下任务触发和执行的完整的任务触发执行总体流程图如下：

上图所示左上角的第一步：任务触发方式主要有以下几种类型：1 根据设置的时间自动触发JobScheduleHelper，2 页面点击操作按钮执行触发，3 父子任务触发，4失败重试触发。

本文重点讲解第一步：任务触发的第一种 1 根据设置的时间自动触发，即上图红色框内标示的部分，具体见JobScheduleHelper这个类。

二任务定时触发流程

详细的JobScheduleHelperCron定时触发这个阶段流程图如下：

具体见JobScheduleHelper这个类结合上面流程图来分析，在工程spring启动的时候触发了JobScheduleHelper类的start（）方法，完整代码如下

  1     public void start(){
  2 
  3         // schedule thread
  4         scheduleThread = new Thread(new Runnable() {
  5             @Override
  6             public void run() {
  7 
  8                 try {
  9                     TimeUnit.MILLISECONDS.sleep(5000 - System.currentTimeMillis()%1000 );
 10                 } catch (InterruptedException e) {
 11                     if (!scheduleThreadToStop) {
 12                         logger.error(e.getMessage(), e);
 13                     }
 14                 }
 15                 logger.info(">>>>>>>>> init xxl-job admin scheduler success.");
 16 
 17                 // pre-read count: treadpool-size * trigger-qps (each trigger cost 50ms, qps = 1000/50 = 20)
 18                 int preReadCount = (XxlJobAdminConfig.getAdminConfig().getTriggerPoolFastMax() + XxlJobAdminConfig.getAdminConfig().getTriggerPoolSlowMax()) * 20;
 19 
 20                 while (!scheduleThreadToStop) {
 21 
 22                     // Scan Job
 23                     long start = System.currentTimeMillis();
 24 
 25                     Connection conn = null;
 26                     Boolean connAutoCommit = null;
 27                     PreparedStatement preparedStatement = null;
 28 
 29                     boolean preReadSuc = true;
 30                     try {
 31 
 32                         conn = XxlJobAdminConfig.getAdminConfig().getDataSource().getConnection();
 33                         connAutoCommit = conn.getAutoCommit();
 34                         conn.setAutoCommit(false);
 35 
 36                         preparedStatement = conn.prepareStatement(  "select * from xxl_job_lock where lock_name = 'schedule_lock' for update" );
 37                         preparedStatement.execute();
 38 
 39                         // tx start
 40 
 41                         // 1、pre read
 42                         long nowTime = System.currentTimeMillis();
 43                         List<XxlJobInfo> scheduleList = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleJobQuery(nowTime + PRE_READ_MS, preReadCount);
 44                         if (scheduleList!=null && scheduleList.size()>0) {
 45                             // 2、push time-ring
 46                             for (XxlJobInfo jobInfo: scheduleList) {
 47 
 48                                 // time-ring jump
 49                                 if (nowTime > jobInfo.getTriggerNextTime() + PRE_READ_MS) {
 50                                     // 2.1、trigger-expire > 5s：pass && make next-trigger-time
 51                                     logger.warn(">>>>>>>>>>> xxl-job, schedule misfire, jobId = " + jobInfo.getId());
 52 
 53                                     // fresh next
 54                                     refreshNextValidTime(jobInfo, new Date());
 55 
 56                                 } else if (nowTime > jobInfo.getTriggerNextTime()) {
 57                                     // 2.2、trigger-expire < 5s：direct-trigger && make next-trigger-time
 58 
 59                                     // 1、trigger
 60                                     JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.CRON, -1, null, null);
 61                                     logger.debug(">>>>>>>>>>> xxl-job, schedule push trigger : jobId = " + jobInfo.getId() );
 62 
 63                                     // 2、fresh next
 64                                     refreshNextValidTime(jobInfo, new Date());
 65 
 66                                     // next-trigger-time in 5s, pre-read again
 67                                     if (jobInfo.getTriggerStatus()==1 && nowTime + PRE_READ_MS > jobInfo.getTriggerNextTime()) {
 68 
 69                                         // 1、make ring second
 70                                         int ringSecond = (int)((jobInfo.getTriggerNextTime()/1000)%60);
 71 
 72                                         // 2、push time ring
 73                                         pushTimeRing(ringSecond, jobInfo.getId());
 74 
 75                                         // 3、fresh next
 76                                         refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));
 77 
 78                                     }
 79 
 80                                 } else {
 81                                     // 2.3、trigger-pre-read：time-ring trigger && make next-trigger-time
 82 
 83                                     // 1、make ring second
 84                                     int ringSecond = (int)((jobInfo.getTriggerNextTime()/1000)%60);
 85 
 86                                     // 2、push time ring
 87                                     pushTimeRing(ringSecond, jobInfo.getId());
 88 
 89                                     // 3、fresh next
 90                                     refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));
 91 
 92                                 }
 93 
 94                             }
 95 
 96                             // 3、update trigger info
 97                             for (XxlJobInfo jobInfo: scheduleList) {
 98                                 XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleUpdate(jobInfo);
 99                             }
100 
101                         } else {
102                             preReadSuc = false;
103                         }
104 
105                         // tx stop
106 
107 
108                     } catch (Exception e) {
109                         if (!scheduleThreadToStop) {
110                             logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread error:{}", e);
111                         }
112                     } finally {
113 
114                         // commit
115                         if (conn != null) {
116                             try {
117                                 conn.commit();
118                             } catch (SQLException e) {
119                                 if (!scheduleThreadToStop) {
120                                     logger.error(e.getMessage(), e);
121                                 }
122                             }
123                             try {
124                                 conn.setAutoCommit(connAutoCommit);
125                             } catch (SQLException e) {
126                                 if (!scheduleThreadToStop) {
127                                     logger.error(e.getMessage(), e);
128                                 }
129                             }
130                             try {
131                                 conn.close();
132                             } catch (SQLException e) {
133                                 if (!scheduleThreadToStop) {
134                                     logger.error(e.getMessage(), e);
135                                 }
136                             }
137                         }
138 
139                         // close PreparedStatement
140                         if (null != preparedStatement) {
141                             try {
142                                 preparedStatement.close();
143                             } catch (SQLException e) {
144                                 if (!scheduleThreadToStop) {
145                                     logger.error(e.getMessage(), e);
146                                 }
147                             }
148                         }
149                     }
150                     long cost = System.currentTimeMillis()-start;
151 
152 
153                     // Wait seconds, align second
154                     if (cost < 1000) {  // scan-overtime, not wait
155                         try {
156                             // pre-read period: success > scan each second; fail > skip this period;
157                             TimeUnit.MILLISECONDS.sleep((preReadSuc?1000:PRE_READ_MS) - System.currentTimeMillis()%1000);
158                         } catch (InterruptedException e) {
159                             if (!scheduleThreadToStop) {
160                                 logger.error(e.getMessage(), e);
161                             }
162                         }
163                     }
164 
165                 }
166 
167                 logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread stop");
168             }
169         });
170         scheduleThread.setDaemon(true);
171         scheduleThread.setName("xxl-job, admin JobScheduleHelper#scheduleThread");
172         scheduleThread.start();
173 
174 
175         // ring thread
176         ringThread = new Thread(new Runnable() {
177             @Override
178             public void run() {
179 
180                 // align second
181                 try {
182                     TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis()%1000 );
183                 } catch (InterruptedException e) {
184                     if (!ringThreadToStop) {
185                         logger.error(e.getMessage(), e);
186                     }
187                 }
188 
189                 while (!ringThreadToStop) {
190 
191                     try {
192                         // second data
193                         List<Integer> ringItemData = new ArrayList<>();
194                         int nowSecond = Calendar.getInstance().get(Calendar.SECOND);   // 避免处理耗时太长，跨过刻度，向前校验一个刻度；
195                         for (int i = 0; i < 2; i++) {
196                             List<Integer> tmpData = ringData.remove( (nowSecond+60-i)%60 );
197                             if (tmpData != null) {
198                                 ringItemData.addAll(tmpData);
199                             }
200                         }
201 
202                         // ring trigger
203                         logger.debug(">>>>>>>>>>> xxl-job, time-ring beat : " + nowSecond + " = " + Arrays.asList(ringItemData) );
204                         if (ringItemData.size() > 0) {
205                             // do trigger
206                             for (int jobId: ringItemData) {
207                                 // do trigger
208                                 JobTriggerPoolHelper.trigger(jobId, TriggerTypeEnum.CRON, -1, null, null);
209                             }
210                             // clear
211                             ringItemData.clear();
212                         }
213                     } catch (Exception e) {
214                         if (!ringThreadToStop) {
215                             logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread error:{}", e);
216                         }
217                     }
218 
219                     // next second, align second
220                     try {
221                         TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis()%1000);
222                     } catch (InterruptedException e) {
223                         if (!ringThreadToStop) {
224                             logger.error(e.getMessage(), e);
225                         }
226                     }
227                 }
228                 logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread stop");
229             }
230         });
231         ringThread.setDaemon(true);
232         ringThread.setName("xxl-job, admin JobScheduleHelper#ringThread");
233         ringThread.start();
234     }

View Code

任务定时触发，流程如下：

1 分布式锁

为了保证分布式一致性先上悲观锁：使用select xx for update来实现

1  conn = XxlJobAdminConfig.getAdminConfig().getDataSource().getConnection();
2   connAutoCommit = conn.getAutoCommit();
3   conn.setAutoCommit(false);
4   preparedStatement = conn.prepareStatement(  "select * from xxl_job_lock where lock_name = 'schedule_lock' for update" );
5   preparedStatement.execute();

2 轮询db，找出trigger_next_time（下次触发时间）在距now 5秒内的任务

 1 List<XxlJobInfo> scheduleList = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleJobQuery(nowTime + PRE_READ_MS, preReadCount);
 2 详细sql如下：
 3 trigger_status代表触发状态处于启动的任务 trigger_next_time代表 任务下次 执行触发的时间
 4 <select id="scheduleJobQuery" parameterType="java.util.HashMap" resultMap="XxlJobInfo">
 5    SELECT *
 6    FROM xxl_job_info AS t
 7    WHERE t.trigger_status = 1
 8       and t.trigger_next_time <![CDATA[ <= ]]> #{maxNextTime}
 9    ORDER BY id ASC
10    LIMIT #{pagesize}
11 </select>

3 触发算法

拿到了距now 5秒内的任务列表数据：scheduleList，分三种情况处理：for循环遍历scheduleList集合

（1）对到达now时间后的任务：（超出now 5秒外）：直接跳过不执行；重置trigger_next_time；

（2）对到达now时间后的任务：（超出now 5秒内）：线程执行触发逻辑；若任务下一次触发时间是在5秒内，则放到时间轮内（Map<Integer, List<Integer>> 秒数(1-60) => 任务id列表）；

再重置trigger_next_time

（3）对未到达now时间的任务：直接放到时间轮内；重置trigger_next_time 。

分别对应下面这个数轴的三个阶段

具体参见下面代码：

1   下面对应代码（1）对到达now时间后的任务（超出now 5秒外）：直接跳过不执行； 重置trigger_next_time
2 if (nowTime > jobInfo.getTriggerNextTime() + 5000) {
3 // 2.1、trigger-expire > 5s：pass && make next-trigger-time
4 logger.warn(">>>>>>>>>>> xxl-job, schedule misfire, jobId = " + jobInfo.getId());
5 // fresh next
6 refreshNextValidTime(jobInfo, new Date());
7 }

 1 下面对应代码（2）对到达now时间后的任务（超出now 5秒内）：线程执行触发逻辑； 若任务下一次触发时间是在5秒内，
 2 
 3 则放到时间轮内（Map<Integer, List<Integer>> 秒数(1-60) => 任务id列表）；重置trigger_next_time
 4 
 5 else if (nowTime > jobInfo.getTriggerNextTime()) {
 6         // 1、trigger
 7        JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.CRON, -1, null, null);
 8         // 2、fresh next
 9         refreshNextValidTime(jobInfo, new Date());
10         // next-trigger-time in 5s, pre-read again
11         if (jobInfo.getTriggerStatus()==1 && nowTime + PRE_READ_MS > jobInfo.getTriggerNextTime()) {
12         // 1、make ring second
13         int ringSecond = (int)((jobInfo.getTriggerNextTime()/1000)%60);
14         // 2、push time ring
15         pushTimeRing(ringSecond, jobInfo.getId());
16          // 3、fresh next
17         refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));
18         }
19 }

 1 下面对应代码（3）对未到达now时间的任务：直接放到时间轮内；重置trigger_next_time
 2 
 3 else {
 4     // 1、make ring second
 5     int ringSecond = (int)((jobInfo.getTriggerNextTime()/1000)%60);
 6     // 2、push time ring
 7     pushTimeRing(ringSecond, jobInfo.getId());
 8     // 3、fresh next
 9     refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));
10 }

上面的refreshNextValidTime方法是更新任务的 trigger_next_time 下次触发时间，xxl_job_info表是记录定时任务的db表，里面有个trigger_next_time（Long）字段，表示下一次触发的时间点任务时间被修改

每一次任务触发后，可以根据cronb表达式计算下一次触发时间戳：Date nextValidTime = new CronExpression(jobInfo.getJobCron()).getNextValidTimeAfter(new Date())），更新trigger_next_time字段。

4 时间轮触发

接下来讲时间轮，时间轮数据结构： Map<Integer, List<Integer>> key是秒数(1-60) value是任务id列表，具体结构如下图：

时间轮的执行代码如下：

 1 public void run() {
 2 
 3                 // align second
 4                 try {
 5                     TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis()%1000 );
 6                 } catch (InterruptedException e) {
 7                     if (!ringThreadToStop) {
 8                         logger.error(e.getMessage(), e);
 9                     }
10                 }
11 
12                 while (!ringThreadToStop) {
13 
14                     try {
15                         // second data
16                         List<Integer> ringItemData = new ArrayList<>();
17                         int nowSecond = Calendar.getInstance().get(Calendar.SECOND);   // 避免处理耗时太长，跨过刻度，向前校验一个刻度；
18                         for (int i = 0; i < 2; i++) {
19                             List<Integer> tmpData = ringData.remove( (nowSecond+60-i)%60 );
20                             if (tmpData != null) {
21                                 ringItemData.addAll(tmpData);
22                             }
23                         }
24 
25                         // ring trigger
26                         logger.debug(">>>>>>>>>>> xxl-job, time-ring beat : " + nowSecond + " = " + Arrays.asList(ringItemData) );
27                         if (ringItemData.size() > 0) {
28                             // do trigger
29                             for (int jobId: ringItemData) {
30                                 // do trigger
31                                 JobTriggerPoolHelper.trigger(jobId, TriggerTypeEnum.CRON, -1, null, null);
32                             }
33                             // clear
34                             ringItemData.clear();
35                         }
36                     } catch (Exception e) {
37                         if (!ringThreadToStop) {
38                             logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread error:{}", e);
39                         }
40                     }
41 
42                     // next second, align second
43                     try {
44                         TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis()%1000);
45                     } catch (InterruptedException e) {
46                         if (!ringThreadToStop) {
47                             logger.error(e.getMessage(), e);
48                         }
49                     }
50                 }
51                 logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread stop");
52             }

时间轮数据结构： Map<Integer, List<Integer>> key是hash计算触发时间获得的秒数(1-60)，value是任务id列表

入轮：扫描任务触发时（1）本次任务处理完成，但下一次触发时间是在5秒内（2）本次任务未达到触发时间

出轮：获取当前时间秒数，从时间轮内移出当前秒数前2个秒数的任务id列表，依次进行触发任务；（避免处理耗时太长，跨过刻度，多向前校验一秒）

增加时间轮的目的是：任务过多可能会延迟，为了保障触发时间尽可能和任务设置的触发时间尽量一致，把即将要触发的任务提前放到时间轮里，每秒来触发时间轮相应节点的任务

三关于这么设计的感悟：看似简单的一个任务触发为什么要搞这么复杂呢？

我的答案是: 因为出于“性能” 和“时效性”这两点综合来考虑，即“中庸之道”。

就拿每次 “从DB查出近期即将要到触发时间任务” 这个场景来看：

1 如果希望“性能”更好，那肯定每次多查出些数据，但这样就不可避免的造成因为任务过多，同一批查出来的位置靠后的某些任务触发就可能会延迟，比如实际触发比设定触发的时间晚几秒。

2 如果希望“时效性”更好，那肯定每次少查出些数据，比如每次只查出来一条或者几条，实际触发时间和设定的触发时间基本一样，但这样造成了频繁查询数据库，性能下降。

故通过“时间轮”来达到既“性能”比较好并且每次查出相对尽量多的数据（目前是取5s内触发的任务），又通过时间轮来保障“时效性”：实际触发时间和设定的触发时间尽量一样。这就是设计这么复杂的原因。