Bladeren bron

调度组件销毁流程优化,先停止调度线程,然后等待时间轮内存量任务处理完成,最终销毁时间轮线程;

xuxueli 6 jaren geleden
bovenliggende
commit
293ffca14d

+ 1 - 0
doc/XXL-JOB官方文档.md Bestand weergeven

1477
 - 11、升级xxl-rpc至较新版本,修复代理服务初始化时远程服务不可用导致长连冗余创建的问题;
1477
 - 11、升级xxl-rpc至较新版本,修复代理服务初始化时远程服务不可用导致长连冗余创建的问题;
1478
 - 12、首页调度报表的日期排序在TIDB下乱序问题修复;
1478
 - 12、首页调度报表的日期排序在TIDB下乱序问题修复;
1479
 - 13、调度中心与执行器双向通讯超时时间调整为3s;
1479
 - 13、调度中心与执行器双向通讯超时时间调整为3s;
1480
+- 14、调度组件销毁流程优化,先停止调度线程,然后等待时间轮内存量任务处理完成,最终销毁时间轮线程;
1480
 
1481
 
1481
 
1482
 
1482
 ### 6.26 版本 v2.1.1 Release Notes[规划中]
1483
 ### 6.26 版本 v2.1.1 Release Notes[规划中]

+ 55 - 16
xxl-job-admin/src/main/java/com/xxl/job/admin/core/thread/JobScheduleHelper.java Bestand weergeven

27
 
27
 
28
     private Thread scheduleThread;
28
     private Thread scheduleThread;
29
     private Thread ringThread;
29
     private Thread ringThread;
30
-    private volatile boolean toStop = false;
30
+    private volatile boolean scheduleThreadToStop = false;
31
+    private volatile boolean ringThreadToStop = false;
31
     private volatile static Map<Integer, List<Integer>> ringData = new ConcurrentHashMap<>();
32
     private volatile static Map<Integer, List<Integer>> ringData = new ConcurrentHashMap<>();
32
 
33
 
33
     public void start(){
34
     public void start(){
40
                 try {
41
                 try {
41
                     TimeUnit.MILLISECONDS.sleep(5000 - System.currentTimeMillis()%1000 );
42
                     TimeUnit.MILLISECONDS.sleep(5000 - System.currentTimeMillis()%1000 );
42
                 } catch (InterruptedException e) {
43
                 } catch (InterruptedException e) {
43
-                    if (!toStop) {
44
+                    if (!scheduleThreadToStop) {
44
                         logger.error(e.getMessage(), e);
45
                         logger.error(e.getMessage(), e);
45
                     }
46
                     }
46
                 }
47
                 }
47
                 logger.info(">>>>>>>>> init xxl-job admin scheduler success.");
48
                 logger.info(">>>>>>>>> init xxl-job admin scheduler success.");
48
 
49
 
49
-                while (!toStop) {
50
+                while (!scheduleThreadToStop) {
50
 
51
 
51
                     // 扫描任务
52
                     // 扫描任务
52
                     long start = System.currentTimeMillis();
53
                     long start = System.currentTimeMillis();
127
 
128
 
128
                         conn.commit();
129
                         conn.commit();
129
                     } catch (Exception e) {
130
                     } catch (Exception e) {
130
-                        if (!toStop) {
131
+                        if (!scheduleThreadToStop) {
131
                             logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread error:{}", e);
132
                             logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread error:{}", e);
132
                         }
133
                         }
133
                     } finally {
134
                     } finally {
152
                             TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis()%1000);
153
                             TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis()%1000);
153
                         }
154
                         }
154
                     } catch (InterruptedException e) {
155
                     } catch (InterruptedException e) {
155
-                        if (!toStop) {
156
+                        if (!scheduleThreadToStop) {
156
                             logger.error(e.getMessage(), e);
157
                             logger.error(e.getMessage(), e);
157
                         }
158
                         }
158
                     }
159
                     }
175
                 try {
176
                 try {
176
                     TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis()%1000 );
177
                     TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis()%1000 );
177
                 } catch (InterruptedException e) {
178
                 } catch (InterruptedException e) {
178
-                    if (!toStop) {
179
+                    if (!ringThreadToStop) {
179
                         logger.error(e.getMessage(), e);
180
                         logger.error(e.getMessage(), e);
180
                     }
181
                     }
181
                 }
182
                 }
182
 
183
 
183
                 int lastSecond = -1;
184
                 int lastSecond = -1;
184
-                while (!toStop) {
185
+                while (!ringThreadToStop) {
185
 
186
 
186
                     try {
187
                     try {
187
                         // second data
188
                         // second data
216
                             ringItemData.clear();
217
                             ringItemData.clear();
217
                         }
218
                         }
218
                     } catch (Exception e) {
219
                     } catch (Exception e) {
219
-                        if (!toStop) {
220
+                        if (!ringThreadToStop) {
220
                             logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread error:{}", e);
221
                             logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread error:{}", e);
221
                         }
222
                         }
222
                     }
223
                     }
225
                     try {
226
                     try {
226
                         TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis()%1000);
227
                         TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis()%1000);
227
                     } catch (InterruptedException e) {
228
                     } catch (InterruptedException e) {
228
-                        if (!toStop) {
229
+                        if (!ringThreadToStop) {
229
                             logger.error(e.getMessage(), e);
230
                             logger.error(e.getMessage(), e);
230
                         }
231
                         }
231
                     }
232
                     }
239
     }
240
     }
240
 
241
 
241
     public void toStop(){
242
     public void toStop(){
242
-        toStop = true;
243
 
243
 
244
-        // interrupt and wait
245
-        scheduleThread.interrupt();
244
+        // 1、stop schedule
245
+        scheduleThreadToStop = true;
246
         try {
246
         try {
247
-            scheduleThread.join();
247
+            TimeUnit.SECONDS.sleep(1);  // wait
248
         } catch (InterruptedException e) {
248
         } catch (InterruptedException e) {
249
             logger.error(e.getMessage(), e);
249
             logger.error(e.getMessage(), e);
250
         }
250
         }
251
+        if (scheduleThread.getState() != Thread.State.TERMINATED){
252
+            // interrupt and wait
253
+            scheduleThread.interrupt();
254
+            try {
255
+                scheduleThread.join();
256
+            } catch (InterruptedException e) {
257
+                logger.error(e.getMessage(), e);
258
+            }
259
+        }
251
 
260
 
252
-        // interrupt and wait
253
-        ringThread.interrupt();
261
+        // if has ring data
262
+        boolean hasRingData = false;
263
+        if (!ringData.isEmpty()) {
264
+            for (int second : ringData.keySet()) {
265
+                List<Integer> tmpData = ringData.get(second);
266
+                if (tmpData!=null && tmpData.size()>0) {
267
+                    hasRingData = true;
268
+                    break;
269
+                }
270
+            }
271
+        }
272
+        if (hasRingData) {
273
+            try {
274
+                TimeUnit.SECONDS.sleep(8);
275
+            } catch (InterruptedException e) {
276
+                logger.error(e.getMessage(), e);
277
+            }
278
+        }
279
+
280
+        // stop ring (wait job-in-memory stop)
281
+        ringThreadToStop = true;
254
         try {
282
         try {
255
-            ringThread.join();
283
+            TimeUnit.SECONDS.sleep(1);
256
         } catch (InterruptedException e) {
284
         } catch (InterruptedException e) {
257
             logger.error(e.getMessage(), e);
285
             logger.error(e.getMessage(), e);
258
         }
286
         }
287
+        if (ringThread.getState() != Thread.State.TERMINATED){
288
+            // interrupt and wait
289
+            ringThread.interrupt();
290
+            try {
291
+                ringThread.join();
292
+            } catch (InterruptedException e) {
293
+                logger.error(e.getMessage(), e);
294
+            }
295
+        }
296
+
297
+        logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper stop");
259
     }
298
     }
260
 
299
 
261
 }
300
 }