调度中心参数配置逻辑重构

v1.8.2
xuxueli 7 years ago
parent 4e0d1be677
commit 0ddef14076

@ -120,7 +120,7 @@ public class JobLogController {
@ResponseBody @ResponseBody
public ReturnT<LogResult> logDetailCat(String executorAddress, long triggerTime, int logId, int fromLineNum){ public ReturnT<LogResult> logDetailCat(String executorAddress, long triggerTime, int logId, int fromLineNum){
try { try {
ExecutorBiz executorBiz = (ExecutorBiz) new NetComClientProxy(ExecutorBiz.class, executorAddress, XxlJobDynamicScheduler.getAccessToken()).getObject(); ExecutorBiz executorBiz = XxlJobDynamicScheduler.getExecutorBiz(executorAddress);
ReturnT<LogResult> logResult = executorBiz.log(triggerTime, logId, fromLineNum); ReturnT<LogResult> logResult = executorBiz.log(triggerTime, logId, fromLineNum);
// is end // is end
@ -154,7 +154,7 @@ public class JobLogController {
// request of kill // request of kill
ReturnT<String> runResult = null; ReturnT<String> runResult = null;
try { try {
ExecutorBiz executorBiz = (ExecutorBiz) new NetComClientProxy(ExecutorBiz.class, log.getExecutorAddress(), XxlJobDynamicScheduler.getAccessToken()).getObject(); ExecutorBiz executorBiz = XxlJobDynamicScheduler.getExecutorBiz(log.getExecutorAddress());
runResult = executorBiz.kill(jobInfo.getId()); runResult = executorBiz.kill(jobInfo.getId());
} catch (Exception e) { } catch (Exception e) {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);

@ -1,10 +1,7 @@
package com.xxl.job.admin.core.route; package com.xxl.job.admin.core.route;
import com.xxl.job.admin.core.schedule.XxlJobDynamicScheduler;
import com.xxl.job.core.biz.ExecutorBiz;
import com.xxl.job.core.biz.model.ReturnT; import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam; import com.xxl.job.core.biz.model.TriggerParam;
import com.xxl.job.core.rpc.netcom.NetComClientProxy;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
@ -25,30 +22,4 @@ public abstract class ExecutorRouter {
*/ */
public abstract ReturnT<String> routeRun(TriggerParam triggerParam, ArrayList<String> addressList); public abstract ReturnT<String> routeRun(TriggerParam triggerParam, ArrayList<String> addressList);
/**
* run executor
* @param triggerParam
* @param address
* @return ReturnT.content: final address
*/
public static ReturnT<String> runExecutor(TriggerParam triggerParam, String address){
ReturnT<String> runResult = null;
try {
ExecutorBiz executorBiz = (ExecutorBiz) new NetComClientProxy(ExecutorBiz.class, address, XxlJobDynamicScheduler.getAccessToken()).getObject();
runResult = executorBiz.run(triggerParam);
} catch (Exception e) {
logger.error(e.getMessage(), e);
runResult = new ReturnT<String>(ReturnT.FAIL_CODE, ""+e );
}
StringBuffer runResultSB = new StringBuffer("触发调度:");
runResultSB.append("<br>address").append(address);
runResultSB.append("<br>code").append(runResult.getCode());
runResultSB.append("<br>msg").append(runResult.getMsg());
runResult.setMsg(runResultSB.toString());
runResult.setContent(address);
return runResult;
}
} }

@ -2,10 +2,10 @@ package com.xxl.job.admin.core.route.strategy;
import com.xxl.job.admin.core.route.ExecutorRouter; import com.xxl.job.admin.core.route.ExecutorRouter;
import com.xxl.job.admin.core.schedule.XxlJobDynamicScheduler; import com.xxl.job.admin.core.schedule.XxlJobDynamicScheduler;
import com.xxl.job.admin.core.trigger.XxlJobTrigger;
import com.xxl.job.core.biz.ExecutorBiz; import com.xxl.job.core.biz.ExecutorBiz;
import com.xxl.job.core.biz.model.ReturnT; import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam; import com.xxl.job.core.biz.model.TriggerParam;
import com.xxl.job.core.rpc.netcom.NetComClientProxy;
import java.util.ArrayList; import java.util.ArrayList;
@ -26,7 +26,7 @@ public class ExecutorRouteBusyover extends ExecutorRouter {
// beat // beat
ReturnT<String> idleBeatResult = null; ReturnT<String> idleBeatResult = null;
try { try {
ExecutorBiz executorBiz = (ExecutorBiz) new NetComClientProxy(ExecutorBiz.class, address, XxlJobDynamicScheduler.getAccessToken()).getObject(); ExecutorBiz executorBiz = XxlJobDynamicScheduler.getExecutorBiz(address);
idleBeatResult = executorBiz.idleBeat(triggerParam.getJobId()); idleBeatResult = executorBiz.idleBeat(triggerParam.getJobId());
} catch (Exception e) { } catch (Exception e) {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);
@ -41,7 +41,7 @@ public class ExecutorRouteBusyover extends ExecutorRouter {
// beat success // beat success
if (idleBeatResult.getCode() == ReturnT.SUCCESS_CODE) { if (idleBeatResult.getCode() == ReturnT.SUCCESS_CODE) {
ReturnT<String> runResult = runExecutor(triggerParam, address); ReturnT<String> runResult = XxlJobTrigger.runExecutor(triggerParam, address);
idleBeatResultSB.append("<br><br>").append(runResult.getMsg()); idleBeatResultSB.append("<br><br>").append(runResult.getMsg());
// result // result

@ -1,6 +1,7 @@
package com.xxl.job.admin.core.route.strategy; package com.xxl.job.admin.core.route.strategy;
import com.xxl.job.admin.core.route.ExecutorRouter; import com.xxl.job.admin.core.route.ExecutorRouter;
import com.xxl.job.admin.core.trigger.XxlJobTrigger;
import com.xxl.job.core.biz.model.ReturnT; import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam; import com.xxl.job.core.biz.model.TriggerParam;
@ -82,7 +83,7 @@ public class ExecutorRouteConsistentHash extends ExecutorRouter {
String address = route(triggerParam.getJobId(), addressList); String address = route(triggerParam.getJobId(), addressList);
// run executor // run executor
ReturnT<String> runResult = runExecutor(triggerParam, address); ReturnT<String> runResult = XxlJobTrigger.runExecutor(triggerParam, address);
runResult.setContent(address); runResult.setContent(address);
return runResult; return runResult;
} }

@ -2,10 +2,10 @@ package com.xxl.job.admin.core.route.strategy;
import com.xxl.job.admin.core.route.ExecutorRouter; import com.xxl.job.admin.core.route.ExecutorRouter;
import com.xxl.job.admin.core.schedule.XxlJobDynamicScheduler; import com.xxl.job.admin.core.schedule.XxlJobDynamicScheduler;
import com.xxl.job.admin.core.trigger.XxlJobTrigger;
import com.xxl.job.core.biz.ExecutorBiz; import com.xxl.job.core.biz.ExecutorBiz;
import com.xxl.job.core.biz.model.ReturnT; import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam; import com.xxl.job.core.biz.model.TriggerParam;
import com.xxl.job.core.rpc.netcom.NetComClientProxy;
import java.util.ArrayList; import java.util.ArrayList;
@ -26,7 +26,7 @@ public class ExecutorRouteFailover extends ExecutorRouter {
// beat // beat
ReturnT<String> beatResult = null; ReturnT<String> beatResult = null;
try { try {
ExecutorBiz executorBiz = (ExecutorBiz) new NetComClientProxy(ExecutorBiz.class, address, XxlJobDynamicScheduler.getAccessToken()).getObject(); ExecutorBiz executorBiz = XxlJobDynamicScheduler.getExecutorBiz(address);
beatResult = executorBiz.beat(); beatResult = executorBiz.beat();
} catch (Exception e) { } catch (Exception e) {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);
@ -41,7 +41,7 @@ public class ExecutorRouteFailover extends ExecutorRouter {
// beat success // beat success
if (beatResult.getCode() == ReturnT.SUCCESS_CODE) { if (beatResult.getCode() == ReturnT.SUCCESS_CODE) {
ReturnT<String> runResult = runExecutor(triggerParam, address); ReturnT<String> runResult = XxlJobTrigger.runExecutor(triggerParam, address);
beatResultSB.append("<br><br>").append(runResult.getMsg()); beatResultSB.append("<br><br>").append(runResult.getMsg());
// result // result

@ -1,7 +1,7 @@
package com.xxl.job.admin.core.route.strategy; package com.xxl.job.admin.core.route.strategy;
import com.xxl.job.admin.core.model.XxlJobLog;
import com.xxl.job.admin.core.route.ExecutorRouter; import com.xxl.job.admin.core.route.ExecutorRouter;
import com.xxl.job.admin.core.trigger.XxlJobTrigger;
import com.xxl.job.core.biz.model.ReturnT; import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam; import com.xxl.job.core.biz.model.TriggerParam;
@ -23,7 +23,7 @@ public class ExecutorRouteFirst extends ExecutorRouter {
String address = route(triggerParam.getJobId(), addressList); String address = route(triggerParam.getJobId(), addressList);
// run executor // run executor
ReturnT<String> runResult = runExecutor(triggerParam, address); ReturnT<String> runResult = XxlJobTrigger.runExecutor(triggerParam, address);
runResult.setContent(address); runResult.setContent(address);
return runResult; return runResult;
} }

@ -1,6 +1,7 @@
package com.xxl.job.admin.core.route.strategy; package com.xxl.job.admin.core.route.strategy;
import com.xxl.job.admin.core.route.ExecutorRouter; import com.xxl.job.admin.core.route.ExecutorRouter;
import com.xxl.job.admin.core.trigger.XxlJobTrigger;
import com.xxl.job.core.biz.model.ReturnT; import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam; import com.xxl.job.core.biz.model.TriggerParam;
@ -62,7 +63,7 @@ public class ExecutorRouteLFU extends ExecutorRouter {
String address = route(triggerParam.getJobId(), addressList); String address = route(triggerParam.getJobId(), addressList);
// run executor // run executor
ReturnT<String> runResult = runExecutor(triggerParam, address); ReturnT<String> runResult = XxlJobTrigger.runExecutor(triggerParam, address);
runResult.setContent(address); runResult.setContent(address);
return runResult; return runResult;
} }

@ -1,6 +1,7 @@
package com.xxl.job.admin.core.route.strategy; package com.xxl.job.admin.core.route.strategy;
import com.xxl.job.admin.core.route.ExecutorRouter; import com.xxl.job.admin.core.route.ExecutorRouter;
import com.xxl.job.admin.core.trigger.XxlJobTrigger;
import com.xxl.job.core.biz.model.ReturnT; import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam; import com.xxl.job.core.biz.model.TriggerParam;
@ -61,7 +62,7 @@ public class ExecutorRouteLRU extends ExecutorRouter {
String address = route(triggerParam.getJobId(), addressList); String address = route(triggerParam.getJobId(), addressList);
// run executor // run executor
ReturnT<String> runResult = runExecutor(triggerParam, address); ReturnT<String> runResult = XxlJobTrigger.runExecutor(triggerParam, address);
runResult.setContent(address); runResult.setContent(address);
return runResult; return runResult;
} }

@ -1,7 +1,7 @@
package com.xxl.job.admin.core.route.strategy; package com.xxl.job.admin.core.route.strategy;
import com.xxl.job.admin.core.model.XxlJobLog;
import com.xxl.job.admin.core.route.ExecutorRouter; import com.xxl.job.admin.core.route.ExecutorRouter;
import com.xxl.job.admin.core.trigger.XxlJobTrigger;
import com.xxl.job.core.biz.model.ReturnT; import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam; import com.xxl.job.core.biz.model.TriggerParam;
@ -22,7 +22,7 @@ public class ExecutorRouteLast extends ExecutorRouter {
String address = route(triggerParam.getJobId(), addressList); String address = route(triggerParam.getJobId(), addressList);
// run executor // run executor
ReturnT<String> runResult = runExecutor(triggerParam, address); ReturnT<String> runResult = XxlJobTrigger.runExecutor(triggerParam, address);
runResult.setContent(address); runResult.setContent(address);
return runResult; return runResult;
} }

@ -1,6 +1,7 @@
package com.xxl.job.admin.core.route.strategy; package com.xxl.job.admin.core.route.strategy;
import com.xxl.job.admin.core.route.ExecutorRouter; import com.xxl.job.admin.core.route.ExecutorRouter;
import com.xxl.job.admin.core.trigger.XxlJobTrigger;
import com.xxl.job.core.biz.model.ReturnT; import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam; import com.xxl.job.core.biz.model.TriggerParam;
@ -25,7 +26,7 @@ public class ExecutorRouteRandom extends ExecutorRouter {
String address = route(triggerParam.getJobId(), addressList); String address = route(triggerParam.getJobId(), addressList);
// run executor // run executor
ReturnT<String> runResult = runExecutor(triggerParam, address); ReturnT<String> runResult = XxlJobTrigger.runExecutor(triggerParam, address);
runResult.setContent(address); runResult.setContent(address);
return runResult; return runResult;
} }

@ -1,6 +1,7 @@
package com.xxl.job.admin.core.route.strategy; package com.xxl.job.admin.core.route.strategy;
import com.xxl.job.admin.core.route.ExecutorRouter; import com.xxl.job.admin.core.route.ExecutorRouter;
import com.xxl.job.admin.core.trigger.XxlJobTrigger;
import com.xxl.job.core.biz.model.ReturnT; import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam; import com.xxl.job.core.biz.model.TriggerParam;
@ -41,7 +42,7 @@ public class ExecutorRouteRound extends ExecutorRouter {
String address = route(triggerParam.getJobId(), addressList); String address = route(triggerParam.getJobId(), addressList);
// run executor // run executor
ReturnT<String> runResult = runExecutor(triggerParam, address); ReturnT<String> runResult = XxlJobTrigger.runExecutor(triggerParam, address);
runResult.setContent(address); runResult.setContent(address);
return runResult; return runResult;
} }

@ -9,29 +9,33 @@ import com.xxl.job.admin.dao.XxlJobInfoDao;
import com.xxl.job.admin.dao.XxlJobLogDao; import com.xxl.job.admin.dao.XxlJobLogDao;
import com.xxl.job.admin.dao.XxlJobRegistryDao; import com.xxl.job.admin.dao.XxlJobRegistryDao;
import com.xxl.job.core.biz.AdminBiz; import com.xxl.job.core.biz.AdminBiz;
import com.xxl.job.core.biz.ExecutorBiz;
import com.xxl.job.core.rpc.netcom.NetComClientProxy;
import com.xxl.job.core.rpc.netcom.NetComServerFactory; import com.xxl.job.core.rpc.netcom.NetComServerFactory;
import org.quartz.*; import org.quartz.*;
import org.quartz.Trigger.TriggerState; import org.quartz.Trigger.TriggerState;
import org.quartz.impl.matchers.GroupMatcher;
import org.quartz.impl.triggers.CronTriggerImpl; import org.quartz.impl.triggers.CronTriggerImpl;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException; import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware; import org.springframework.context.ApplicationContextAware;
import org.springframework.util.Assert; import org.springframework.util.Assert;
import java.util.*; import java.util.Date;
import java.util.HashSet;
import java.util.concurrent.ConcurrentHashMap;
/** /**
* base quartz scheduler util * base quartz scheduler util
* @author xuxueli 2015-12-19 16:13:53 * @author xuxueli 2015-12-19 16:13:53
*/ */
public final class XxlJobDynamicScheduler implements ApplicationContextAware, InitializingBean { public final class XxlJobDynamicScheduler implements ApplicationContextAware {
private static final Logger logger = LoggerFactory.getLogger(XxlJobDynamicScheduler.class); private static final Logger logger = LoggerFactory.getLogger(XxlJobDynamicScheduler.class);
// Scheduler // ---------------------- param ----------------------
// scheduler
private static Scheduler scheduler; private static Scheduler scheduler;
public void setScheduler(Scheduler scheduler) { public void setScheduler(Scheduler scheduler) {
XxlJobDynamicScheduler.scheduler = scheduler; XxlJobDynamicScheduler.scheduler = scheduler;
@ -42,11 +46,25 @@ public final class XxlJobDynamicScheduler implements ApplicationContextAware, In
public void setAccessToken(String accessToken) { public void setAccessToken(String accessToken) {
this.accessToken = accessToken; this.accessToken = accessToken;
} }
public static String getAccessToken() {
return accessToken; // dao
public static XxlJobLogDao xxlJobLogDao;
public static XxlJobInfoDao xxlJobInfoDao;
public static XxlJobRegistryDao xxlJobRegistryDao;
public static XxlJobGroupDao xxlJobGroupDao;
public static AdminBiz adminBiz;
// ---------------------- applicationContext ----------------------
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
XxlJobDynamicScheduler.xxlJobLogDao = applicationContext.getBean(XxlJobLogDao.class);
XxlJobDynamicScheduler.xxlJobInfoDao = applicationContext.getBean(XxlJobInfoDao.class);
XxlJobDynamicScheduler.xxlJobRegistryDao = applicationContext.getBean(XxlJobRegistryDao.class);
XxlJobDynamicScheduler.xxlJobGroupDao = applicationContext.getBean(XxlJobGroupDao.class);
XxlJobDynamicScheduler.adminBiz = applicationContext.getBean(AdminBiz.class);
} }
// init // ---------------------- init + destroy ----------------------
public void init() throws Exception { public void init() throws Exception {
// admin registry monitor run // admin registry monitor run
JobRegistryMonitorHelper.getInstance().start(); JobRegistryMonitorHelper.getInstance().start();
@ -54,13 +72,15 @@ public final class XxlJobDynamicScheduler implements ApplicationContextAware, In
// admin monitor run // admin monitor run
JobFailMonitorHelper.getInstance().start(); JobFailMonitorHelper.getInstance().start();
// rpc-service, base on spring-mvc // admin-server(spring-mvc)
NetComServerFactory.putService(AdminBiz.class, XxlJobDynamicScheduler.adminBiz); NetComServerFactory.putService(AdminBiz.class, XxlJobDynamicScheduler.adminBiz);
NetComServerFactory.setAccessToken(accessToken); NetComServerFactory.setAccessToken(accessToken);
// valid
Assert.notNull(scheduler, "quartz scheduler is null");
logger.info(">>>>>>>>> init quartz scheduler success.[{}]", scheduler);
} }
// destroy
public void destroy(){ public void destroy(){
// admin registry stop // admin registry stop
JobRegistryMonitorHelper.getInstance().toStop(); JobRegistryMonitorHelper.getInstance().toStop();
@ -69,63 +89,34 @@ public final class XxlJobDynamicScheduler implements ApplicationContextAware, In
JobFailMonitorHelper.getInstance().toStop(); JobFailMonitorHelper.getInstance().toStop();
} }
// xxlJobLogDao、xxlJobInfoDao // ---------------------- executor-client ----------------------
public static XxlJobLogDao xxlJobLogDao; private static ConcurrentHashMap<String, ExecutorBiz> executorBizRepository = new ConcurrentHashMap<String, ExecutorBiz>();
public static XxlJobInfoDao xxlJobInfoDao; public static ExecutorBiz getExecutorBiz(String address) throws Exception {
public static XxlJobRegistryDao xxlJobRegistryDao; // valid
public static XxlJobGroupDao xxlJobGroupDao; if (address==null || address.trim().length()==0) {
public static AdminBiz adminBiz; return null;
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
XxlJobDynamicScheduler.xxlJobLogDao = applicationContext.getBean(XxlJobLogDao.class);
XxlJobDynamicScheduler.xxlJobInfoDao = applicationContext.getBean(XxlJobInfoDao.class);
XxlJobDynamicScheduler.xxlJobRegistryDao = applicationContext.getBean(XxlJobRegistryDao.class);
XxlJobDynamicScheduler.xxlJobGroupDao = applicationContext.getBean(XxlJobGroupDao.class);
XxlJobDynamicScheduler.adminBiz = applicationContext.getBean(AdminBiz.class);
} }
@Override // load-cache
public void afterPropertiesSet() throws Exception { address = address.trim();
Assert.notNull(scheduler, "quartz scheduler is null"); ExecutorBiz executorBiz = executorBizRepository.get(address);
logger.info(">>>>>>>>> init quartz scheduler success.[{}]", scheduler); if (executorBiz != null) {
return executorBiz;
} }
// getJobKeys // set-cache
@Deprecated executorBiz = (ExecutorBiz) new NetComClientProxy(ExecutorBiz.class, address, accessToken).getObject();
public static List<Map<String, Object>> getJobList(){ executorBizRepository.put(address, executorBiz);
List<Map<String, Object>> jobList = new ArrayList<Map<String,Object>>(); return executorBiz;
try {
if (scheduler.getJobGroupNames()==null || scheduler.getJobGroupNames().size()==0) {
return null;
}
String groupName = scheduler.getJobGroupNames().get(0);
Set<JobKey> jobKeys = scheduler.getJobKeys(GroupMatcher.jobGroupEquals(groupName));
if (jobKeys!=null && jobKeys.size()>0) {
for (JobKey jobKey : jobKeys) {
TriggerKey triggerKey = TriggerKey.triggerKey(jobKey.getName(), Scheduler.DEFAULT_GROUP);
Trigger trigger = scheduler.getTrigger(triggerKey);
JobDetail jobDetail = scheduler.getJobDetail(jobKey);
TriggerState triggerState = scheduler.getTriggerState(triggerKey);
Map<String, Object> jobMap = new HashMap<String, Object>();
jobMap.put("TriggerKey", triggerKey);
jobMap.put("Trigger", trigger);
jobMap.put("JobDetail", jobDetail);
jobMap.put("TriggerState", triggerState);
jobList.add(jobMap);
}
} }
} catch (SchedulerException e) { // ---------------------- schedule util ----------------------
e.printStackTrace();
return null;
}
return jobList;
}
// fill job info /**
* fill job info
*
* @param jobInfo
*/
public static void fillJobInfo(XxlJobInfo jobInfo) { public static void fillJobInfo(XxlJobInfo jobInfo) {
// TriggerKey : name + group // TriggerKey : name + group
String group = String.valueOf(jobInfo.getJobGroup()); String group = String.valueOf(jobInfo.getJobGroup());
@ -156,14 +147,28 @@ public final class XxlJobDynamicScheduler implements ApplicationContextAware, In
} }
} }
// check if exists /**
* check if exists
*
* @param jobName
* @param jobGroup
* @return
* @throws SchedulerException
*/
public static boolean checkExists(String jobName, String jobGroup) throws SchedulerException{ public static boolean checkExists(String jobName, String jobGroup) throws SchedulerException{
TriggerKey triggerKey = TriggerKey.triggerKey(jobName, jobGroup); TriggerKey triggerKey = TriggerKey.triggerKey(jobName, jobGroup);
return scheduler.checkExists(triggerKey); return scheduler.checkExists(triggerKey);
} }
// addJob 新增 /**
@SuppressWarnings("unchecked") * addJob
*
* @param jobName
* @param jobGroup
* @param cronExpression
* @return
* @throws SchedulerException
*/
public static boolean addJob(String jobName, String jobGroup, String cronExpression) throws SchedulerException { public static boolean addJob(String jobName, String jobGroup, String cronExpression) throws SchedulerException {
// TriggerKey : name + group // TriggerKey : name + group
TriggerKey triggerKey = TriggerKey.triggerKey(jobName, jobGroup); TriggerKey triggerKey = TriggerKey.triggerKey(jobName, jobGroup);
@ -196,7 +201,15 @@ public final class XxlJobDynamicScheduler implements ApplicationContextAware, In
return true; return true;
} }
// reschedule /**
* rescheduleJob
*
* @param jobGroup
* @param jobName
* @param cronExpression
* @return
* @throws SchedulerException
*/
public static boolean rescheduleJob(String jobGroup, String jobName, String cronExpression) throws SchedulerException { public static boolean rescheduleJob(String jobGroup, String jobName, String cronExpression) throws SchedulerException {
// TriggerKey valid if_exists // TriggerKey valid if_exists
@ -245,7 +258,14 @@ public final class XxlJobDynamicScheduler implements ApplicationContextAware, In
return true; return true;
} }
// unscheduleJob /**
* unscheduleJob
*
* @param jobName
* @param jobGroup
* @return
* @throws SchedulerException
*/
public static boolean removeJob(String jobName, String jobGroup) throws SchedulerException { public static boolean removeJob(String jobName, String jobGroup) throws SchedulerException {
// TriggerKey : name + group // TriggerKey : name + group
TriggerKey triggerKey = TriggerKey.triggerKey(jobName, jobGroup); TriggerKey triggerKey = TriggerKey.triggerKey(jobName, jobGroup);
@ -257,7 +277,14 @@ public final class XxlJobDynamicScheduler implements ApplicationContextAware, In
return true; return true;
} }
// Pause /**
* pause
*
* @param jobName
* @param jobGroup
* @return
* @throws SchedulerException
*/
public static boolean pauseJob(String jobName, String jobGroup) throws SchedulerException { public static boolean pauseJob(String jobName, String jobGroup) throws SchedulerException {
// TriggerKey : name + group // TriggerKey : name + group
TriggerKey triggerKey = TriggerKey.triggerKey(jobName, jobGroup); TriggerKey triggerKey = TriggerKey.triggerKey(jobName, jobGroup);
@ -273,7 +300,14 @@ public final class XxlJobDynamicScheduler implements ApplicationContextAware, In
return result; return result;
} }
// resume /**
* resume
*
* @param jobName
* @param jobGroup
* @return
* @throws SchedulerException
*/
public static boolean resumeJob(String jobName, String jobGroup) throws SchedulerException { public static boolean resumeJob(String jobName, String jobGroup) throws SchedulerException {
// TriggerKey : name + group // TriggerKey : name + group
TriggerKey triggerKey = TriggerKey.triggerKey(jobName, jobGroup); TriggerKey triggerKey = TriggerKey.triggerKey(jobName, jobGroup);
@ -289,7 +323,14 @@ public final class XxlJobDynamicScheduler implements ApplicationContextAware, In
return result; return result;
} }
// run /**
* run
*
* @param jobName
* @param jobGroup
* @return
* @throws SchedulerException
*/
public static boolean triggerJob(String jobName, String jobGroup) throws SchedulerException { public static boolean triggerJob(String jobName, String jobGroup) throws SchedulerException {
// TriggerKey : name + group // TriggerKey : name + group
JobKey jobKey = new JobKey(jobName, jobGroup); JobKey jobKey = new JobKey(jobName, jobGroup);
@ -305,5 +346,41 @@ public final class XxlJobDynamicScheduler implements ApplicationContextAware, In
return result; return result;
} }
/**
* finaAllJobList
*
* @return
*//*
@Deprecated
public static List<Map<String, Object>> finaAllJobList(){
List<Map<String, Object>> jobList = new ArrayList<Map<String,Object>>();
try {
if (scheduler.getJobGroupNames()==null || scheduler.getJobGroupNames().size()==0) {
return null;
}
String groupName = scheduler.getJobGroupNames().get(0);
Set<JobKey> jobKeys = scheduler.getJobKeys(GroupMatcher.jobGroupEquals(groupName));
if (jobKeys!=null && jobKeys.size()>0) {
for (JobKey jobKey : jobKeys) {
TriggerKey triggerKey = TriggerKey.triggerKey(jobKey.getName(), Scheduler.DEFAULT_GROUP);
Trigger trigger = scheduler.getTrigger(triggerKey);
JobDetail jobDetail = scheduler.getJobDetail(jobKey);
TriggerState triggerState = scheduler.getTriggerState(triggerKey);
Map<String, Object> jobMap = new HashMap<String, Object>();
jobMap.put("TriggerKey", triggerKey);
jobMap.put("Trigger", trigger);
jobMap.put("JobDetail", jobDetail);
jobMap.put("TriggerState", triggerState);
jobList.add(jobMap);
}
}
} catch (SchedulerException e) {
e.printStackTrace();
return null;
}
return jobList;
}*/
} }

@ -5,9 +5,9 @@ import com.xxl.job.admin.core.model.XxlJobGroup;
import com.xxl.job.admin.core.model.XxlJobInfo; import com.xxl.job.admin.core.model.XxlJobInfo;
import com.xxl.job.admin.core.model.XxlJobLog; import com.xxl.job.admin.core.model.XxlJobLog;
import com.xxl.job.admin.core.route.ExecutorRouteStrategyEnum; import com.xxl.job.admin.core.route.ExecutorRouteStrategyEnum;
import com.xxl.job.admin.core.route.ExecutorRouter;
import com.xxl.job.admin.core.schedule.XxlJobDynamicScheduler; import com.xxl.job.admin.core.schedule.XxlJobDynamicScheduler;
import com.xxl.job.admin.core.thread.JobFailMonitorHelper; import com.xxl.job.admin.core.thread.JobFailMonitorHelper;
import com.xxl.job.core.biz.ExecutorBiz;
import com.xxl.job.core.biz.model.ReturnT; import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam; import com.xxl.job.core.biz.model.TriggerParam;
import com.xxl.job.core.enums.ExecutorBlockStrategyEnum; import com.xxl.job.core.enums.ExecutorBlockStrategyEnum;
@ -90,12 +90,12 @@ public class XxlJobTrigger {
triggerParam.setBroadcastTotal(addressList.size()); // update02 triggerParam.setBroadcastTotal(addressList.size()); // update02
// 4.2、trigger-run (route run / trigger remote executor) // 4.2、trigger-run (route run / trigger remote executor)
triggerResult = ExecutorRouter.runExecutor(triggerParam, address); // update03 triggerResult = runExecutor(triggerParam, address); // update03
triggerMsgSb.append("<br><br><span style=\"color:#00c0ef;\" > >>>>>>>>>>>触发调度<<<<<<<<<<< </span><br>").append(triggerResult.getMsg()); triggerMsgSb.append("<br><br><span style=\"color:#00c0ef;\" > >>>>>>>>>>>触发调度<<<<<<<<<<< </span><br>").append(triggerResult.getMsg());
// 4.3、trigger (fail retry) // 4.3、trigger (fail retry)
if (triggerResult.getCode()!=ReturnT.SUCCESS_CODE && failStrategy == ExecutorFailStrategyEnum.FAIL_RETRY) { if (triggerResult.getCode()!=ReturnT.SUCCESS_CODE && failStrategy == ExecutorFailStrategyEnum.FAIL_RETRY) {
triggerResult = ExecutorRouter.runExecutor(triggerParam, address); // update04 triggerResult = runExecutor(triggerParam, address); // update04
triggerMsgSb.append("<br><br><span style=\"color:#F39C12;\" > >>>>>>>>>>>失败重试<<<<<<<<<<< </span><br>").append(triggerResult.getMsg()); triggerMsgSb.append("<br><br><span style=\"color:#F39C12;\" > >>>>>>>>>>>失败重试<<<<<<<<<<< </span><br>").append(triggerResult.getMsg());
} }
} }
@ -179,4 +179,30 @@ public class XxlJobTrigger {
logger.debug(">>>>>>>>>>> xxl-job trigger end, jobId:{}", jobLog.getId()); logger.debug(">>>>>>>>>>> xxl-job trigger end, jobId:{}", jobLog.getId());
} }
/**
* run executor
* @param triggerParam
* @param address
* @return ReturnT.content: final address
*/
public static ReturnT<String> runExecutor(TriggerParam triggerParam, String address){
ReturnT<String> runResult = null;
try {
ExecutorBiz executorBiz = XxlJobDynamicScheduler.getExecutorBiz(address);
runResult = executorBiz.run(triggerParam);
} catch (Exception e) {
logger.error(e.getMessage(), e);
runResult = new ReturnT<String>(ReturnT.FAIL_CODE, ""+e );
}
StringBuffer runResultSB = new StringBuffer("触发调度:");
runResultSB.append("<br>address").append(address);
runResultSB.append("<br>code").append(runResult.getCode());
runResultSB.append("<br>msg").append(runResult.getMsg());
runResult.setMsg(runResultSB.toString());
runResult.setContent(address);
return runResult;
}
} }

@ -26,7 +26,7 @@ import java.util.concurrent.ConcurrentHashMap;
public class XxlJobExecutor implements ApplicationContextAware { public class XxlJobExecutor implements ApplicationContextAware {
private static final Logger logger = LoggerFactory.getLogger(XxlJobExecutor.class); private static final Logger logger = LoggerFactory.getLogger(XxlJobExecutor.class);
// ---------------------------------- param ------------------------------------ // ---------------------- param ----------------------
private String ip; private String ip;
private int port = 9999; private int port = 9999;
private String appName; private String appName;
@ -54,7 +54,7 @@ public class XxlJobExecutor implements ApplicationContextAware {
} }
// ---------------------------------- applicationContext ------------------------------------ // ---------------------- applicationContext ----------------------
private static ApplicationContext applicationContext; private static ApplicationContext applicationContext;
@Override @Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
@ -65,7 +65,7 @@ public class XxlJobExecutor implements ApplicationContextAware {
} }
// ---------------------------------- start + stop ------------------------------------ // ---------------------- start + stop ----------------------
public void start() throws Exception { public void start() throws Exception {
// init admin-client // init admin-client
initAdminBizList(adminAddresses, accessToken); initAdminBizList(adminAddresses, accessToken);
@ -95,7 +95,7 @@ public class XxlJobExecutor implements ApplicationContextAware {
} }
// ---------------------------------- admin-client ------------------------------------ // ---------------------- admin-client ----------------------
private static List<AdminBiz> adminBizList; private static List<AdminBiz> adminBizList;
private static void initAdminBizList(String adminAddresses, String accessToken) throws Exception { private static void initAdminBizList(String adminAddresses, String accessToken) throws Exception {
if (adminAddresses!=null && adminAddresses.trim().length()>0) { if (adminAddresses!=null && adminAddresses.trim().length()>0) {
@ -116,7 +116,7 @@ public class XxlJobExecutor implements ApplicationContextAware {
} }
// ---------------------------------- executor-server ------------------------------------ // ---------------------- executor-server(jetty) ----------------------
private NetComServerFactory serverFactory = new NetComServerFactory(); private NetComServerFactory serverFactory = new NetComServerFactory();
private void initExecutorServer(int port, String ip, String appName, String accessToken) throws Exception { private void initExecutorServer(int port, String ip, String appName, String accessToken) throws Exception {
NetComServerFactory.putService(ExecutorBiz.class, new ExecutorBizImpl()); // rpc-service, base on jetty NetComServerFactory.putService(ExecutorBiz.class, new ExecutorBizImpl()); // rpc-service, base on jetty
@ -128,7 +128,7 @@ public class XxlJobExecutor implements ApplicationContextAware {
} }
// ---------------------------------- job handler repository ------------------------------------ // ---------------------- job handler repository ----------------------
private static ConcurrentHashMap<String, IJobHandler> jobHandlerRepository = new ConcurrentHashMap<String, IJobHandler>(); private static ConcurrentHashMap<String, IJobHandler> jobHandlerRepository = new ConcurrentHashMap<String, IJobHandler>();
public static IJobHandler registJobHandler(String name, IJobHandler jobHandler){ public static IJobHandler registJobHandler(String name, IJobHandler jobHandler){
logger.info("xxl-job register jobhandler success, name:{}, jobHandler:{}", name, jobHandler); logger.info("xxl-job register jobhandler success, name:{}, jobHandler:{}", name, jobHandler);
@ -156,7 +156,7 @@ public class XxlJobExecutor implements ApplicationContextAware {
} }
// ---------------------------------- job thread repository ------------------------------------ // ---------------------- job thread repository ----------------------
private static ConcurrentHashMap<Integer, JobThread> JobThreadRepository = new ConcurrentHashMap<Integer, JobThread>(); private static ConcurrentHashMap<Integer, JobThread> JobThreadRepository = new ConcurrentHashMap<Integer, JobThread>();
public static JobThread registJobThread(int jobId, IJobHandler handler, String removeOldReason){ public static JobThread registJobThread(int jobId, IJobHandler handler, String removeOldReason){
JobThread newJobThread = new JobThread(jobId, handler); JobThread newJobThread = new JobThread(jobId, handler);

Loading…
Cancel
Save