msb-public-java
/
xxl-job
mirror of https://gitee.com/xuxueli0323/xxl-job.git
4
2
Code Issues Projects Releases Wiki Activity

refactor(scheduler):重构调度中心组件管理机制

Browse Source 
- 移除组件单例模式，统一通过XxlJobAdminBootstrap管理
- 优化任务触发和回调逻辑，避免资源泄漏
- 组件生命周期方法标准化，提升可维护性
- 调整任务完成处理类命名和调用方式
- 统一调度相关组件的启动和停止流程- 优化失败监控组件命名和调用链路
- 规范化各调度线程池组件的初始化逻辑
pull/72/head
xuxueli 2 months ago
parent 1a109ee050
commit 3f6a051a0a
11 changed files with 153 additions and 116 deletions
Show Stats Download Patch File Download Diff File

10
doc/XXL-JOB官方文档.md
Unescape View File

@ -2573,11 +2573,11 @@ public void execute() {
- 1、【升级】升级多项maven依赖至较新版本，如 netty、groovy、spring、spring-ai、dify 等；
- 2、【修复】合并PR-2369，修复脚本任务参数取值问题；
- 3、【优化】报表统计SQL优化，修复小概率情况下查询null值问题；
- 4、【重构】调度中心底层组件重构，组件初始化以及销毁逻辑统一处理，避免资源泄漏风险；重构任务完成和触发逻辑，提升系统可维护性；
- 5、【ING】底层组件移除单例写法，汇总factory统一管理；
- 6、【ING】UI框架重构升级，提升交互体验；
- 7、【ING】调整资源加载逻辑，移除不必要的拦截器逻辑，提升页面加载效率；
- 8、【ING】事务SQL下沉至Mapper统一管理，降低维护成本；
- 4、【重构】调度中心底层组件重构，组件初始化以及销毁逻辑统一处理，任务触发及和回调逻辑优化，避免资源泄漏风险；
- 5、【重构】调度中心底层组件模块化拆分，移除组件单例以及静态代码逻辑，提升组件可维护性；
- 6、【ING】事务SQL下沉至Mapper统一管理，降低维护成本；
- 7、【ING】UI框架重构升级，提升交互体验；
- 8、【ING】调整资源加载逻辑，移除不必要的拦截器逻辑，提升页面加载效率；
- 9、【ING】规范API交互协议，通用响应结构体调整为Response；
- 10、【ING】Http通讯组件升级，基于接口代理方式重构；

4
xxl-job-admin/src/main/java/com/xxl/job/admin/scheduler/complete/JobCompleter.java
Unescape View File

@ -4,7 +4,7 @@ import com.xxl.job.admin.mapper.XxlJobInfoMapper;
import com.xxl.job.admin.mapper.XxlJobLogMapper;
import com.xxl.job.admin.model.XxlJobInfo;
import com.xxl.job.admin.model.XxlJobLog;
import com.xxl.job.admin.scheduler.thread.JobTriggerPoolHelper;
import com.xxl.job.admin.scheduler.config.XxlJobAdminBootstrap;
import com.xxl.job.admin.scheduler.trigger.TriggerTypeEnum;
import com.xxl.job.admin.util.I18nUtil;
import com.xxl.job.core.biz.model.ReturnT;
@ -82,7 +82,7 @@ public class JobCompleter {
}
// trigger child job
JobTriggerPoolHelper.trigger(childJobId, TriggerTypeEnum.PARENT, -1, null, null, null);
XxlJobAdminBootstrap.getInstance().getJobTriggerPoolHelper().trigger(childJobId, TriggerTypeEnum.PARENT, -1, null, null, null);
ReturnT<String> triggerChildResult = ReturnT.ofSuccess();
// add msg

73
xxl-job-admin/src/main/java/com/xxl/job/admin/scheduler/config/XxlJobAdminBootstrap.java
Unescape View File

@ -58,6 +58,24 @@ public class XxlJobAdminBootstrap implements InitializingBean, DisposableBean {
doStop();
}
// job module
private JobTriggerPoolHelper jobTriggerPoolHelper;
private JobRegistryHelper jobRegistryHelper;
private JobFailAlarmMonitorHelper jobFailAlarmMonitorHelper;
private JobCompleteHelper jobCompleteHelper;
private JobLogReportHelper jobLogReportHelper;
private JobScheduleHelper jobScheduleHelper;
public JobTriggerPoolHelper getJobTriggerPoolHelper() {
return jobTriggerPoolHelper;
}
public JobRegistryHelper getJobRegistryHelper() {
return jobRegistryHelper;
}
public JobCompleteHelper getJobCompleteHelper() {
return jobCompleteHelper;
}
/**
* do start
*/
@ -65,23 +83,29 @@ public class XxlJobAdminBootstrap implements InitializingBean, DisposableBean {
// init i18n
initI18n();
// admin trigger pool start
JobTriggerPoolHelper.toStart();
// trigger-pool start
jobTriggerPoolHelper = new JobTriggerPoolHelper();
jobTriggerPoolHelper.start();
// admin registry monitor run
JobRegistryHelper.getInstance().start();
// registry monitor start
jobRegistryHelper = new JobRegistryHelper();
jobRegistryHelper.start();
// admin fail-monitor run
JobFailMonitorHelper.getInstance().start();
// fail-alarm monitor start
jobFailAlarmMonitorHelper = new JobFailAlarmMonitorHelper();
jobFailAlarmMonitorHelper.start();
// admin lose-monitor run ( depend on JobTriggerPoolHelper )
JobCompleteHelper.getInstance().start();
// job complate start ( depend on JobTriggerPoolHelper ) for callback and result-lost
jobCompleteHelper = new JobCompleteHelper();
jobCompleteHelper.start();
// admin log report start
JobLogReportHelper.getInstance().start();
// log-report start
jobLogReportHelper = new JobLogReportHelper();
jobLogReportHelper.start();
// start-schedule ( depend on JobTriggerPoolHelper )
JobScheduleHelper.getInstance().start();
// job-schedule start ( depend on JobTriggerPoolHelper )
jobScheduleHelper = new JobScheduleHelper();
jobScheduleHelper.start();
logger.info(">>>>>>>>> xxl-job admin start success.");
}
@ -90,23 +114,24 @@ public class XxlJobAdminBootstrap implements InitializingBean, DisposableBean {
* do stop
*/
private void doStop(){
// stop-schedule
JobScheduleHelper.getInstance().toStop();
// job-schedule stop
jobScheduleHelper.stop();
// log-report stop
jobLogReportHelper.stop();
// admin log report stop
JobLogReportHelper.getInstance().toStop();
// job complate stop
jobCompleteHelper.stop();
// admin lose-monitor stop
JobCompleteHelper.getInstance().toStop();
// fail-alarm monitor stop
jobFailAlarmMonitorHelper.stop();
// admin fail-monitor stop
JobFailMonitorHelper.getInstance().toStop();
// registry monitor stop
jobRegistryHelper.stop();
// admin registry stop
JobRegistryHelper.getInstance().toStop();
// trigger-pool stop
jobTriggerPoolHelper.stop();
// admin trigger pool stop
JobTriggerPoolHelper.toStop();
logger.info(">>>>>>>>> xxl-job admin stopped.");
}

18
xxl-job-admin/src/main/java/com/xxl/job/admin/scheduler/thread/JobCompleteHelper.java
Unescape View File

@ -1,7 +1,7 @@
package com.xxl.job.admin.scheduler.thread;
import com.xxl.job.admin.scheduler.config.XxlJobAdminBootstrap;
import com.xxl.job.admin.model.XxlJobLog;
import com.xxl.job.admin.scheduler.config.XxlJobAdminBootstrap;
import com.xxl.job.admin.util.I18nUtil;
import com.xxl.job.core.biz.model.HandleCallbackParam;
import com.xxl.job.core.biz.model.ReturnT;
@ -14,23 +14,22 @@ import java.util.List;
import java.util.concurrent.*;
/**
* job lose-monitor instance
* job complate, for callback and result-lost
*
* @author xuxueli 2015-9-1 18:05:56
*/
public class JobCompleteHelper {
private static Logger logger = LoggerFactory.getLogger(JobCompleteHelper.class);
private static JobCompleteHelper instance = new JobCompleteHelper();
public static JobCompleteHelper getInstance(){
return instance;
}
// ---------------------- monitor ----------------------
private ThreadPoolExecutor callbackThreadPool = null;
private Thread monitorThread;
private volatile boolean toStop = false;
/**
* start
*/
public void start(){
// for callback
@ -116,7 +115,10 @@ public class JobCompleteHelper {
monitorThread.start();
}
public void toStop(){
/**
* stop
*/
public void stop(){
toStop = true;
// stop registryOrRemoveThreadPool

23
xxl-job-admin/src/main/java/com/xxl/job/admin/scheduler/thread/JobFailMonitorHelper.java → xxl-job-admin/src/main/java/com/xxl/job/admin/scheduler/thread/JobFailAlarmMonitorHelper.java
Unescape View File

@ -1,8 +1,8 @@
package com.xxl.job.admin.scheduler.thread;
import com.xxl.job.admin.scheduler.config.XxlJobAdminBootstrap;
import com.xxl.job.admin.model.XxlJobInfo;
import com.xxl.job.admin.model.XxlJobLog;
import com.xxl.job.admin.scheduler.config.XxlJobAdminBootstrap;
import com.xxl.job.admin.scheduler.trigger.TriggerTypeEnum;
import com.xxl.job.admin.util.I18nUtil;
import org.slf4j.Logger;
@ -12,22 +12,22 @@ import java.util.List;
import java.util.concurrent.TimeUnit;
/**
* job monitor instance
* job fail-monitor helper
*
* @author xuxueli 2015-9-1 18:05:56
*/
public class JobFailMonitorHelper {
private static Logger logger = LoggerFactory.getLogger(JobFailMonitorHelper.class);
public class JobFailAlarmMonitorHelper {
private static Logger logger = LoggerFactory.getLogger(JobFailAlarmMonitorHelper.class);
private static JobFailMonitorHelper instance = new JobFailMonitorHelper();
public static JobFailMonitorHelper getInstance(){
return instance;
}
// ---------------------- monitor ----------------------
private Thread monitorThread;
private volatile boolean toStop = false;
/**
* start
*/
public void start(){
monitorThread = new Thread(new Runnable() {
@ -52,7 +52,7 @@ public class JobFailMonitorHelper {
// 1、fail retry monitor
if (log.getExecutorFailRetryCount() > 0) {
JobTriggerPoolHelper.trigger(log.getJobId(), TriggerTypeEnum.RETRY, (log.getExecutorFailRetryCount()-1), log.getExecutorShardingParam(), log.getExecutorParam(), null);
XxlJobAdminBootstrap.getInstance().getJobTriggerPoolHelper().trigger(log.getJobId(), TriggerTypeEnum.RETRY, (log.getExecutorFailRetryCount()-1), log.getExecutorShardingParam(), log.getExecutorParam(), null);
String retryMsg = "<br><br><span style=\"color:#F39C12;\" > >>>>>>>>>>>"+ I18nUtil.getString("jobconf_trigger_type_retry") +"<<<<<<<<<<< </span><br>";
log.setTriggerMsg(log.getTriggerMsg() + retryMsg);
XxlJobAdminBootstrap.getInstance().getXxlJobLogMapper().updateTriggerInfo(log);
@ -96,7 +96,10 @@ public class JobFailMonitorHelper {
monitorThread.start();
}
public void toStop(){
/**
* stop
*/
public void stop(){
toStop = true;
// interrupt and wait
monitorThread.interrupt();

14
xxl-job-admin/src/main/java/com/xxl/job/admin/scheduler/thread/JobLogReportHelper.java
Unescape View File

@ -19,14 +19,13 @@ import java.util.concurrent.TimeUnit;
public class JobLogReportHelper {
private static Logger logger = LoggerFactory.getLogger(JobLogReportHelper.class);
private static JobLogReportHelper instance = new JobLogReportHelper();
public static JobLogReportHelper getInstance(){
return instance;
}
private Thread logrThread;
private volatile boolean toStop = false;
/**
* start
*/
public void start(){
logrThread = new Thread(new Runnable() {
@ -138,7 +137,10 @@ public class JobLogReportHelper {
logrThread.start();
}
public void toStop(){
/**
* stop
*/
public void stop(){
toStop = true;
// interrupt and wait
logrThread.interrupt();

25
xxl-job-admin/src/main/java/com/xxl/job/admin/scheduler/thread/JobRegistryHelper.java
Unescape View File

@ -14,21 +14,22 @@ import java.util.*;
import java.util.concurrent.*;
/**
* job registry instance
* job registry instance helper
*
* @author xuxueli 2016-10-02 19:10:24
*/
public class JobRegistryHelper {
private static Logger logger = LoggerFactory.getLogger(JobRegistryHelper.class);
private static JobRegistryHelper instance = new JobRegistryHelper();
public static JobRegistryHelper getInstance(){
return instance;
}
private ThreadPoolExecutor registryOrRemoveThreadPool = null;
private Thread registryMonitorThread;
private volatile boolean toStop = false;
/**
* start
*/
public void start(){
// for registry or remove
@ -128,7 +129,11 @@ public class JobRegistryHelper {
registryMonitorThread.start();
}
public void toStop(){
/**
* stop
*/
public void stop(){
toStop = true;
// stop registryOrRemoveThreadPool
@ -144,8 +149,11 @@ public class JobRegistryHelper {
}
// ---------------------- helper ----------------------
// ---------------------- tool ----------------------
/**
* registry
*/
public ReturnT<String> registry(RegistryParam registryParam) {
// valid
@ -178,6 +186,9 @@ public class JobRegistryHelper {
return ReturnT.ofSuccess();
}
/**
* registry remove
*/
public ReturnT<String> registryRemove(RegistryParam registryParam) {
// valid

26
xxl-job-admin/src/main/java/com/xxl/job/admin/scheduler/thread/JobScheduleHelper.java
Unescape View File

@ -1,8 +1,8 @@
package com.xxl.job.admin.scheduler.thread;
import com.xxl.job.admin.model.XxlJobInfo;
import com.xxl.job.admin.scheduler.config.XxlJobAdminBootstrap;
import com.xxl.job.admin.scheduler.cron.CronExpression;
import com.xxl.job.admin.model.XxlJobInfo;
import com.xxl.job.admin.scheduler.enums.MisfireStrategyEnum;
import com.xxl.job.admin.scheduler.enums.ScheduleTypeEnum;
import com.xxl.job.admin.scheduler.trigger.TriggerTypeEnum;
@ -19,12 +19,8 @@ import java.util.concurrent.TimeUnit;
* @author xuxueli 2019-05-21
*/
public class JobScheduleHelper {
private static Logger logger = LoggerFactory.getLogger(JobScheduleHelper.class);
private static final Logger logger = LoggerFactory.getLogger(JobScheduleHelper.class);
private static JobScheduleHelper instance = new JobScheduleHelper();
public static JobScheduleHelper getInstance(){
return instance;
}
public static final long PRE_READ_MS = 5000; // pre read
@ -34,6 +30,9 @@ public class JobScheduleHelper {
private volatile boolean ringThreadToStop = false;
private volatile static Map<Integer, List<Integer>> ringData = new ConcurrentHashMap<>();
/**
* start
*/
public void start(){
// schedule thread
@ -90,7 +89,7 @@ public class JobScheduleHelper {
MisfireStrategyEnum misfireStrategyEnum = MisfireStrategyEnum.match(jobInfo.getMisfireStrategy(), MisfireStrategyEnum.DO_NOTHING);
if (MisfireStrategyEnum.FIRE_ONCE_NOW == misfireStrategyEnum) {
// FIRE_ONCE_NOW 》 trigger
JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.MISFIRE, -1, null, null, null);
XxlJobAdminBootstrap.getInstance().getJobTriggerPoolHelper().trigger(jobInfo.getId(), TriggerTypeEnum.MISFIRE, -1, null, null, null);
logger.debug(">>>>>>>>>>> xxl-job, schedule push trigger : jobId = " + jobInfo.getId() );
}
@ -101,7 +100,7 @@ public class JobScheduleHelper {
// 2.2、trigger-expire < 5s：direct-trigger && make next-trigger-time
// 1、trigger
JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.CRON, -1, null, null, null);
XxlJobAdminBootstrap.getInstance().getJobTriggerPoolHelper().trigger(jobInfo.getId(), TriggerTypeEnum.CRON, -1, null, null, null);
logger.debug(">>>>>>>>>>> xxl-job, schedule push trigger : jobId = " + jobInfo.getId() );
// 2、fresh next
@ -249,7 +248,7 @@ public class JobScheduleHelper {
// do trigger
for (int jobId: ringItemData) {
// do trigger
JobTriggerPoolHelper.trigger(jobId, TriggerTypeEnum.CRON, -1, null, null, null);
XxlJobAdminBootstrap.getInstance().getJobTriggerPoolHelper().trigger(jobId, TriggerTypeEnum.CRON, -1, null, null, null);
}
// clear
ringItemData.clear();
@ -306,7 +305,10 @@ public class JobScheduleHelper {
logger.debug(">>>>>>>>>>> xxl-job, schedule push time-ring : " + ringSecond + " = " + Arrays.asList(ringItemData) );
}
public void toStop(){
/**
* stop
*/
public void stop(){
// 1、stop schedule
scheduleThreadToStop = true;
@ -366,6 +368,10 @@ public class JobScheduleHelper {
// ---------------------- tools ----------------------
/**
* generate next valid time
*/
public static Date generateNextValidTime(XxlJobInfo jobInfo, Date fromTime) throws Exception {
ScheduleTypeEnum scheduleTypeEnum = ScheduleTypeEnum.match(jobInfo.getScheduleType(), null);
if (ScheduleTypeEnum.CRON == scheduleTypeEnum) {

61
xxl-job-admin/src/main/java/com/xxl/job/admin/scheduler/thread/JobTriggerPoolHelper.java
Unescape View File

@ -23,6 +23,9 @@ public class JobTriggerPoolHelper {
private ThreadPoolExecutor fastTriggerPool = null;
private ThreadPoolExecutor slowTriggerPool = null;
/**
* start
*/
public void start(){
fastTriggerPool = new ThreadPoolExecutor(
10,
@ -63,7 +66,9 @@ public class JobTriggerPoolHelper {
});
}
/**
* stop
*/
public void stop() {
//triggerPool.shutdown();
fastTriggerPool.shutdownNow();
@ -77,15 +82,27 @@ public class JobTriggerPoolHelper {
private volatile ConcurrentMap<Integer, AtomicInteger> jobTimeoutCountMap = new ConcurrentHashMap<>();
// ---------------------- tool ----------------------
/**
* add trigger
* trigger job
*
* @param jobId
* @param triggerType
* @param failRetryCount
* >=0: use this param
* <0: use param from job info config
* @param executorShardingParam
* @param executorParam
* null: use job param
* not null: cover job param
*/
public void addTrigger(final int jobId,
final TriggerTypeEnum triggerType,
final int failRetryCount,
final String executorShardingParam,
final String executorParam,
final String addressList) {
public void trigger(final int jobId,
final TriggerTypeEnum triggerType,
final int failRetryCount,
final String executorShardingParam,
final String executorParam,
final String addressList) {
// choose thread pool
ThreadPoolExecutor triggerPool_ = fastTriggerPool;
@ -134,32 +151,4 @@ public class JobTriggerPoolHelper {
});
}
// ---------------------- helper ----------------------
private static JobTriggerPoolHelper helper = new JobTriggerPoolHelper();
public static void toStart() {
helper.start();
}
public static void toStop() {
helper.stop();
}
/**
* @param jobId
* @param triggerType
* @param failRetryCount
* >=0: use this param
* <0: use param from job info config
* @param executorShardingParam
* @param executorParam
* null: use job param
* not null: cover job param
*/
public static void trigger(int jobId, TriggerTypeEnum triggerType, int failRetryCount, String executorShardingParam, String executorParam, String addressList) {
helper.addTrigger(jobId, triggerType, failRetryCount, executorShardingParam, executorParam, addressList);
}
}

9
xxl-job-admin/src/main/java/com/xxl/job/admin/service/impl/AdminBizImpl.java
Unescape View File

@ -1,7 +1,6 @@
package com.xxl.job.admin.service.impl;
import com.xxl.job.admin.scheduler.thread.JobCompleteHelper;
import com.xxl.job.admin.scheduler.thread.JobRegistryHelper;
import com.xxl.job.admin.scheduler.config.XxlJobAdminBootstrap;
import com.xxl.job.core.biz.AdminBiz;
import com.xxl.job.core.biz.model.HandleCallbackParam;
import com.xxl.job.core.biz.model.RegistryParam;
@ -19,17 +18,17 @@ public class AdminBizImpl implements AdminBiz {
@Override
public ReturnT<String> callback(List<HandleCallbackParam> callbackParamList) {
return JobCompleteHelper.getInstance().callback(callbackParamList);
return XxlJobAdminBootstrap.getInstance().getJobCompleteHelper().callback(callbackParamList);
}
@Override
public ReturnT<String> registry(RegistryParam registryParam) {
return JobRegistryHelper.getInstance().registry(registryParam);
return XxlJobAdminBootstrap.getInstance().getJobRegistryHelper().registry(registryParam);
}
@Override
public ReturnT<String> registryRemove(RegistryParam registryParam) {
return JobRegistryHelper.getInstance().registryRemove(registryParam);
return XxlJobAdminBootstrap.getInstance().getJobRegistryHelper().registryRemove(registryParam);
}
}

6
xxl-job-admin/src/main/java/com/xxl/job/admin/service/impl/XxlJobServiceImpl.java
Unescape View File

@ -4,12 +4,12 @@ import com.xxl.job.admin.mapper.*;
import com.xxl.job.admin.model.XxlJobGroup;
import com.xxl.job.admin.model.XxlJobInfo;
import com.xxl.job.admin.model.XxlJobLogReport;
import com.xxl.job.admin.scheduler.config.XxlJobAdminBootstrap;
import com.xxl.job.admin.scheduler.cron.CronExpression;
import com.xxl.job.admin.scheduler.route.ExecutorRouteStrategyEnum;
import com.xxl.job.admin.scheduler.enums.MisfireStrategyEnum;
import com.xxl.job.admin.scheduler.enums.ScheduleTypeEnum;
import com.xxl.job.admin.scheduler.route.ExecutorRouteStrategyEnum;
import com.xxl.job.admin.scheduler.thread.JobScheduleHelper;
import com.xxl.job.admin.scheduler.thread.JobTriggerPoolHelper;
import com.xxl.job.admin.scheduler.trigger.TriggerTypeEnum;
import com.xxl.job.admin.service.XxlJobService;
import com.xxl.job.admin.util.I18nUtil;
@ -414,7 +414,7 @@ public class XxlJobServiceImpl implements XxlJobService {
executorParam = "";
}
JobTriggerPoolHelper.trigger(jobId, TriggerTypeEnum.MANUAL, -1, null, executorParam, addressList);
XxlJobAdminBootstrap.getInstance().getJobTriggerPoolHelper().trigger(jobId, TriggerTypeEnum.MANUAL, -1, null, executorParam, addressList);
return ReturnT.ofSuccess();
}

Write Preview
Loading…
Cancel
Save