Przeglądaj źródła

分片任务失败重试优化,仅重试当前失败的分片;

xuxueli 7 lat temu
rodzic
commit
54fce811eb

+ 2 - 1
doc/XXL-JOB官方文档.md Wyświetl plik

1282
 - 32、底层RPC序列化协议调整为hessian2;
1282
 - 32、底层RPC序列化协议调整为hessian2;
1283
 - 33、修复表字段 “t.order”与数据库关键字冲突查询失败的问题,
1283
 - 33、修复表字段 “t.order”与数据库关键字冲突查询失败的问题,
1284
 - 34、调度中心提供API服务,支持通过API服务对任务进行查询、新增、更新、启停等操作;
1284
 - 34、调度中心提供API服务,支持通过API服务对任务进行查询、新增、更新、启停等操作;
1285
-- 35、【迭代中】分片任务失败重试优化,仅重试当前失败的分片;
1285
+- 35、分片任务失败重试优化,仅重试当前失败的分片;
1286
+- 36、【迭代中】任务参数数据框调整,手动触发时支持动态输入参数;
1286
 
1287
 
1287
 
1288
 
1288
 ### TODO LIST
1289
 ### TODO LIST

+ 1 - 0
doc/db/tables_xxl_job.sql Wyświetl plik

179
   `executor_address` varchar(255) DEFAULT NULL COMMENT '执行器地址,本次执行的地址',
179
   `executor_address` varchar(255) DEFAULT NULL COMMENT '执行器地址,本次执行的地址',
180
   `executor_handler` varchar(255) DEFAULT NULL COMMENT '执行器任务handler',
180
   `executor_handler` varchar(255) DEFAULT NULL COMMENT '执行器任务handler',
181
   `executor_param` varchar(512) DEFAULT NULL COMMENT '执行器任务参数',
181
   `executor_param` varchar(512) DEFAULT NULL COMMENT '执行器任务参数',
182
+  `executor_sharding_param` varchar(20) DEFAULT NULL COMMENT '执行器任务分片参数,格式如 1/2',
182
   `executor_fail_retry_count` int(11) NOT NULL DEFAULT '0' COMMENT '失败重试次数',
183
   `executor_fail_retry_count` int(11) NOT NULL DEFAULT '0' COMMENT '失败重试次数',
183
   `trigger_time` datetime DEFAULT NULL COMMENT '调度-时间',
184
   `trigger_time` datetime DEFAULT NULL COMMENT '调度-时间',
184
   `trigger_code` int(11) NOT NULL COMMENT '调度-结果',
185
   `trigger_code` int(11) NOT NULL COMMENT '调度-结果',

+ 1 - 1
xxl-job-admin/src/main/java/com/xxl/job/admin/controller/JobInfoController.java Wyświetl plik

92
 	@ResponseBody
92
 	@ResponseBody
93
 	//@PermessionLimit(limit = false)
93
 	//@PermessionLimit(limit = false)
94
 	public ReturnT<String> triggerJob(int id) {
94
 	public ReturnT<String> triggerJob(int id) {
95
-		JobTriggerPoolHelper.trigger(id, -1, TriggerTypeEnum.MANUAL);
95
+		JobTriggerPoolHelper.trigger(id, TriggerTypeEnum.MANUAL, -1, null);
96
 		return ReturnT.SUCCESS;
96
 		return ReturnT.SUCCESS;
97
 	}
97
 	}
98
 	
98
 	

+ 1 - 1
xxl-job-admin/src/main/java/com/xxl/job/admin/core/jobbean/RemoteHttpJobBean.java Wyświetl plik

29
 
29
 
30
 		// trigger
30
 		// trigger
31
 		//XxlJobTrigger.trigger(jobId);
31
 		//XxlJobTrigger.trigger(jobId);
32
-		JobTriggerPoolHelper.trigger(jobId, -1, TriggerTypeEnum.CRON);
32
+		JobTriggerPoolHelper.trigger(jobId, TriggerTypeEnum.CRON, -1, null);
33
 	}
33
 	}
34
 
34
 
35
 }
35
 }

+ 9 - 0
xxl-job-admin/src/main/java/com/xxl/job/admin/core/model/XxlJobLog.java Wyświetl plik

18
 	private String executorAddress;
18
 	private String executorAddress;
19
 	private String executorHandler;
19
 	private String executorHandler;
20
 	private String executorParam;
20
 	private String executorParam;
21
+	private String executorShardingParam;
21
 	private int executorFailRetryCount;
22
 	private int executorFailRetryCount;
22
 	
23
 	
23
 	// trigger info
24
 	// trigger info
78
 		this.executorParam = executorParam;
79
 		this.executorParam = executorParam;
79
 	}
80
 	}
80
 
81
 
82
+	public String getExecutorShardingParam() {
83
+		return executorShardingParam;
84
+	}
85
+
86
+	public void setExecutorShardingParam(String executorShardingParam) {
87
+		this.executorShardingParam = executorShardingParam;
88
+	}
89
+
81
 	public int getExecutorFailRetryCount() {
90
 	public int getExecutorFailRetryCount() {
82
 		return executorFailRetryCount;
91
 		return executorFailRetryCount;
83
 	}
92
 	}

+ 1 - 4
xxl-job-admin/src/main/java/com/xxl/job/admin/core/thread/JobFailMonitorHelper.java Wyświetl plik

73
 									XxlJobInfo info = XxlJobDynamicScheduler.xxlJobInfoDao.loadById(log.getJobId());
73
 									XxlJobInfo info = XxlJobDynamicScheduler.xxlJobInfoDao.loadById(log.getJobId());
74
 
74
 
75
 									if (log.getExecutorFailRetryCount() > 0) {
75
 									if (log.getExecutorFailRetryCount() > 0) {
76
-
77
-										// TODO,分片任务失败重试优化,仅重试失败分片
78
-
79
-										JobTriggerPoolHelper.trigger(log.getJobId(), (log.getExecutorFailRetryCount()-1), TriggerTypeEnum.RETRY);
76
+										JobTriggerPoolHelper.trigger(log.getJobId(), TriggerTypeEnum.RETRY, (log.getExecutorFailRetryCount()-1), log.getExecutorShardingParam());
80
 										String retryMsg = "<br><br><span style=\"color:#F39C12;\" > >>>>>>>>>>>"+ I18nUtil.getString("jobconf_trigger_type_retry") +"<<<<<<<<<<< </span><br>";
77
 										String retryMsg = "<br><br><span style=\"color:#F39C12;\" > >>>>>>>>>>>"+ I18nUtil.getString("jobconf_trigger_type_retry") +"<<<<<<<<<<< </span><br>";
81
 										log.setTriggerMsg(log.getTriggerMsg() + retryMsg);
78
 										log.setTriggerMsg(log.getTriggerMsg() + retryMsg);
82
 										XxlJobDynamicScheduler.xxlJobLogDao.updateTriggerInfo(log);
79
 										XxlJobDynamicScheduler.xxlJobLogDao.updateTriggerInfo(log);

+ 4 - 4
xxl-job-admin/src/main/java/com/xxl/job/admin/core/thread/JobTriggerPoolHelper.java Wyświetl plik

29
             new ThreadPoolExecutor.CallerRunsPolicy());
29
             new ThreadPoolExecutor.CallerRunsPolicy());
30
 
30
 
31
 
31
 
32
-    public void addTrigger(final int jobId, final int failRetryCount, final TriggerTypeEnum triggerType) {
32
+    public void addTrigger(final int jobId, final TriggerTypeEnum triggerType, final int failRetryCount, final String executorShardingParam) {
33
         triggerPool.execute(new Runnable() {
33
         triggerPool.execute(new Runnable() {
34
             @Override
34
             @Override
35
             public void run() {
35
             public void run() {
36
-                XxlJobTrigger.trigger(jobId, failRetryCount, triggerType);
36
+                XxlJobTrigger.trigger(jobId, triggerType, failRetryCount, executorShardingParam);
37
             }
37
             }
38
         });
38
         });
39
     }
39
     }
55
      * 			<0: use param from job info config
55
      * 			<0: use param from job info config
56
      *
56
      *
57
      */
57
      */
58
-    public static void trigger(int jobId, int failRetryCount, TriggerTypeEnum triggerType) {
59
-        helper.addTrigger(jobId, failRetryCount, triggerType);
58
+    public static void trigger(int jobId, TriggerTypeEnum triggerType, int failRetryCount, String executorShardingParam) {
59
+        helper.addTrigger(jobId, triggerType, failRetryCount, executorShardingParam);
60
     }
60
     }
61
 
61
 
62
     public static void toStop() {
62
     public static void toStop() {

+ 25 - 13
xxl-job-admin/src/main/java/com/xxl/job/admin/core/trigger/XxlJobTrigger.java Wyświetl plik

13
 import com.xxl.job.core.enums.ExecutorBlockStrategyEnum;
13
 import com.xxl.job.core.enums.ExecutorBlockStrategyEnum;
14
 import com.xxl.job.core.util.IpUtil;
14
 import com.xxl.job.core.util.IpUtil;
15
 import org.apache.commons.collections4.CollectionUtils;
15
 import org.apache.commons.collections4.CollectionUtils;
16
+import org.apache.commons.lang3.StringUtils;
16
 import org.slf4j.Logger;
17
 import org.slf4j.Logger;
17
 import org.slf4j.LoggerFactory;
18
 import org.slf4j.LoggerFactory;
18
 
19
 
19
-import java.util.ArrayList;
20
 import java.util.Date;
20
 import java.util.Date;
21
-import java.util.List;
22
 
21
 
23
 /**
22
 /**
24
  * xxl-job trigger
23
  * xxl-job trigger
36
      * 			<0: use param from job info config
35
      * 			<0: use param from job info config
37
      *
36
      *
38
      */
37
      */
39
-    public static void trigger(int jobId, int failRetryCount, TriggerTypeEnum triggerType) {
38
+    public static void trigger(int jobId, TriggerTypeEnum triggerType, int failRetryCount, String executorShardingParam) {
40
         // load data
39
         // load data
41
-        XxlJobInfo jobInfo = XxlJobDynamicScheduler.xxlJobInfoDao.loadById(jobId);              // job info
40
+        XxlJobInfo jobInfo = XxlJobDynamicScheduler.xxlJobInfoDao.loadById(jobId);
42
         if (jobInfo == null) {
41
         if (jobInfo == null) {
43
             logger.warn(">>>>>>>>>>>> trigger fail, jobId invalid,jobId={}", jobId);
42
             logger.warn(">>>>>>>>>>>> trigger fail, jobId invalid,jobId={}", jobId);
44
             return;
43
             return;
45
         }
44
         }
46
         int finalFailRetryCount = failRetryCount>=0?failRetryCount:jobInfo.getExecutorFailRetryCount();
45
         int finalFailRetryCount = failRetryCount>=0?failRetryCount:jobInfo.getExecutorFailRetryCount();
47
-        XxlJobGroup group = XxlJobDynamicScheduler.xxlJobGroupDao.load(jobInfo.getJobGroup());  // group info
46
+        XxlJobGroup group = XxlJobDynamicScheduler.xxlJobGroupDao.load(jobInfo.getJobGroup());
48
 
47
 
49
         // process trigger
48
         // process trigger
50
-        if (ExecutorRouteStrategyEnum.SHARDING_BROADCAST == ExecutorRouteStrategyEnum.match(jobInfo.getExecutorRouteStrategy(), null) && CollectionUtils.isNotEmpty(group.getRegistryList())) {
51
-            for (int i = 0; i < group.getRegistryList().size(); i++) {
52
-                processTrigger(group, jobInfo, finalFailRetryCount, triggerType, i);
49
+        if (triggerType==TriggerTypeEnum.RETRY && executorShardingParam!=null) {
50
+            String[] shardingArr = executorShardingParam.split("/");
51
+            if (shardingArr.length==2 && StringUtils.isNumeric(shardingArr[0]) && StringUtils.isNumeric(shardingArr[1])); {
52
+                processTrigger(group, jobInfo, finalFailRetryCount, triggerType, Integer.valueOf(shardingArr[0]), Integer.valueOf(shardingArr[1]));
53
             }
53
             }
54
         } else {
54
         } else {
55
-            processTrigger(group, jobInfo, finalFailRetryCount, triggerType, 0);
55
+            if (CollectionUtils.isNotEmpty(group.getRegistryList())) {
56
+                if (ExecutorRouteStrategyEnum.SHARDING_BROADCAST == ExecutorRouteStrategyEnum.match(jobInfo.getExecutorRouteStrategy(), null)) {
57
+                    for (int i = 0; i < group.getRegistryList().size(); i++) {
58
+                        processTrigger(group, jobInfo, finalFailRetryCount, triggerType, i, group.getRegistryList().size());
59
+                    }
60
+                } else {
61
+                    processTrigger(group, jobInfo, finalFailRetryCount, triggerType, 0, 1);
62
+                }
63
+            } else {
64
+                processTrigger(group, jobInfo, finalFailRetryCount, triggerType, 0, 0);
65
+            }
56
         }
66
         }
57
     }
67
     }
58
 
68
 
59
-    private static void processTrigger(XxlJobGroup group, XxlJobInfo jobInfo, int finalFailRetryCount, TriggerTypeEnum triggerType, int index){
69
+    private static void processTrigger(XxlJobGroup group, XxlJobInfo jobInfo, int finalFailRetryCount, TriggerTypeEnum triggerType, int index, int total){
60
 
70
 
61
         // param
71
         // param
62
         ExecutorBlockStrategyEnum blockStrategy = ExecutorBlockStrategyEnum.match(jobInfo.getExecutorBlockStrategy(), ExecutorBlockStrategyEnum.SERIAL_EXECUTION);  // block strategy
72
         ExecutorBlockStrategyEnum blockStrategy = ExecutorBlockStrategyEnum.match(jobInfo.getExecutorBlockStrategy(), ExecutorBlockStrategyEnum.SERIAL_EXECUTION);  // block strategy
83
         triggerParam.setGlueSource(jobInfo.getGlueSource());
93
         triggerParam.setGlueSource(jobInfo.getGlueSource());
84
         triggerParam.setGlueUpdatetime(jobInfo.getGlueUpdatetime().getTime());
94
         triggerParam.setGlueUpdatetime(jobInfo.getGlueUpdatetime().getTime());
85
         triggerParam.setBroadcastIndex(index);
95
         triggerParam.setBroadcastIndex(index);
86
-        triggerParam.setBroadcastTotal(group.getRegistryList()!=null?group.getRegistryList().size():0);
96
+        triggerParam.setBroadcastTotal(total);
87
 
97
 
88
         // 3、init address
98
         // 3、init address
89
         String address = null;
99
         String address = null;
90
         ReturnT<String> routeAddressResult = null;
100
         ReturnT<String> routeAddressResult = null;
101
+        String shardingParam = (ExecutorRouteStrategyEnum.SHARDING_BROADCAST==executorRouteStrategyEnum &&total>0)?String.valueOf(triggerParam.getBroadcastIndex()).concat("/").concat(String.valueOf(triggerParam.getBroadcastTotal())):null;
91
         if (CollectionUtils.isNotEmpty(group.getRegistryList())) {
102
         if (CollectionUtils.isNotEmpty(group.getRegistryList())) {
92
             if (ExecutorRouteStrategyEnum.SHARDING_BROADCAST == executorRouteStrategyEnum) {
103
             if (ExecutorRouteStrategyEnum.SHARDING_BROADCAST == executorRouteStrategyEnum) {
93
                 address = group.getRegistryList().get(index);
104
                 address = group.getRegistryList().get(index);
117
                 .append( (group.getAddressType() == 0)?I18nUtil.getString("jobgroup_field_addressType_0"):I18nUtil.getString("jobgroup_field_addressType_1") );
128
                 .append( (group.getAddressType() == 0)?I18nUtil.getString("jobgroup_field_addressType_0"):I18nUtil.getString("jobgroup_field_addressType_1") );
118
         triggerMsgSb.append("<br>").append(I18nUtil.getString("jobconf_trigger_exe_regaddress")).append(":").append(group.getRegistryList());
129
         triggerMsgSb.append("<br>").append(I18nUtil.getString("jobconf_trigger_exe_regaddress")).append(":").append(group.getRegistryList());
119
         triggerMsgSb.append("<br>").append(I18nUtil.getString("jobinfo_field_executorRouteStrategy")).append(":").append(executorRouteStrategyEnum.getTitle());
130
         triggerMsgSb.append("<br>").append(I18nUtil.getString("jobinfo_field_executorRouteStrategy")).append(":").append(executorRouteStrategyEnum.getTitle());
120
-        if (ExecutorRouteStrategyEnum.SHARDING_BROADCAST == ExecutorRouteStrategyEnum.match(jobInfo.getExecutorRouteStrategy(), null)) {
121
-            triggerMsgSb.append("("+index+"/"+(group.getRegistryList()!=null?group.getRegistryList().size():0)+")");
131
+        if (shardingParam != null) {
132
+            triggerMsgSb.append("("+shardingParam+")");
122
         }
133
         }
123
         triggerMsgSb.append("<br>").append(I18nUtil.getString("jobinfo_field_executorBlockStrategy")).append(":").append(blockStrategy.getTitle());
134
         triggerMsgSb.append("<br>").append(I18nUtil.getString("jobinfo_field_executorBlockStrategy")).append(":").append(blockStrategy.getTitle());
124
         triggerMsgSb.append("<br>").append(I18nUtil.getString("jobinfo_field_timeout")).append(":").append(jobInfo.getExecutorTimeout());
135
         triggerMsgSb.append("<br>").append(I18nUtil.getString("jobinfo_field_timeout")).append(":").append(jobInfo.getExecutorTimeout());
131
         jobLog.setExecutorAddress(address);
142
         jobLog.setExecutorAddress(address);
132
         jobLog.setExecutorHandler(jobInfo.getExecutorHandler());
143
         jobLog.setExecutorHandler(jobInfo.getExecutorHandler());
133
         jobLog.setExecutorParam(jobInfo.getExecutorParam());
144
         jobLog.setExecutorParam(jobInfo.getExecutorParam());
145
+        jobLog.setExecutorShardingParam(shardingParam);
134
         jobLog.setExecutorFailRetryCount(finalFailRetryCount);
146
         jobLog.setExecutorFailRetryCount(finalFailRetryCount);
135
         //jobLog.setTriggerTime();
147
         //jobLog.setTriggerTime();
136
         jobLog.setTriggerCode(triggerResult.getCode());
148
         jobLog.setTriggerCode(triggerResult.getCode());

+ 1 - 1
xxl-job-admin/src/main/java/com/xxl/job/admin/service/impl/AdminBizImpl.java Wyświetl plik

71
                     int childJobId = (StringUtils.isNotBlank(childJobIds[i]) && StringUtils.isNumeric(childJobIds[i]))?Integer.valueOf(childJobIds[i]):-1;
71
                     int childJobId = (StringUtils.isNotBlank(childJobIds[i]) && StringUtils.isNumeric(childJobIds[i]))?Integer.valueOf(childJobIds[i]):-1;
72
                     if (childJobId > 0) {
72
                     if (childJobId > 0) {
73
 
73
 
74
-                        JobTriggerPoolHelper.trigger(childJobId, 0, TriggerTypeEnum.PARENT);
74
+                        JobTriggerPoolHelper.trigger(childJobId, TriggerTypeEnum.PARENT, 0, null);
75
                         ReturnT<String> triggerChildResult = ReturnT.SUCCESS;
75
                         ReturnT<String> triggerChildResult = ReturnT.SUCCESS;
76
 
76
 
77
                         // add msg
77
                         // add msg

+ 3 - 0
xxl-job-admin/src/main/resources/mybatis-mapper/XxlJobLogMapper.xml Wyświetl plik

12
 		<result column="executor_address" property="executorAddress" />
12
 		<result column="executor_address" property="executorAddress" />
13
 		<result column="executor_handler" property="executorHandler" />
13
 		<result column="executor_handler" property="executorHandler" />
14
 	    <result column="executor_param" property="executorParam" />
14
 	    <result column="executor_param" property="executorParam" />
15
+		<result column="executor_sharding_param" property="executorShardingParam" />
15
 		<result column="executor_fail_retry_count" property="executorFailRetryCount" />
16
 		<result column="executor_fail_retry_count" property="executorFailRetryCount" />
16
 	    
17
 	    
17
 	    <result column="trigger_time" property="triggerTime" />
18
 	    <result column="trigger_time" property="triggerTime" />
31
 		t.executor_address,
32
 		t.executor_address,
32
 		t.executor_handler,
33
 		t.executor_handler,
33
 		t.executor_param,
34
 		t.executor_param,
35
+		t.executor_sharding_param,
34
 		t.executor_fail_retry_count,
36
 		t.executor_fail_retry_count,
35
 		t.trigger_time,
37
 		t.trigger_time,
36
 		t.trigger_code,
38
 		t.trigger_code,
141
 			`executor_address`= #{executorAddress},
143
 			`executor_address`= #{executorAddress},
142
 			`executor_handler`=#{executorHandler},
144
 			`executor_handler`=#{executorHandler},
143
 			`executor_param`= #{executorParam},
145
 			`executor_param`= #{executorParam},
146
+			`executor_sharding_param`= #{executorShardingParam},
144
 			`executor_fail_retry_count`= #{executorFailRetryCount}
147
 			`executor_fail_retry_count`= #{executorFailRetryCount}
145
 		WHERE `id`= #{id}
148
 		WHERE `id`= #{id}
146
 	</update>
149
 	</update>