From 4d94321beb53c57abab784046bf9e71cb5e7361f Mon Sep 17 00:00:00 2001 From: xuxueli <931591021@qq.com> Date: Sun, 14 Jun 2026 01:11:37 +0800 Subject: [PATCH] =?UTF-8?q?refactor:=E8=B0=83=E5=BA=A6=E7=BB=84=E4=BB=B6?= =?UTF-8?q?=E5=AE=88=E6=8A=A4=E7=BA=BF=E7=A8=8B=E4=BB=A3=E7=A0=81=E9=87=8D?= =?UTF-8?q?=E6=9E=84=EF=BC=8C=E6=8F=90=E5=8D=87=E7=A8=B3=E5=AE=9A=E6=80=A7?= =?UTF-8?q?=E4=BB=A5=E5=8F=8A=E5=8F=AF=E7=BB=B4=E6=8A=A4=E6=80=A7=EF=BC=9B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- doc/XXL-JOB官方文档.md | 5 +- .../scheduler/thread/JobCompleteHelper.java | 93 +++----- .../thread/JobFailAlarmMonitorHelper.java | 103 ++++----- .../scheduler/thread/JobLogReportHelper.java | 198 ++++++++---------- .../scheduler/thread/JobRegistryHelper.java | 130 +++++------- .../scheduler/thread/JobScheduleHelper.java | 11 +- 6 files changed, 214 insertions(+), 326 deletions(-) diff --git a/doc/XXL-JOB官方文档.md b/doc/XXL-JOB官方文档.md index 6957a44a..6648a7f8 100644 --- a/doc/XXL-JOB官方文档.md +++ b/doc/XXL-JOB官方文档.md @@ -2846,8 +2846,9 @@ alter table xxl_job_log - 3、【优化】任务参数长度调整，最长支持2048字符； - 4、【升级】调度中心UI交互优化，任务及日志管理支持下拉框模糊搜索，提升交互体验； - 5、【修复】XxlJobFileAppender自定义地址callbackLogPath设置无效问题修复；合并ISSUS-3963； -- 6、【TODO】调度中心OpenAPI完善，提供任务管理能力；封装Agent Skill并推送ClawHub； -- 7、【TODO】AccessToken升级：执行器维度隔离，支持线上化配置；升级双端OpenApi，适配AccessToken升级； +- 6、【优化】调度组件守护线程代码重构，提升稳定性以及可维护性； +- 7、【TODO】调度中心OpenAPI完善，提供任务管理能力；封装Agent Skill并推送ClawHub； +- 8、【TODO】AccessToken升级：执行器维度隔离，支持线上化配置；升级双端OpenApi，适配AccessToken升级； ### TODO LIST diff --git a/xxl-job-admin/src/main/java/com/xxl/job/admin/business/scheduler/thread/JobCompleteHelper.java b/xxl-job-admin/src/main/java/com/xxl/job/admin/business/scheduler/thread/JobCompleteHelper.java index a22d6f5e..4cdbd340 100644 --- a/xxl-job-admin/src/main/java/com/xxl/job/admin/business/scheduler/thread/JobCompleteHelper.java +++ b/xxl-job-admin/src/main/java/com/xxl/job/admin/business/scheduler/thread/JobCompleteHelper.java @@ -3,8 +3,9 @@ package com.xxl.job.admin.business.scheduler.thread; import com.xxl.job.admin.business.model.XxlJobLog; import com.xxl.job.admin.business.scheduler.config.XxlJobAdminBootstrap; import com.xxl.job.admin.framework.util.I18nUtil; -import com.xxl.job.core.openapi.model.CallbackRequest; import com.xxl.job.core.context.XxlJobContext; +import com.xxl.job.core.openapi.model.CallbackRequest; +import com.xxl.tool.concurrent.CyclicThread; import com.xxl.tool.core.DateTool; import com.xxl.tool.response.Response; import org.slf4j.Logger; @@ -25,15 +26,14 @@ public class JobCompleteHelper { // ---------------------- monitor ---------------------- private ThreadPoolExecutor callbackThreadPool = null; - private Thread monitorThread; - private volatile boolean toStop = false; + private CyclicThread jobMonitorThread; /** * start */ public void start(){ - // for callback + // 1、callbackThreadPool callbackThreadPool = new ThreadPoolExecutor( 2, 20, @@ -55,88 +55,55 @@ public class JobCompleteHelper { }); - // for monitor - monitorThread = new Thread(new Runnable() { - + // 2、jobMonitorThread + jobMonitorThread = new CyclicThread("JobCompleteHelper#jobMonitorThread", true, new Runnable() { @Override public void run() { + // 任务结果丢失处理：调度记录停留在 "运行中" 状态超过10min，且对应执行器心跳注册失败不在线，则将本地调度主动标记失败； + Date losedTime = DateTool.addMinutes(new Date(), -10); + List losedJobIds = XxlJobAdminBootstrap.getInstance().getXxlJobLogMapper().findLostJobIds(losedTime); - // wait for JobTriggerPoolHelper-init - try { - TimeUnit.MILLISECONDS.sleep(50); - } catch (Throwable e) { - if (!toStop) { - logger.error(e.getMessage(), e); - } - } + if (losedJobIds!=null && losedJobIds.size()>0) { + for (Long logId: losedJobIds) { - // monitor - while (!toStop) { - try { - // 任务结果丢失处理：调度记录停留在 "运行中" 状态超过10min，且对应执行器心跳注册失败不在线，则将本地调度主动标记失败； - Date losedTime = DateTool.addMinutes(new Date(), -10); - List losedJobIds = XxlJobAdminBootstrap.getInstance().getXxlJobLogMapper().findLostJobIds(losedTime); + XxlJobLog jobLog = new XxlJobLog(); + jobLog.setId(logId); - if (losedJobIds!=null && losedJobIds.size()>0) { - for (Long logId: losedJobIds) { + jobLog.setHandleTime(new Date()); + jobLog.setHandleCode(XxlJobContext.HANDLE_CODE_FAIL); + jobLog.setHandleMsg( I18nUtil.getString("joblog_lost_fail") ); - XxlJobLog jobLog = new XxlJobLog(); - jobLog.setId(logId); - - jobLog.setHandleTime(new Date()); - jobLog.setHandleCode(XxlJobContext.HANDLE_CODE_FAIL); - jobLog.setHandleMsg( I18nUtil.getString("joblog_lost_fail") ); - - XxlJobAdminBootstrap.getInstance().getJobCompleter().complete(jobLog); - } - - } - } catch (Throwable e) { - if (!toStop) { - logger.error(">>>>>>>>>>> xxl-job, job fail monitor thread error:{}", e); - } + XxlJobAdminBootstrap.getInstance().getJobCompleter().complete(jobLog); } - try { - TimeUnit.SECONDS.sleep(60); - } catch (Throwable e) { - if (!toStop) { - logger.error(e.getMessage(), e); - } - } - - } - - logger.info(">>>>>>>>>>> xxl-job, JobLosedMonitorHelper stop"); - + } } - }); - monitorThread.setDaemon(true); - monitorThread.setName("xxl-job, admin JobLosedMonitorHelper"); - monitorThread.start(); + }, 60 * 1000L, true); + jobMonitorThread.start(); + } /** * stop */ public void stop(){ - toStop = true; - // stop registryOrRemoveThreadPool + // 1、callbackThreadPool callbackThreadPool.shutdownNow(); - // stop monitorThread (interrupt and wait) - monitorThread.interrupt(); - try { - monitorThread.join(); - } catch (Throwable e) { - logger.error(e.getMessage(), e); - } + // 2、jobMonitorThread + jobMonitorThread.stop(); } // ---------------------- helper ---------------------- + /** + * callback + * + * @param callbackParamList callback param + * @return callback result + */ public Response callback(List callbackParamList) { callbackThreadPool.execute(new Runnable() { diff --git a/xxl-job-admin/src/main/java/com/xxl/job/admin/business/scheduler/thread/JobFailAlarmMonitorHelper.java b/xxl-job-admin/src/main/java/com/xxl/job/admin/business/scheduler/thread/JobFailAlarmMonitorHelper.java index 3280428f..34161b75 100644 --- a/xxl-job-admin/src/main/java/com/xxl/job/admin/business/scheduler/thread/JobFailAlarmMonitorHelper.java +++ b/xxl-job-admin/src/main/java/com/xxl/job/admin/business/scheduler/thread/JobFailAlarmMonitorHelper.java @@ -5,11 +5,11 @@ import com.xxl.job.admin.business.model.XxlJobLog; import com.xxl.job.admin.business.scheduler.config.XxlJobAdminBootstrap; import com.xxl.job.admin.business.scheduler.trigger.TriggerTypeEnum; import com.xxl.job.admin.framework.util.I18nUtil; +import com.xxl.tool.concurrent.CyclicThread; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.List; -import java.util.concurrent.TimeUnit; /** * job fail-monitor helper @@ -22,77 +22,53 @@ public class JobFailAlarmMonitorHelper { // ---------------------- monitor ---------------------- - private Thread monitorThread; - private volatile boolean toStop = false; + /** + * monitor thread + */ + private CyclicThread monitorThread; /** * start */ public void start(){ - monitorThread = new Thread(new Runnable() { + monitorThread = new CyclicThread("JobFailAlarmMonitorHelper#monitorThread", true, new Runnable() { @Override public void run() { - - // monitor - while (!toStop) { - try { - - List failLogIds = XxlJobAdminBootstrap.getInstance().getXxlJobLogMapper().findFailJobLogIds(1000); - if (failLogIds!=null && !failLogIds.isEmpty()) { - for (long failLogId: failLogIds) { - - // lock log - int lockRet = XxlJobAdminBootstrap.getInstance().getXxlJobLogMapper().updateAlarmStatus(failLogId, 0, -1); - if (lockRet < 1) { - continue; - } - XxlJobLog log = XxlJobAdminBootstrap.getInstance().getXxlJobLogMapper().load(failLogId); - XxlJobInfo info = XxlJobAdminBootstrap.getInstance().getXxlJobInfoMapper().loadById(log.getJobId()); - - // 1、fail retry monitor - if (log.getExecutorFailRetryCount() > 0) { - XxlJobAdminBootstrap.getInstance().getJobTriggerPoolHelper().trigger(log.getJobId(), TriggerTypeEnum.RETRY, (log.getExecutorFailRetryCount()-1), log.getExecutorShardingParam(), log.getExecutorParam(), null); - String retryMsg = "

>>>>>>>>>>>"+ I18nUtil.getString("jobconf_trigger_type_retry") +"<<<<<<<<<<<
"; - log.setTriggerMsg(log.getTriggerMsg() + retryMsg); - XxlJobAdminBootstrap.getInstance().getXxlJobLogMapper().updateTriggerInfo(log); - } - - // 2、fail alarm monitor - int newAlarmStatus = 0; // 告警状态：0-默认、-1=锁定状态、1-无需告警、2-告警成功、3-告警失败 - if (info != null) { - boolean alarmResult = XxlJobAdminBootstrap.getInstance().getJobAlarmer().alarm(info, log); - newAlarmStatus = alarmResult?2:3; - } else { - newAlarmStatus = 1; - } - - XxlJobAdminBootstrap.getInstance().getXxlJobLogMapper().updateAlarmStatus(failLogId, -1, newAlarmStatus); - } + List failLogIds = XxlJobAdminBootstrap.getInstance().getXxlJobLogMapper().findFailJobLogIds(1000); + if (failLogIds!=null && !failLogIds.isEmpty()) { + for (long failLogId: failLogIds) { + + // lock log + int lockRet = XxlJobAdminBootstrap.getInstance().getXxlJobLogMapper().updateAlarmStatus(failLogId, 0, -1); + if (lockRet < 1) { + continue; } - - } catch (Throwable e) { - if (!toStop) { - logger.error(">>>>>>>>>>> xxl-job, job fail monitor thread error:{}", e.getMessage(), e); + XxlJobLog log = XxlJobAdminBootstrap.getInstance().getXxlJobLogMapper().load(failLogId); + XxlJobInfo info = XxlJobAdminBootstrap.getInstance().getXxlJobInfoMapper().loadById(log.getJobId()); + + // 1、fail retry monitor + if (log.getExecutorFailRetryCount() > 0) { + XxlJobAdminBootstrap.getInstance().getJobTriggerPoolHelper().trigger(log.getJobId(), TriggerTypeEnum.RETRY, (log.getExecutorFailRetryCount()-1), log.getExecutorShardingParam(), log.getExecutorParam(), null); + String retryMsg = "

>>>>>>>>>>>"+ I18nUtil.getString("jobconf_trigger_type_retry") +"<<<<<<<<<<<
"; + log.setTriggerMsg(log.getTriggerMsg() + retryMsg); + XxlJobAdminBootstrap.getInstance().getXxlJobLogMapper().updateTriggerInfo(log); } - } - - try { - TimeUnit.SECONDS.sleep(10); - } catch (Throwable e) { - if (!toStop) { - logger.error(e.getMessage(), e); - } - } - } - - logger.info(">>>>>>>>>>> xxl-job, job fail monitor thread stop"); + // 2、fail alarm monitor + int newAlarmStatus = 0; // 告警状态：0-默认、-1=锁定状态、1-无需告警、2-告警成功、3-告警失败 + if (info != null) { + boolean alarmResult = XxlJobAdminBootstrap.getInstance().getJobAlarmer().alarm(info, log); + newAlarmStatus = alarmResult?2:3; + } else { + newAlarmStatus = 1; + } + XxlJobAdminBootstrap.getInstance().getXxlJobLogMapper().updateAlarmStatus(failLogId, -1, newAlarmStatus); + } + } } - }); - monitorThread.setDaemon(true); - monitorThread.setName("xxl-job, admin JobFailMonitorHelper"); + }, 10 * 1000L, true); monitorThread.start(); } @@ -100,14 +76,7 @@ public class JobFailAlarmMonitorHelper { * stop */ public void stop(){ - toStop = true; - // interrupt and wait - monitorThread.interrupt(); - try { - monitorThread.join(); - } catch (Throwable e) { - logger.error(e.getMessage(), e); - } + monitorThread.stop(); } } diff --git a/xxl-job-admin/src/main/java/com/xxl/job/admin/business/scheduler/thread/JobLogReportHelper.java b/xxl-job-admin/src/main/java/com/xxl/job/admin/business/scheduler/thread/JobLogReportHelper.java index e1bd8cec..181a70a1 100644 --- a/xxl-job-admin/src/main/java/com/xxl/job/admin/business/scheduler/thread/JobLogReportHelper.java +++ b/xxl-job-admin/src/main/java/com/xxl/job/admin/business/scheduler/thread/JobLogReportHelper.java @@ -1,7 +1,8 @@ package com.xxl.job.admin.business.scheduler.thread; -import com.xxl.job.admin.business.scheduler.config.XxlJobAdminBootstrap; import com.xxl.job.admin.business.model.XxlJobLogReport; +import com.xxl.job.admin.business.scheduler.config.XxlJobAdminBootstrap; +import com.xxl.tool.concurrent.CyclicThread; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -9,7 +10,7 @@ import java.util.Calendar; import java.util.Date; import java.util.List; import java.util.Map; -import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; /** * job log report helper @@ -19,144 +20,109 @@ import java.util.concurrent.TimeUnit; public class JobLogReportHelper { private static final Logger logger = LoggerFactory.getLogger(JobLogReportHelper.class); - - private Thread logReportThread; - private volatile boolean toStop = false; + private CyclicThread logReportThread; + private AtomicLong lastCleanLogTime; /** * start */ public void start(){ - logReportThread = new Thread(new Runnable() { + /** + * last clean log time （ Thread-safe concurrent reading and writing ） + */ + lastCleanLogTime = new AtomicLong(0); + + // log report thread + logReportThread = new CyclicThread("JobLogReportHelper#logReportThread", true, new Runnable() { @Override public void run() { - // last clean log time - long lastCleanLogTime = 0; - - - while (!toStop) { - - // 1、log-report refresh: refresh log report in 3 days - try { - - for (int i = 0; i < 3; i++) { - - // today - Calendar itemDay = Calendar.getInstance(); - itemDay.add(Calendar.DAY_OF_MONTH, -i); - itemDay.set(Calendar.HOUR_OF_DAY, 0); - itemDay.set(Calendar.MINUTE, 0); - itemDay.set(Calendar.SECOND, 0); - itemDay.set(Calendar.MILLISECOND, 0); - - Date todayFrom = itemDay.getTime(); - - itemDay.set(Calendar.HOUR_OF_DAY, 23); - itemDay.set(Calendar.MINUTE, 59); - itemDay.set(Calendar.SECOND, 59); - itemDay.set(Calendar.MILLISECOND, 999); - - Date todayTo = itemDay.getTime(); - - // refresh log-report every minute - XxlJobLogReport xxlJobLogReport = new XxlJobLogReport(); - xxlJobLogReport.setTriggerDay(todayFrom); - xxlJobLogReport.setRunningCount(0); - xxlJobLogReport.setSucCount(0); - xxlJobLogReport.setFailCount(0); - xxlJobLogReport.setUpdateTime(new Date()); - - // fill count-data - Map triggerCountMap = XxlJobAdminBootstrap.getInstance().getXxlJobLogMapper().findLogReport(todayFrom, todayTo); - if (triggerCountMap!=null && !triggerCountMap.isEmpty()) { - int triggerDayCount = triggerCountMap.containsKey("triggerDayCount")?Integer.parseInt(String.valueOf(triggerCountMap.get("triggerDayCount"))):0; - int triggerDayCountRunning = triggerCountMap.containsKey("triggerDayCountRunning")?Integer.parseInt(String.valueOf(triggerCountMap.get("triggerDayCountRunning"))):0; - int triggerDayCountSuc = triggerCountMap.containsKey("triggerDayCountSuc")?Integer.parseInt(String.valueOf(triggerCountMap.get("triggerDayCountSuc"))):0; - int triggerDayCountFail = triggerDayCount - triggerDayCountRunning - triggerDayCountSuc; - - xxlJobLogReport.setRunningCount(triggerDayCountRunning); - xxlJobLogReport.setSucCount(triggerDayCountSuc); - xxlJobLogReport.setFailCount(triggerDayCountFail); - } - - // do refresh: - XxlJobAdminBootstrap.getInstance().getXxlJobLogReportMapper().saveOrUpdate(xxlJobLogReport); // 0-fail; 1-save suc; 2-update suc; - /*if (ret < 1) { - XxlJobAdminBootstrap.getInstance().getXxlJobLogReportMapper().save(xxlJobLogReport); - }*/ - } - - } catch (Throwable e) { - if (!toStop) { - logger.error(">>>>>>>>>>> xxl-job, JobLogReportHelper(log-report refresh) error:{}", e.getMessage(), e); - } + // 1、log-report refresh: refresh log report in 3 days + for (int i = 0; i < 3; i++) { + + // today + Calendar itemDay = Calendar.getInstance(); + itemDay.add(Calendar.DAY_OF_MONTH, -i); + itemDay.set(Calendar.HOUR_OF_DAY, 0); + itemDay.set(Calendar.MINUTE, 0); + itemDay.set(Calendar.SECOND, 0); + itemDay.set(Calendar.MILLISECOND, 0); + + Date todayFrom = itemDay.getTime(); + + itemDay.set(Calendar.HOUR_OF_DAY, 23); + itemDay.set(Calendar.MINUTE, 59); + itemDay.set(Calendar.SECOND, 59); + itemDay.set(Calendar.MILLISECOND, 999); + + Date todayTo = itemDay.getTime(); + + // refresh log-report every minute + XxlJobLogReport xxlJobLogReport = new XxlJobLogReport(); + xxlJobLogReport.setTriggerDay(todayFrom); + xxlJobLogReport.setRunningCount(0); + xxlJobLogReport.setSucCount(0); + xxlJobLogReport.setFailCount(0); + xxlJobLogReport.setUpdateTime(new Date()); + + // fill count-data + Map triggerCountMap = XxlJobAdminBootstrap.getInstance().getXxlJobLogMapper().findLogReport(todayFrom, todayTo); + if (triggerCountMap!=null && !triggerCountMap.isEmpty()) { + int triggerDayCount = triggerCountMap.containsKey("triggerDayCount")?Integer.parseInt(String.valueOf(triggerCountMap.get("triggerDayCount"))):0; + int triggerDayCountRunning = triggerCountMap.containsKey("triggerDayCountRunning")?Integer.parseInt(String.valueOf(triggerCountMap.get("triggerDayCountRunning"))):0; + int triggerDayCountSuc = triggerCountMap.containsKey("triggerDayCountSuc")?Integer.parseInt(String.valueOf(triggerCountMap.get("triggerDayCountSuc"))):0; + int triggerDayCountFail = triggerDayCount - triggerDayCountRunning - triggerDayCountSuc; + + xxlJobLogReport.setRunningCount(triggerDayCountRunning); + xxlJobLogReport.setSucCount(triggerDayCountSuc); + xxlJobLogReport.setFailCount(triggerDayCountFail); } - // 2、log-clean: switch open & once each day - try { - if (XxlJobAdminBootstrap.getInstance().getLogretentiondays()>0 - && System.currentTimeMillis() - lastCleanLogTime > 24*60*60*1000) { - - // expire-time - Calendar expiredDay = Calendar.getInstance(); - expiredDay.add(Calendar.DAY_OF_MONTH, -1 * XxlJobAdminBootstrap.getInstance().getLogretentiondays()); - expiredDay.set(Calendar.HOUR_OF_DAY, 0); - expiredDay.set(Calendar.MINUTE, 0); - expiredDay.set(Calendar.SECOND, 0); - expiredDay.set(Calendar.MILLISECOND, 0); - Date clearBeforeTime = expiredDay.getTime(); - - // clean expired log - List logIds = null; - do { - logIds = XxlJobAdminBootstrap.getInstance().getXxlJobLogMapper().findClearLogIds(0, 0, clearBeforeTime, 0, 1000); - if (logIds!=null && !logIds.isEmpty()) { - XxlJobAdminBootstrap.getInstance().getXxlJobLogMapper().clearLog(logIds); - } - } while (logIds!=null && !logIds.isEmpty()); - - // update clean time - lastCleanLogTime = System.currentTimeMillis(); - } - } catch (Throwable e) { - if (!toStop) { - logger.error(">>>>>>>>>>> xxl-job, JobLogReportHelper(log-clean) error:{}", e.getMessage(), e); - } - } + // do refresh: + XxlJobAdminBootstrap.getInstance().getXxlJobLogReportMapper().saveOrUpdate(xxlJobLogReport); // 0-fail; 1-save suc; 2-update suc; + /*if (ret < 1) { + XxlJobAdminBootstrap.getInstance().getXxlJobLogReportMapper().save(xxlJobLogReport); + }*/ + } - try { - TimeUnit.MINUTES.sleep(1); - } catch (Throwable e) { - if (!toStop) { - logger.error(e.getMessage(), e); + // 2、log-clean: switch open & once each day + if (XxlJobAdminBootstrap.getInstance().getLogretentiondays()>0 + && System.currentTimeMillis() - lastCleanLogTime.longValue() > 24*60*60*1000) { + + // expire-time + Calendar expiredDay = Calendar.getInstance(); + expiredDay.add(Calendar.DAY_OF_MONTH, -1 * XxlJobAdminBootstrap.getInstance().getLogretentiondays()); + expiredDay.set(Calendar.HOUR_OF_DAY, 0); + expiredDay.set(Calendar.MINUTE, 0); + expiredDay.set(Calendar.SECOND, 0); + expiredDay.set(Calendar.MILLISECOND, 0); + Date clearBeforeTime = expiredDay.getTime(); + + // clean expired log + List logIds = null; + do { + logIds = XxlJobAdminBootstrap.getInstance().getXxlJobLogMapper().findClearLogIds(0, 0, clearBeforeTime, 0, 1000); + if (logIds!=null && !logIds.isEmpty()) { + XxlJobAdminBootstrap.getInstance().getXxlJobLogMapper().clearLog(logIds); } - } + } while (logIds!=null && !logIds.isEmpty()); + // update clean time + lastCleanLogTime.set(System.currentTimeMillis()); } - logger.info(">>>>>>>>>>> xxl-job, job log report thread stop"); - } - }); - logReportThread.setDaemon(true); - logReportThread.setName("xxl-job, admin JobLogReportHelper"); + }, 60 * 1000L, true); logReportThread.start(); + } /** * stop */ public void stop(){ - toStop = true; - // interrupt and wait - logReportThread.interrupt(); - try { - logReportThread.join(); - } catch (Throwable e) { - logger.error(e.getMessage(), e); - } + logReportThread.stop(); } } diff --git a/xxl-job-admin/src/main/java/com/xxl/job/admin/business/scheduler/thread/JobRegistryHelper.java b/xxl-job-admin/src/main/java/com/xxl/job/admin/business/scheduler/thread/JobRegistryHelper.java index 4a57d599..4d43a5a4 100644 --- a/xxl-job-admin/src/main/java/com/xxl/job/admin/business/scheduler/thread/JobRegistryHelper.java +++ b/xxl-job-admin/src/main/java/com/xxl/job/admin/business/scheduler/thread/JobRegistryHelper.java @@ -3,9 +3,10 @@ package com.xxl.job.admin.business.scheduler.thread; import com.xxl.job.admin.business.model.XxlJobGroup; import com.xxl.job.admin.business.model.XxlJobRegistry; import com.xxl.job.admin.business.scheduler.config.XxlJobAdminBootstrap; +import com.xxl.job.core.constant.Const; import com.xxl.job.core.constant.RegistType; import com.xxl.job.core.openapi.model.RegistryRequest; -import com.xxl.job.core.constant.Const; +import com.xxl.tool.concurrent.CyclicThread; import com.xxl.tool.core.StringTool; import com.xxl.tool.response.Response; import org.slf4j.Logger; @@ -20,20 +21,25 @@ import java.util.concurrent.*; * @author xuxueli 2016-10-02 19:10:24 */ public class JobRegistryHelper { - private static Logger logger = LoggerFactory.getLogger(JobRegistryHelper.class); + private static final Logger logger = LoggerFactory.getLogger(JobRegistryHelper.class); + /** + * registry or remove thread pool + */ private ThreadPoolExecutor registryOrRemoveThreadPool = null; - private Thread registryMonitorThread; - private volatile boolean toStop = false; + /** + * registry monitor thread + */ + private CyclicThread registryMonitorThread; /** * start */ public void start(){ - // for registry or remove + // 1、for registry or remove registryOrRemoveThreadPool = new ThreadPoolExecutor( 2, 10, @@ -54,79 +60,61 @@ public class JobRegistryHelper { } }); - // for monitor - registryMonitorThread = new Thread(new Runnable() { + // 2、for registry monitor + registryMonitorThread = new CyclicThread("JobRegistryHelper#registryMonitorThread", true, new Runnable() { @Override public void run() { - while (!toStop) { - try { - // auto registry group - List groupList = XxlJobAdminBootstrap.getInstance().getXxlJobGroupMapper().findByAddressType(0); - if (groupList!=null && !groupList.isEmpty()) { - - // remove dead address (admin/executor) - List ids = XxlJobAdminBootstrap.getInstance().getXxlJobRegistryMapper().findDead(Const.DEAD_TIMEOUT, new Date()); - if (ids!=null && !ids.isEmpty()) { - XxlJobAdminBootstrap.getInstance().getXxlJobRegistryMapper().removeDead(ids); - } + // auto registry group + List groupList = XxlJobAdminBootstrap.getInstance().getXxlJobGroupMapper().findByAddressType(0); + if (groupList!=null && !groupList.isEmpty()) { + + // remove dead address (admin/executor) + List ids = XxlJobAdminBootstrap.getInstance().getXxlJobRegistryMapper().findDead(Const.DEAD_TIMEOUT, new Date()); + if (ids!=null && !ids.isEmpty()) { + XxlJobAdminBootstrap.getInstance().getXxlJobRegistryMapper().removeDead(ids); + } - // fresh online address (admin/executor) - HashMap> appAddressMap = new HashMap>(); - List list = XxlJobAdminBootstrap.getInstance().getXxlJobRegistryMapper().findAll(Const.DEAD_TIMEOUT, new Date()); - if (list != null) { - for (XxlJobRegistry item: list) { - if (RegistType.EXECUTOR.name().equals(item.getRegistryGroup())) { - String appname = item.getRegistryKey(); - List registryList = appAddressMap.get(appname); - if (registryList == null) { - registryList = new ArrayList(); - } - - if (!registryList.contains(item.getRegistryValue())) { - registryList.add(item.getRegistryValue()); - } - appAddressMap.put(appname, registryList); - } + // fresh online address (admin/executor) + HashMap> appAddressMap = new HashMap>(); + List list = XxlJobAdminBootstrap.getInstance().getXxlJobRegistryMapper().findAll(Const.DEAD_TIMEOUT, new Date()); + if (list != null) { + for (XxlJobRegistry item: list) { + if (RegistType.EXECUTOR.name().equals(item.getRegistryGroup())) { + String appname = item.getRegistryKey(); + List registryList = appAddressMap.get(appname); + if (registryList == null) { + registryList = new ArrayList(); } - } - // fresh group address - for (XxlJobGroup group: groupList) { - List registryList = appAddressMap.get(group.getAppname()); - String addressListStr = null; - if (registryList!=null && !registryList.isEmpty()) { - Collections.sort(registryList); - StringBuilder addressListSB = new StringBuilder(); - for (String item:registryList) { - addressListSB.append(item).append(","); - } - addressListStr = addressListSB.toString(); - addressListStr = addressListStr.substring(0, addressListStr.length()-1); + if (!registryList.contains(item.getRegistryValue())) { + registryList.add(item.getRegistryValue()); } - group.setAddressList(addressListStr); - group.setUpdateTime(new Date()); - - XxlJobAdminBootstrap.getInstance().getXxlJobGroupMapper().update(group); + appAddressMap.put(appname, registryList); } } - } catch (Throwable e) { - if (!toStop) { - logger.error(">>>>>>>>>>> xxl-job, job registry monitor thread error:{}", e); - } } - try { - TimeUnit.SECONDS.sleep(Const.BEAT_TIMEOUT); - } catch (Throwable e) { - if (!toStop) { - logger.error(">>>>>>>>>>> xxl-job, job registry monitor thread error:{}", e); + + // fresh group address + for (XxlJobGroup group: groupList) { + List registryList = appAddressMap.get(group.getAppname()); + String addressListStr = null; + if (registryList!=null && !registryList.isEmpty()) { + Collections.sort(registryList); + StringBuilder addressListSB = new StringBuilder(); + for (String item:registryList) { + addressListSB.append(item).append(","); + } + addressListStr = addressListSB.toString(); + addressListStr = addressListStr.substring(0, addressListStr.length()-1); } + group.setAddressList(addressListStr); + group.setUpdateTime(new Date()); + + XxlJobAdminBootstrap.getInstance().getXxlJobGroupMapper().update(group); } } - logger.info(">>>>>>>>>>> xxl-job, job registry monitor thread stop"); } - }); - registryMonitorThread.setDaemon(true); - registryMonitorThread.setName("xxl-job, admin JobRegistryMonitorHelper-registryMonitorThread"); + }, Const.BEAT_TIMEOUT * 1000L, true); registryMonitorThread.start(); } @@ -135,18 +123,12 @@ public class JobRegistryHelper { * stop */ public void stop(){ - toStop = true; - // stop registryOrRemoveThreadPool + // 1、registryOrRemoveThreadPool registryOrRemoveThreadPool.shutdownNow(); - // stop monitor (interrupt and wait) - registryMonitorThread.interrupt(); - try { - registryMonitorThread.join(); - } catch (Throwable e) { - logger.error(e.getMessage(), e); - } + // 2、registryMonitorThread + registryMonitorThread.stop(); } diff --git a/xxl-job-admin/src/main/java/com/xxl/job/admin/business/scheduler/thread/JobScheduleHelper.java b/xxl-job-admin/src/main/java/com/xxl/job/admin/business/scheduler/thread/JobScheduleHelper.java index 5d4bfa88..e60c531f 100644 --- a/xxl-job-admin/src/main/java/com/xxl/job/admin/business/scheduler/thread/JobScheduleHelper.java +++ b/xxl-job-admin/src/main/java/com/xxl/job/admin/business/scheduler/thread/JobScheduleHelper.java @@ -44,7 +44,11 @@ public class JobScheduleHelper { */ public void start(){ - // schedule thread + // init thread flag + scheduleThreadToStop = false; + ringThreadToStop = false; + + // 1、schedule thread scheduleThread = new Thread(new Runnable() { @Override public void run() { @@ -191,8 +195,7 @@ public class JobScheduleHelper { scheduleThread.setName("xxl-job, admin JobScheduleHelper#scheduleThread"); scheduleThread.start(); - - // ring thread + // 2、ring thread ringThread = new Thread(new Runnable() { @Override public void run() { @@ -348,7 +351,7 @@ public class JobScheduleHelper { } } - // stop ring (wait job-in-memory stop) + // 2、stop ring (wait job-in-memory stop) ringThreadToStop = true; try { TimeUnit.SECONDS.sleep(1);