Browse Source

修复任务监控线程被耗时任务阻塞的问题

xuxueli 8 năm trước cách đây
mục cha
commit
ed31b1b66a

+ 2 - 1
doc/XXL-JOB官方文档.md

@@ -571,7 +571,7 @@ XXL-JOB首先定制了Quartz原生表结构前缀(XXL_JOB_QRTZ_)。
 #### 5.3.2 系统组成
 - **调度模块(调度中心)**:
     负责管理调度信息,按照调度配置发出调度请求,自身不承担业务代码。调度系统与任务解耦,提高了系统可用性和稳定性,同时调度系统性能不再受限于任务模块;
-    支持可视化、简单且动态的管理调度信息,包括任务新建,更新,删除,GLUE开发和任务报警等,所有上述操作都会实时生效,同时支持监控调度结果以及执行日志,支持执行器Failover。
+    支持可视化、简单且动态的管理调度信息,包括任务新建,更新,删除,GLUE开发和任务报警等,所有上述操作都会实时生效,同时支持监控调度结果以及执行日志,支持执行器Failover。
 - **执行模块(执行器)**:
     负责接收调度请求并执行任务逻辑。任务模块专注于任务的执行等操作,开发和维护更加简单和高效;
     接收“调度中心”的执行请求、终止请求和日志请求等。
@@ -1020,6 +1020,7 @@ Tips: 历史版本(V1.3.x)目前已经Release至稳定版本, 进入维护阶段
 
 ### 6.20 版本 V1.9.0 特性[迭代中]
 - 1、新增任务运行模式 "GLUE模式(NodeJS) ",支持NodeJS脚本任务;
+- 2、修复任务监控线程被耗时任务阻塞的问题;
 
 ### TODO LIST
 - 1、任务权限管理:执行器为粒度分配权限,核心操作校验权限;

+ 17 - 11
xxl-job-admin/src/main/java/com/xxl/job/admin/core/thread/JobFailMonitorHelper.java

@@ -6,6 +6,7 @@ import com.xxl.job.admin.core.model.XxlJobLog;
 import com.xxl.job.admin.core.schedule.XxlJobDynamicScheduler;
 import com.xxl.job.admin.core.util.MailUtil;
 import com.xxl.job.core.biz.model.ReturnT;
+import org.apache.commons.collections.CollectionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -35,33 +36,38 @@ public class JobFailMonitorHelper {
 
 			@Override
 			public void run() {
-
 				// monitor
 				while (!toStop) {
 					try {
-						Integer jobLogId = JobFailMonitorHelper.instance.queue.take();
-						if (jobLogId != null && jobLogId > 0) {
-							XxlJobLog log = XxlJobDynamicScheduler.xxlJobLogDao.load(jobLogId);
-							if (log!=null) {
-								if (ReturnT.SUCCESS_CODE==log.getTriggerCode() && log.getHandleCode()==0) {
-									// job running, wait + again monitor
-									TimeUnit.SECONDS.sleep(10);
+						List<Integer> jobLogIdList = new ArrayList<Integer>();
+						int drainToNum = JobFailMonitorHelper.instance.queue.drainTo(jobLogIdList);
 
+						if (CollectionUtils.isNotEmpty(jobLogIdList)) {
+							for (Integer jobLogId : jobLogIdList) {
+								if (jobLogId==null || jobLogId==0) {
+									continue;
+								}
+								XxlJobLog log = XxlJobDynamicScheduler.xxlJobLogDao.load(jobLogId);
+								if (log == null) {
+									continue;
+								}
+								if (ReturnT.SUCCESS_CODE == log.getTriggerCode() && log.getHandleCode() == 0) {
 									JobFailMonitorHelper.monitor(jobLogId);
 									logger.info(">>>>>>>>>>> job monitor, job running, JobLogId:{}", jobLogId);
 								}
-								if (ReturnT.SUCCESS_CODE==log.getTriggerCode() && ReturnT.SUCCESS_CODE==log.getHandleCode()) {
+								if (ReturnT.SUCCESS_CODE == log.getTriggerCode() && ReturnT.SUCCESS_CODE == log.getHandleCode()) {
 									// job success, pass
 									logger.info(">>>>>>>>>>> job monitor, job success, JobLogId:{}", jobLogId);
 								}
-
-								if (ReturnT.FAIL_CODE == log.getTriggerCode()|| ReturnT.FAIL_CODE==log.getHandleCode()) {
+								if (ReturnT.FAIL_CODE == log.getTriggerCode() || ReturnT.FAIL_CODE == log.getHandleCode()) {
 									// job fail,
 									sendMonitorEmail(log);
 									logger.info(">>>>>>>>>>> job monitor, job fail, JobLogId:{}", jobLogId);
 								}
 							}
 						}
+
+						TimeUnit.SECONDS.sleep(10);
 					} catch (Exception e) {
 						logger.error("job monitor error:{}", e);
 					}