diff --git a/README.md b/README.md index bb5a9b55..b5fe19ff 100644 --- a/README.md +++ b/README.md @@ -857,6 +857,7 @@ Tips: 历史版本(V1.3.x)目前已经Release至稳定版本, 进入维护阶段 - 1、任务Cron更新逻辑优化,改为rescheduleJob,同时防止cron重复设置; - 2、优化:API回调服务失败状态码优化,方便问题排查; - 3、XxlJobLogger的日志多参数支持; +- 4、路由策略新增 "忙碌转移" 模式:按照顺序依次进行空闲检测,第一个空闲检测成功的机器选定为目标执行器并发起调度; #### TODO LIST - 1、任务权限管理:执行器为粒度分配权限,核心操作校验权限; @@ -868,7 +869,7 @@ Tips: 历史版本(V1.3.x)目前已经Release至稳定版本, 进入维护阶段 - 7、调度任务优先级; - 8、移除quartz依赖,重写调度模块:新增或恢复任务时将下次执行记录插入delayqueue,调度中心集群竞争分布式锁,成功节点批量加载到期delayqueue数据,批量执行。 - 9、任务线程轮空30次后自动销毁,降低低频任务的无效线程消耗。 -- 10、路由策略新增 "忙碌转移" 模式,发现执行器运行中,主动转移下一个执行器调度任务; + ## 七、其他 diff --git a/xxl-job-admin/src/main/java/com/xxl/job/admin/core/jobbean/RemoteHttpJobBean.java b/xxl-job-admin/src/main/java/com/xxl/job/admin/core/jobbean/RemoteHttpJobBean.java index fba3dd54..13be274f 100644 --- a/xxl-job-admin/src/main/java/com/xxl/job/admin/core/jobbean/RemoteHttpJobBean.java +++ b/xxl-job-admin/src/main/java/com/xxl/job/admin/core/jobbean/RemoteHttpJobBean.java @@ -114,76 +114,82 @@ public class RemoteHttpJobBean extends QuartzJobBean { return new ReturnT(ReturnT.FAIL_CODE, triggerSb.toString()); } - // trigger remote executor - if (addressList.size() == 1) { - String address = addressList.get(0); - jobLog.setExecutorAddress(address); - - ReturnT runResult = runExecutor(triggerParam, address); - triggerSb.append("
----------------------
").append(runResult.getMsg()); + // executor route strategy + ExecutorRouteStrategyEnum executorRouteStrategyEnum = ExecutorRouteStrategyEnum.match(jobInfo.getExecutorRouteStrategy(), null); + if (executorRouteStrategyEnum == null) { + triggerSb.append("
----------------------
").append("调度失败:").append("执行器路由策略为空"); + return new ReturnT(ReturnT.FAIL_CODE, triggerSb.toString()); + } + triggerSb.append("
路由策略:").append(executorRouteStrategyEnum.name() + "-" + executorRouteStrategyEnum.getTitle()); - return new ReturnT(runResult.getCode(), triggerSb.toString()); - } else { - // executor route strategy - ExecutorRouteStrategyEnum executorRouteStrategyEnum = ExecutorRouteStrategyEnum.match(jobInfo.getExecutorRouteStrategy(), null); - triggerSb.append("
路由策略:").append(executorRouteStrategyEnum!=null?(executorRouteStrategyEnum.name() + "-" + executorRouteStrategyEnum.getTitle()):null); - if (executorRouteStrategyEnum == null) { - triggerSb.append("
----------------------
").append("调度失败:").append("执行器路由策略为空"); - return new ReturnT(ReturnT.FAIL_CODE, triggerSb.toString()); - } + // trigger remote executor + if (executorRouteStrategyEnum == ExecutorRouteStrategyEnum.FAILOVER) { + for (String address : addressList) { + // beat + ReturnT beatResult = null; + try { + ExecutorBiz executorBiz = (ExecutorBiz) new NetComClientProxy(ExecutorBiz.class, address).getObject(); + beatResult = executorBiz.beat(); + } catch (Exception e) { + logger.error("", e); + beatResult = new ReturnT(ReturnT.FAIL_CODE, ""+e ); + } + triggerSb.append("
----------------------
") + .append("心跳检测:") + .append("
address:").append(address) + .append("
code:").append(beatResult.getCode()) + .append("
msg:").append(beatResult.getMsg()); - if (executorRouteStrategyEnum != ExecutorRouteStrategyEnum.FAILOVER) { - // get address - String address = executorRouteStrategyEnum.getRouter().route(jobInfo.getId(), addressList); - jobLog.setExecutorAddress(address); + // beat success + if (beatResult.getCode() == ReturnT.SUCCESS_CODE) { + jobLog.setExecutorAddress(address); - // run - ReturnT runResult = runExecutor(triggerParam, address); - triggerSb.append("
----------------------
").append(runResult.getMsg()); + ReturnT runResult = runExecutor(triggerParam, address); + triggerSb.append("
----------------------
").append(runResult.getMsg()); - return new ReturnT(runResult.getCode(), triggerSb.toString()); - } else { - for (String address : addressList) { - // beat - ReturnT beatResult = beatExecutor(address); - triggerSb.append("
----------------------
").append(beatResult.getMsg()); + return new ReturnT(runResult.getCode(), triggerSb.toString()); + } + } + return new ReturnT(ReturnT.FAIL_CODE, triggerSb.toString()); + } else if (executorRouteStrategyEnum == ExecutorRouteStrategyEnum.BUSYOVER) { + for (String address : addressList) { + // beat + ReturnT idleBeatResult = null; + try { + ExecutorBiz executorBiz = (ExecutorBiz) new NetComClientProxy(ExecutorBiz.class, address).getObject(); + idleBeatResult = executorBiz.idleBeat(triggerParam.getJobId()); + } catch (Exception e) { + logger.error("", e); + idleBeatResult = new ReturnT(ReturnT.FAIL_CODE, ""+e ); + } + triggerSb.append("
----------------------
") + .append("空闲检测:") + .append("
address:").append(address) + .append("
code:").append(idleBeatResult.getCode()) + .append("
msg:").append(idleBeatResult.getMsg()); - if (beatResult.getCode() == ReturnT.SUCCESS_CODE) { - jobLog.setExecutorAddress(address); + // beat success + if (idleBeatResult.getCode() == ReturnT.SUCCESS_CODE) { + jobLog.setExecutorAddress(address); - ReturnT runResult = runExecutor(triggerParam, address); - triggerSb.append("
----------------------
").append(runResult.getMsg()); + ReturnT runResult = runExecutor(triggerParam, address); + triggerSb.append("
----------------------
").append(runResult.getMsg()); - return new ReturnT(runResult.getCode(), triggerSb.toString()); - } + return new ReturnT(runResult.getCode(), triggerSb.toString()); } - return new ReturnT(ReturnT.FAIL_CODE, triggerSb.toString()); } - } - } - - /** - * run executor - * @param address - * @return - */ - public ReturnT beatExecutor(String address){ - ReturnT beatResult = null; - try { - ExecutorBiz executorBiz = (ExecutorBiz) new NetComClientProxy(ExecutorBiz.class, address).getObject(); - beatResult = executorBiz.beat(); - } catch (Exception e) { - logger.error("", e); - beatResult = new ReturnT(ReturnT.FAIL_CODE, ""+e ); - } + return new ReturnT(ReturnT.FAIL_CODE, triggerSb.toString()); + } else { + // get address + String address = executorRouteStrategyEnum.getRouter().route(jobInfo.getId(), addressList); + jobLog.setExecutorAddress(address); - StringBuffer sb = new StringBuffer("心跳检测:"); - sb.append("
address:").append(address); - sb.append("
code:").append(beatResult.getCode()); - sb.append("
msg:").append(beatResult.getMsg()); - beatResult.setMsg(sb.toString()); + // run + ReturnT runResult = runExecutor(triggerParam, address); + triggerSb.append("
----------------------
").append(runResult.getMsg()); - return beatResult; + return new ReturnT(runResult.getCode(), triggerSb.toString()); + } } /** diff --git a/xxl-job-admin/src/main/java/com/xxl/job/admin/core/route/ExecutorRouteStrategyEnum.java b/xxl-job-admin/src/main/java/com/xxl/job/admin/core/route/ExecutorRouteStrategyEnum.java index 99244d2b..2efea6a0 100644 --- a/xxl-job-admin/src/main/java/com/xxl/job/admin/core/route/ExecutorRouteStrategyEnum.java +++ b/xxl-job-admin/src/main/java/com/xxl/job/admin/core/route/ExecutorRouteStrategyEnum.java @@ -14,7 +14,8 @@ public enum ExecutorRouteStrategyEnum { CONSISTENT_HASH("一致性HASH", new ExecutorRouteConsistentHash()), LEAST_FREQUENTLY_USED("最不经常使用", new ExecutorRouteLFU()), LEAST_RECENTLY_USED("最近最久未使用", new ExecutorRouteLRU()), - FAILOVER("故障转移", null); + FAILOVER("故障转移", null), + BUSYOVER("忙碌转移", null); ExecutorRouteStrategyEnum(String title, ExecutorRouter router) { this.title = title; diff --git a/xxl-job-core/src/main/java/com/xxl/job/core/biz/ExecutorBiz.java b/xxl-job-core/src/main/java/com/xxl/job/core/biz/ExecutorBiz.java index 40465725..6051e095 100644 --- a/xxl-job-core/src/main/java/com/xxl/job/core/biz/ExecutorBiz.java +++ b/xxl-job-core/src/main/java/com/xxl/job/core/biz/ExecutorBiz.java @@ -15,6 +15,14 @@ public interface ExecutorBiz { */ public ReturnT beat(); + /** + * idle beat + * + * @param jobId + * @return + */ + public ReturnT idleBeat(int jobId); + /** * kill * @param jobId diff --git a/xxl-job-core/src/main/java/com/xxl/job/core/biz/impl/ExecutorBizImpl.java b/xxl-job-core/src/main/java/com/xxl/job/core/biz/impl/ExecutorBizImpl.java index 0b408b29..57547676 100644 --- a/xxl-job-core/src/main/java/com/xxl/job/core/biz/impl/ExecutorBizImpl.java +++ b/xxl-job-core/src/main/java/com/xxl/job/core/biz/impl/ExecutorBizImpl.java @@ -29,6 +29,22 @@ public class ExecutorBizImpl implements ExecutorBiz { return ReturnT.SUCCESS; } + @Override + public ReturnT idleBeat(int jobId) { + + // isRunningOrHasQueue + boolean isRunningOrHasQueue = false; + JobThread jobThread = XxlJobExecutor.loadJobThread(jobId); + if (jobThread != null && jobThread.isRunningOrHasQueue()) { + isRunningOrHasQueue = true; + } + + if (isRunningOrHasQueue) { + return new ReturnT(ReturnT.FAIL_CODE, "job thread is running or has trigger queue."); + } + return ReturnT.SUCCESS; + } + @Override public ReturnT kill(int jobId) { // kill handlerThread, and create new one diff --git a/xxl-job-core/src/main/java/com/xxl/job/core/thread/ExecutorRegistryThread.java b/xxl-job-core/src/main/java/com/xxl/job/core/thread/ExecutorRegistryThread.java index ac3ab039..c567b9bb 100644 --- a/xxl-job-core/src/main/java/com/xxl/job/core/thread/ExecutorRegistryThread.java +++ b/xxl-job-core/src/main/java/com/xxl/job/core/thread/ExecutorRegistryThread.java @@ -46,7 +46,8 @@ public class ExecutorRegistryThread extends Thread { try { RegistryParam registryParam = new RegistryParam(RegistryConfig.RegistType.EXECUTOR.name(), appName, executorAddress); ReturnT registryResult = AdminApiUtil.callApiFailover(AdminApiUtil.REGISTRY, registryParam); - logger.info(">>>>>>>>>>> xxl-job registry, RegistryParam:{}, registryResult:{}", new Object[]{registryParam.toString(), registryResult.toString()}); + logger.info(">>>>>>>>>>> xxl-job Executor registry {}, RegistryParam:{}, registryResult:{}", + new Object[]{(registryResult.getCode()==ReturnT.SUCCESS_CODE?"success":"fail"), registryParam.toString(), registryResult.toString()}); } catch (Exception e) { logger.error(">>>>>>>>>>> xxl-job ExecutorRegistryThread Exception:", e); }