From 700b3dd351eb4354a65f8a0ff8c7f1f2a347384a Mon Sep 17 00:00:00 2001 From: xuxueli <931591021@qq.com> Date: Sat, 18 Aug 2018 23:40:31 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=87=E6=A1=A3=E6=9B=B4=E6=96=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 8 +- doc/XXL-JOB官方文档.md | 22 +++-- .../job/admin/core/trigger/XxlJobTrigger.java | 80 +++++++++---------- 3 files changed, 53 insertions(+), 57 deletions(-) diff --git a/README.md b/README.md index 7da8e2bd..a283f22f 100644 --- a/README.md +++ b/README.md @@ -51,10 +51,10 @@ XXL-JOB是一个轻量级分布式任务调度平台，其核心设计目标是 - 6、弹性扩容缩容：一旦有新执行器机器上线或者下线，下次调度时将会重新分配任务； - 7、路由策略：执行器集群部署时提供丰富的路由策略，包括：第一个、最后一个、轮询、随机、一致性HASH、最不经常使用、最近最久未使用、故障转移、忙碌转移等； - 8、故障转移：任务路由策略选择"故障转移"情况下，如果执行器集群中某一台机器故障，将会自动Failover切换到一台正常的执行器发送调度请求。 -- 9、失败处理策略；调度失败时的处理策略，策略包括：失败告警、失败重试； -- 10、失败重试：调度中心调度失败且启用"调度失败重试"策略时，将会自动重试一次；执行器执行失败且启用"执行失败重试"策略，或回调失败重试状态时，也将会自动重试一次； -- 11、阻塞处理策略：调度过于密集执行器来不及处理时的处理策略，策略包括：单机串行（默认）、丢弃后续调度、覆盖之前调度； -- 12、任务超时控制：支持设置任务超时时间，任务运行超时的情况下，将会主动中断任务； +- 9、阻塞处理策略：调度过于密集执行器来不及处理时的处理策略，策略包括：单机串行（默认）、丢弃后续调度、覆盖之前调度； +- 10、任务超时控制：支持自定义任务超时时间，任务运行超时将会主动中断任务； +- 11、任务失败重试：支持自定义任务失败重试次数，当任务失败时将会按照预设的失败重试次数主动进行重试； +- 12、失败处理策略；调度失败时的处理策略，默认提供失败告警、失败重试等策略； - 13、分片广播任务：执行器集群部署时，任务路由策略选择"分片广播"情况下，一次任务调度将会广播触发集群中所有执行器执行一次任务，可根据分片参数开发分片任务； - 14、动态分片：分片广播任务以执行器为维度进行分片，支持动态扩容执行器集群从而动态增加分片数量，协同进行业务处理；在进行大数据量业务操作时可显著提升任务处理能力和速度。 - 15、事件触发：除了"Cron方式"和"任务依赖方式"触发任务执行之外，支持基于事件的触发任务方式。调度中心提供触发任务单次执行的API服务，可根据业务事件灵活触发。 diff --git a/doc/XXL-JOB官方文档.md b/doc/XXL-JOB官方文档.md index 44fc4214..1b353ba3 100644 --- a/doc/XXL-JOB官方文档.md +++ b/doc/XXL-JOB官方文档.md @@ -22,10 +22,10 @@ XXL-JOB是一个轻量级分布式任务调度平台，其核心设计目标是 - 6、弹性扩容缩容：一旦有新执行器机器上线或者下线，下次调度时将会重新分配任务； - 7、路由策略：执行器集群部署时提供丰富的路由策略，包括：第一个、最后一个、轮询、随机、一致性HASH、最不经常使用、最近最久未使用、故障转移、忙碌转移等； - 8、故障转移：任务路由策略选择"故障转移"情况下，如果执行器集群中某一台机器故障，将会自动Failover切换到一台正常的执行器发送调度请求。 -- 9、失败处理策略；调度失败时的处理策略，策略包括：失败告警、失败重试； -- 10、失败重试：调度中心调度失败且启用"调度失败重试"策略时，将会自动重试一次；执行器执行失败且启用"执行失败重试"策略，或回调失败重试状态时，也将会自动重试一次； -- 11、阻塞处理策略：调度过于密集执行器来不及处理时的处理策略，策略包括：单机串行（默认）、丢弃后续调度、覆盖之前调度； -- 12、任务超时控制：支持设置任务超时时间，任务运行超时的情况下，将会主动中断任务； +- 9、阻塞处理策略：调度过于密集执行器来不及处理时的处理策略，策略包括：单机串行（默认）、丢弃后续调度、覆盖之前调度； +- 10、任务超时控制：支持自定义任务超时时间，任务运行超时将会主动中断任务； +- 11、任务失败重试：支持自定义任务失败重试次数，当任务失败时将会按照预设的失败重试次数主动进行重试； +- 12、失败处理策略；调度失败时的处理策略，默认提供失败告警、失败重试等策略； - 13、分片广播任务：执行器集群部署时，任务路由策略选择"分片广播"情况下，一次任务调度将会广播触发集群中所有执行器执行一次任务，可根据分片参数开发分片任务； - 14、动态分片：分片广播任务以执行器为维度进行分片，支持动态扩容执行器集群从而动态增加分片数量，协同进行业务处理；在进行大数据量业务操作时可显著提升任务处理能力和速度。 - 15、事件触发：除了"Cron方式"和"任务依赖方式"触发任务执行之外，支持基于事件的触发任务方式。调度中心提供触发任务单次执行的API服务，可根据业务事件灵活触发。 @@ -417,7 +417,7 @@ XXL-JOB是一个轻量级分布式任务调度平台，其核心设计目标是 ### 配置属性详细说明： - 执行器：任务的绑定的执行器，任务触发调度时将会自动发现注册成功的执行器, 实现任务自动发现功能; 另一方面也可以方便的进行任务分组。每个任务必须绑定一个执行器, 可在 "执行器管理" 进行设置; - - 描述：任务的描述信息，便于任务管理； + - 任务描述：任务的描述信息，便于任务管理； - 路由策略：当执行器集群部署时，提供丰富的路由策略，包括； FIRST（第一个）：固定选择第一个机器； LAST（最后一个）：固定选择最后一个机器； @@ -436,17 +436,17 @@ XXL-JOB是一个轻量级分布式任务调度平台，其核心设计目标是 GLUE模式(Java)：任务以源码方式维护在调度中心；该模式的任务实际上是一段继承自IJobHandler的Java类代码并 "groovy" 源码方式维护，它在执行器项目中运行，可使用@Resource/@Autowire注入执行器里中的其他服务； GLUE模式(Shell)：任务以源码方式维护在调度中心；该模式的任务实际上是一段 "shell" 脚本； GLUE模式(Python)：任务以源码方式维护在调度中心；该模式的任务实际上是一段 "python" 脚本； + GLUE模式(PHP)：任务以源码方式维护在调度中心；该模式的任务实际上是一段 "php" 脚本； GLUE模式(NodeJS)：任务以源码方式维护在调度中心；该模式的任务实际上是一段 "nodejs" 脚本； - JobHandler：运行模式为 "BEAN模式" 时生效，对应执行器中新开发的JobHandler类“@JobHandler”注解自定义的value值； + - 执行参数：任务执行所需的参数，多个参数时用逗号分隔，任务执行时将会把多个参数转换成数组传入； - 子任务：每个任务都拥有一个唯一的任务ID(任务ID可以从任务列表获取)，当本任务执行结束并且执行成功时，将会触发子任务ID所对应的任务的一次主动调度。 + - 任务超时时间：支持自定义任务超时时间，任务运行超时将会主动中断任务； - 阻塞处理策略：调度过于密集执行器来不及处理时的处理策略； 单机串行（默认）：调度请求进入单机执行器后，调度请求进入FIFO队列并以串行方式运行； 丢弃后续调度：调度请求进入单机执行器后，发现执行器存在运行的调度任务，本次请求将会被丢弃并标记为失败； 覆盖之前调度：调度请求进入单机执行器后，发现执行器存在运行的调度任务，将会终止运行中的调度任务并清空队列，然后运行本地调度任务； - - 失败处理策略；调度失败时的处理策略； - 调度失败重试：调度中心调度失败且启用该策略时，将会自动重试一次； - 执行失败重试：执行器执行失败且启用该策略，或回调失败重试状态时，也将会自动重试一次；注意，执行失败的重试调度，也将会触发执行失败重试，因此可能导致重试死循环，请谨慎使用； - - 执行参数：任务执行所需的参数，多个参数时用逗号分隔，任务执行时将会把多个参数转换成数组传入； + - 失败重试次数；支持自定义任务失败重试次数，当任务失败时将会按照预设的失败重试次数主动进行重试； - 报警邮件：任务调度失败时邮件通知的邮箱地址，支持配置多邮箱地址，配置多个邮箱地址时用逗号分隔； - 负责人：任务的负责人； @@ -930,9 +930,7 @@ echo "分片总数 total = $3" 一次完整任务流程包括"调度（调度中心） + 执行（执行器）"两个阶段。 - "故障转移"发生在调度阶段，在执行器集群部署时，如果某一台执行器发生故障，该策略支持自动进行Failover切换到一台正常的执行器机器并且完成调度请求流程。 -- "失败重试"发生在"调度 + 执行"两个阶段，如下： - - 调度失败重试：调度中心调度失败且启用"调度失败重试"策略时，将会自动重试一次； - - 执行失败重试：执行器执行失败且启用"执行失败重试"策略，或回调失败重试状态（IJobHandler.FAIL_RETRY）时，也将会自动重试一次； +- "失败重试"发生在"调度 + 执行"两个阶段，支持通过自定义任务失败重试次数，当任务失败时将会按照预设的失败重试次数主动进行重试； ### 5.14 执行器灰度上线 调度中心与业务解耦，只需部署一次后常年不需要维护。但是，执行器中托管运行着业务作业，作业上线和变更需要重启执行器，尤其是Bean模式任务。 diff --git a/xxl-job-admin/src/main/java/com/xxl/job/admin/core/trigger/XxlJobTrigger.java b/xxl-job-admin/src/main/java/com/xxl/job/admin/core/trigger/XxlJobTrigger.java index e908816c..174816ab 100644 --- a/xxl-job-admin/src/main/java/com/xxl/job/admin/core/trigger/XxlJobTrigger.java +++ b/xxl-job-admin/src/main/java/com/xxl/job/admin/core/trigger/XxlJobTrigger.java @@ -57,6 +57,7 @@ public class XxlJobTrigger { // broadcast if (ExecutorRouteStrategyEnum.SHARDING_BROADCAST == executorRouteStrategyEnum && CollectionUtils.isNotEmpty(addressList)) { + boolean onceFailed = false; for (int i = 0; i < addressList.size(); i++) { String address = addressList.get(i); @@ -86,50 +87,49 @@ public class XxlJobTrigger { triggerMsgSb.append("
").append(I18nUtil.getString("jobinfo_field_timeout")).append("：").append(jobInfo.getExecutorTimeout()); triggerMsgSb.append("
").append(I18nUtil.getString("jobinfo_field_executorFailRetryCount")).append("：").append(finalFailRetryCount); - // 3、trigger-valid - if (triggerResult.getCode()==ReturnT.SUCCESS_CODE && CollectionUtils.isEmpty(addressList)) { - triggerResult.setCode(ReturnT.FAIL_CODE); - triggerMsgSb.append("
----------------------
").append(I18nUtil.getString("jobconf_trigger_address_empty")); + + // 3.1、trigger-param + TriggerParam triggerParam = new TriggerParam(); + triggerParam.setJobId(jobInfo.getId()); + triggerParam.setExecutorHandler(jobInfo.getExecutorHandler()); + triggerParam.setExecutorParams(jobInfo.getExecutorParam()); + triggerParam.setExecutorBlockStrategy(jobInfo.getExecutorBlockStrategy()); + triggerParam.setExecutorTimeout(jobInfo.getExecutorTimeout()); + triggerParam.setLogId(jobLog.getId()); + triggerParam.setLogDateTim(jobLog.getTriggerTime().getTime()); + triggerParam.setGlueType(jobInfo.getGlueType()); + triggerParam.setGlueSource(jobInfo.getGlueSource()); + triggerParam.setGlueUpdatetime(jobInfo.getGlueUpdatetime().getTime()); + triggerParam.setBroadcastIndex(i); + triggerParam.setBroadcastTotal(addressList.size()); // update02 + + // 3.2、trigger-run (route run / trigger remote executor) + triggerResult = runExecutor(triggerParam, address); // update03 + triggerMsgSb.append("

>>>>>>>>>>>"+ I18nUtil.getString("jobconf_trigger_run") +"<<<<<<<<<<<
").append(triggerResult.getMsg()); + + + // 3.3、trigger (fail retry) + if (triggerResult.getCode()!=ReturnT.SUCCESS_CODE) { + onceFailed = true; } - if (triggerResult.getCode() == ReturnT.SUCCESS_CODE) { - // 4.1、trigger-param - TriggerParam triggerParam = new TriggerParam(); - triggerParam.setJobId(jobInfo.getId()); - triggerParam.setExecutorHandler(jobInfo.getExecutorHandler()); - triggerParam.setExecutorParams(jobInfo.getExecutorParam()); - triggerParam.setExecutorBlockStrategy(jobInfo.getExecutorBlockStrategy()); - triggerParam.setExecutorTimeout(jobInfo.getExecutorTimeout()); - triggerParam.setLogId(jobLog.getId()); - triggerParam.setLogDateTim(jobLog.getTriggerTime().getTime()); - triggerParam.setGlueType(jobInfo.getGlueType()); - triggerParam.setGlueSource(jobInfo.getGlueSource()); - triggerParam.setGlueUpdatetime(jobInfo.getGlueUpdatetime().getTime()); - triggerParam.setBroadcastIndex(i); - triggerParam.setBroadcastTotal(addressList.size()); // update02 - - // 4.2、trigger-run (route run / trigger remote executor) - triggerResult = runExecutor(triggerParam, address); // update03 - triggerMsgSb.append("

>>>>>>>>>>>"+ I18nUtil.getString("jobconf_trigger_run") +"<<<<<<<<<<<
").append(triggerResult.getMsg()); - - // 4.3、trigger (fail retry) - if (triggerResult.getCode()!=ReturnT.SUCCESS_CODE && finalFailRetryCount > 0) { - JobTriggerPoolHelper.trigger(jobId, (finalFailRetryCount-1)); - triggerMsgSb.append("

>>>>>>>>>>>"+ I18nUtil.getString("jobconf_fail_trigger_retry") +"<<<<<<<<<<<
"); - } + if (addressList.size()==i+1 && onceFailed && finalFailRetryCount > 0) { // each trigger only retry once + JobTriggerPoolHelper.trigger(jobId, (finalFailRetryCount-1)); + triggerMsgSb.append("

>>>>>>>>>>>"+ I18nUtil.getString("jobconf_fail_trigger_retry") +"<<<<<<<<<<<
"); } - // 5、save trigger-info + // 4、save trigger-info jobLog.setExecutorAddress(triggerResult.getContent()); jobLog.setTriggerCode(triggerResult.getCode()); jobLog.setTriggerMsg(triggerMsgSb.toString()); XxlJobDynamicScheduler.xxlJobLogDao.updateTriggerInfo(jobLog); - // 6、monitor trigger + // 5、monitor trigger JobFailMonitorHelper.monitor(jobLog.getId()); logger.debug(">>>>>>>>>>> xxl-job trigger end, jobId:{}", jobLog.getId()); } + } else { // 1、save log-id XxlJobLog jobLog = new XxlJobLog(); @@ -157,14 +157,12 @@ public class XxlJobTrigger { triggerMsgSb.append("
").append(I18nUtil.getString("jobinfo_field_timeout")).append("：").append(jobInfo.getExecutorTimeout()); triggerMsgSb.append("
").append(I18nUtil.getString("jobinfo_field_executorFailRetryCount")).append("：").append(finalFailRetryCount); - // 3、trigger-valid - if (triggerResult.getCode()==ReturnT.SUCCESS_CODE && CollectionUtils.isEmpty(addressList)) { + // 3.0、trigger-valid + if (CollectionUtils.isEmpty(addressList)) { triggerResult.setCode(ReturnT.FAIL_CODE); triggerMsgSb.append("
----------------------
").append(I18nUtil.getString("jobconf_trigger_address_empty")); - } - - if (triggerResult.getCode() == ReturnT.SUCCESS_CODE) { - // 4.1、trigger-param + } else { + // 3.1、trigger-param TriggerParam triggerParam = new TriggerParam(); triggerParam.setJobId(jobInfo.getId()); triggerParam.setExecutorHandler(jobInfo.getExecutorHandler()); @@ -179,24 +177,24 @@ public class XxlJobTrigger { triggerParam.setBroadcastIndex(0); triggerParam.setBroadcastTotal(1); - // 4.2、trigger-run (route run / trigger remote executor) + // 3.2、trigger-run (route run / trigger remote executor) triggerResult = executorRouteStrategyEnum.getRouter().routeRun(triggerParam, addressList); triggerMsgSb.append("

>>>>>>>>>>>"+ I18nUtil.getString("jobconf_trigger_run") +"<<<<<<<<<<<
").append(triggerResult.getMsg()); - // 4.3、trigger (fail retry) + // 3.3、trigger (fail retry) if (triggerResult.getCode()!=ReturnT.SUCCESS_CODE && finalFailRetryCount > 0) { JobTriggerPoolHelper.trigger(jobId, (finalFailRetryCount-1)); triggerMsgSb.append("

>>>>>>>>>>>"+ I18nUtil.getString("jobconf_fail_trigger_retry") +"<<<<<<<<<<<
"); } } - // 5、save trigger-info + // 4、save trigger-info jobLog.setExecutorAddress(triggerResult.getContent()); jobLog.setTriggerCode(triggerResult.getCode()); jobLog.setTriggerMsg(triggerMsgSb.toString()); XxlJobDynamicScheduler.xxlJobLogDao.updateTriggerInfo(jobLog); - // 6、monitor trigger + // 5、monitor trigger JobFailMonitorHelper.monitor(jobLog.getId()); logger.debug(">>>>>>>>>>> xxl-job trigger end, jobId:{}", jobLog.getId()); }