Selaa lähdekoodia

调度中心任务注册检测逻辑优化;

xuxueli 8 vuotta sitten
vanhempi
commit
7b052d7331

+ 0 - 18
xxl-job-admin/src/main/java/com/xxl/job/admin/controller/JobGroupController.java Näytä tiedosto

@@ -1,12 +1,9 @@
1 1
 package com.xxl.job.admin.controller;
2 2
 
3 3
 import com.xxl.job.admin.core.model.XxlJobGroup;
4
-import com.xxl.job.admin.core.thread.JobRegistryMonitorHelper;
5 4
 import com.xxl.job.admin.dao.IXxlJobGroupDao;
6 5
 import com.xxl.job.admin.dao.IXxlJobInfoDao;
7 6
 import com.xxl.job.core.biz.model.ReturnT;
8
-import com.xxl.job.core.enums.RegistryConfig;
9
-import org.apache.commons.collections.CollectionUtils;
10 7
 import org.apache.commons.lang.StringUtils;
11 8
 import org.springframework.stereotype.Controller;
12 9
 import org.springframework.ui.Model;
@@ -14,7 +11,6 @@ import org.springframework.web.bind.annotation.RequestMapping;
14 11
 import org.springframework.web.bind.annotation.ResponseBody;
15 12
 
16 13
 import javax.annotation.Resource;
17
-import java.util.Arrays;
18 14
 import java.util.List;
19 15
 
20 16
 /**
@@ -36,20 +32,6 @@ public class JobGroupController {
36 32
 		// job group (executor)
37 33
 		List<XxlJobGroup> list = xxlJobGroupDao.findAll();
38 34
 
39
-		if (CollectionUtils.isNotEmpty(list)) {
40
-			for (XxlJobGroup group: list) {
41
-				List<String> registryList = null;
42
-				if (group.getAddressType() == 0) {
43
-					registryList = JobRegistryMonitorHelper.discover(RegistryConfig.RegistType.EXECUTOR.name(), group.getAppName());
44
-				} else {
45
-					if (StringUtils.isNotBlank(group.getAddressList())) {
46
-						registryList = Arrays.asList(group.getAddressList().split(","));
47
-					}
48
-				}
49
-				group.setRegistryList(registryList);
50
-			}
51
-		}
52
-
53 35
 		model.addAttribute("list", list);
54 36
 		return "jobgroup/jobgroup.index";
55 37
 	}

+ 11 - 8
xxl-job-admin/src/main/java/com/xxl/job/admin/core/model/XxlJobGroup.java Näytä tiedosto

@@ -1,5 +1,9 @@
1 1
 package com.xxl.job.admin.core.model;
2 2
 
3
+import org.apache.commons.lang.StringUtils;
4
+
5
+import java.util.ArrayList;
6
+import java.util.Arrays;
3 7
 import java.util.List;
4 8
 
5 9
 /**
@@ -16,6 +20,12 @@ public class XxlJobGroup {
16 20
 
17 21
     // registry list
18 22
     private List<String> registryList;  // 执行器地址列表(系统注册)
23
+    public List<String> getRegistryList() {
24
+        if (StringUtils.isNotBlank(addressList)) {
25
+            registryList = new ArrayList<String>(Arrays.asList(addressList.split(",")));
26
+        }
27
+        return registryList;
28
+    }
19 29
 
20 30
     public int getId() {
21 31
         return id;
@@ -49,14 +59,6 @@ public class XxlJobGroup {
49 59
         this.order = order;
50 60
     }
51 61
 
52
-    public List<String> getRegistryList() {
53
-        return registryList;
54
-    }
55
-
56
-    public void setRegistryList(List<String> registryList) {
57
-        this.registryList = registryList;
58
-    }
59
-
60 62
     public int getAddressType() {
61 63
         return addressType;
62 64
     }
@@ -72,4 +74,5 @@ public class XxlJobGroup {
72 74
     public void setAddressList(String addressList) {
73 75
         this.addressList = addressList;
74 76
     }
77
+
75 78
 }

+ 36 - 28
xxl-job-admin/src/main/java/com/xxl/job/admin/core/thread/JobRegistryMonitorHelper.java Näytä tiedosto

@@ -1,14 +1,17 @@
1 1
 package com.xxl.job.admin.core.thread;
2 2
 
3
+import com.xxl.job.admin.core.model.XxlJobGroup;
3 4
 import com.xxl.job.admin.core.model.XxlJobRegistry;
4 5
 import com.xxl.job.admin.core.schedule.XxlJobDynamicScheduler;
5 6
 import com.xxl.job.core.enums.RegistryConfig;
7
+import org.apache.commons.collections.CollectionUtils;
8
+import org.apache.commons.lang.StringUtils;
6 9
 import org.slf4j.Logger;
7 10
 import org.slf4j.LoggerFactory;
8 11
 
9 12
 import java.util.ArrayList;
13
+import java.util.HashMap;
10 14
 import java.util.List;
11
-import java.util.concurrent.ConcurrentHashMap;
12 15
 import java.util.concurrent.TimeUnit;
13 16
 
14 17
 /**
@@ -23,8 +26,6 @@ public class JobRegistryMonitorHelper {
23 26
 		return instance;
24 27
 	}
25 28
 
26
-	private ConcurrentHashMap<String, List<String>> registMap = new ConcurrentHashMap<String, List<String>>();
27
-
28 29
 	private Thread registryThread;
29 30
 	private boolean toStop = false;
30 31
 	public void start(){
@@ -33,26 +34,42 @@ public class JobRegistryMonitorHelper {
33 34
 			public void run() {
34 35
 				while (!toStop) {
35 36
 					try {
36
-                        // remove dead admin/executor
37
-						XxlJobDynamicScheduler.xxlJobRegistryDao.removeDead(RegistryConfig.DEAD_TIMEOUT);
37
+						// auto registry group
38
+						List<XxlJobGroup> groupList = XxlJobDynamicScheduler.xxlJobGroupDao.findByAddressType(0);
39
+						if (CollectionUtils.isNotEmpty(groupList)) {
38 40
 
39
-                        // fresh registry map
40
-						ConcurrentHashMap<String, List<String>> temp = new ConcurrentHashMap<String, List<String>>();
41
-						List<XxlJobRegistry> list = XxlJobDynamicScheduler.xxlJobRegistryDao.findAll(RegistryConfig.DEAD_TIMEOUT);
42
-						if (list != null) {
43
-							for (XxlJobRegistry item: list) {
44
-								String groupKey = makeGroupKey(item.getRegistryGroup(), item.getRegistryKey());
45
-								List<String> registryList = temp.get(groupKey);
46
-								if (registryList == null) {
47
-									registryList = new ArrayList<String>();
48
-								}
49
-								if (!registryList.contains(item.getRegistryValue())) {
50
-									registryList.add(item.getRegistryValue());
41
+							// remove dead address (admin/executor)
42
+							XxlJobDynamicScheduler.xxlJobRegistryDao.removeDead(RegistryConfig.DEAD_TIMEOUT);
43
+
44
+							// fresh online address (admin/executor)
45
+							HashMap<String, List<String>> appAddressMap = new HashMap<String, List<String>>();
46
+							List<XxlJobRegistry> list = XxlJobDynamicScheduler.xxlJobRegistryDao.findAll(RegistryConfig.DEAD_TIMEOUT);
47
+							if (list != null) {
48
+								for (XxlJobRegistry item: list) {
49
+									if (RegistryConfig.RegistType.EXECUTOR.name().equals(item.getRegistryGroup())) {
50
+										String appName = item.getRegistryKey();
51
+										List<String> registryList = appAddressMap.get(appName);
52
+										if (registryList == null) {
53
+											registryList = new ArrayList<String>();
54
+										}
55
+
56
+										if (!registryList.contains(item.getRegistryValue())) {
57
+											registryList.add(item.getRegistryValue());
58
+										}
59
+										appAddressMap.put(appName, registryList);
60
+									}
51 61
 								}
52
-								temp.put(groupKey, registryList);
62
+							}
63
+
64
+							// fresh group address
65
+							for (XxlJobGroup group: groupList) {
66
+								List<String> registryList = appAddressMap.get(group.getAppName());
67
+								String addressListStr = StringUtils.join(registryList, ",");
68
+
69
+								group.setAddressList(addressListStr);
70
+								XxlJobDynamicScheduler.xxlJobGroupDao.update(group);
53 71
 							}
54 72
 						}
55
-						registMap = temp;
56 73
 					} catch (Exception e) {
57 74
 						logger.error("job registry instance error:{}", e);
58 75
 					}
@@ -73,13 +90,4 @@ public class JobRegistryMonitorHelper {
73 90
 		//registryThread.interrupt();
74 91
 	}
75 92
 
76
-	private static String makeGroupKey(String registryGroup, String registryKey){
77
-		return registryGroup.concat("_").concat(registryKey);
78
-	}
79
-	
80
-	public static List<String> discover(String registryGroup, String registryKey){
81
-		String groupKey = makeGroupKey(registryGroup, registryKey);
82
-		return instance.registMap.get(groupKey);
83
-	}
84
-	
85 93
 }

+ 3 - 14
xxl-job-admin/src/main/java/com/xxl/job/admin/core/trigger/XxlJobTrigger.java Näytä tiedosto

@@ -7,18 +7,14 @@ import com.xxl.job.admin.core.model.XxlJobLog;
7 7
 import com.xxl.job.admin.core.route.ExecutorRouteStrategyEnum;
8 8
 import com.xxl.job.admin.core.schedule.XxlJobDynamicScheduler;
9 9
 import com.xxl.job.admin.core.thread.JobFailMonitorHelper;
10
-import com.xxl.job.admin.core.thread.JobRegistryMonitorHelper;
11 10
 import com.xxl.job.core.biz.model.ReturnT;
12 11
 import com.xxl.job.core.biz.model.TriggerParam;
13 12
 import com.xxl.job.core.enums.ExecutorBlockStrategyEnum;
14
-import com.xxl.job.core.enums.RegistryConfig;
15 13
 import org.apache.commons.collections.CollectionUtils;
16
-import org.apache.commons.lang.StringUtils;
17 14
 import org.slf4j.Logger;
18 15
 import org.slf4j.LoggerFactory;
19 16
 
20 17
 import java.util.ArrayList;
21
-import java.util.Arrays;
22 18
 import java.util.Date;
23 19
 
24 20
 /**
@@ -90,17 +86,10 @@ public class XxlJobTrigger {
90 86
         StringBuffer triggerSb = new StringBuffer();
91 87
 
92 88
         // exerutor address list
93
-        ArrayList<String> addressList = null;
94 89
         XxlJobGroup group = XxlJobDynamicScheduler.xxlJobGroupDao.load(jobInfo.getJobGroup());
95
-        if (group.getAddressType() == 0) {
96
-            triggerSb.append("注册方式:自动注册");
97
-            addressList = (ArrayList<String>) JobRegistryMonitorHelper.discover(RegistryConfig.RegistType.EXECUTOR.name(), group.getAppName());
98
-        } else {
99
-            triggerSb.append("注册方式:手动录入");
100
-            if (StringUtils.isNotBlank(group.getAddressList())) {
101
-                addressList = new ArrayList<String>(Arrays.asList(group.getAddressList().split(",")));
102
-            }
103
-        }
90
+        triggerSb.append( (group.getAddressType() == 0)?"注册方式:自动注册":"注册方式:手动录入" );
91
+        ArrayList<String> addressList = (ArrayList<String>) group.getRegistryList();
92
+
104 93
         triggerSb.append("<br>阻塞处理策略:").append(ExecutorBlockStrategyEnum.match(jobInfo.getExecutorBlockStrategy(), ExecutorBlockStrategyEnum.SERIAL_EXECUTION).getTitle());
105 94
         triggerSb.append("<br>失败处理策略:").append(ExecutorFailStrategyEnum.match(jobInfo.getExecutorBlockStrategy(), ExecutorFailStrategyEnum.FAIL_ALARM).getTitle());
106 95
         triggerSb.append("<br>地址列表:").append(addressList!=null?addressList.toString():"");

+ 2 - 0
xxl-job-admin/src/main/java/com/xxl/job/admin/dao/IXxlJobGroupDao.java Näytä tiedosto

@@ -11,6 +11,8 @@ public interface IXxlJobGroupDao {
11 11
 
12 12
     public List<XxlJobGroup> findAll();
13 13
 
14
+    public List<XxlJobGroup> findByAddressType(int addressType);
15
+
14 16
     public int save(XxlJobGroup xxlJobGroup);
15 17
 
16 18
     public int update(XxlJobGroup xxlJobGroup);

+ 5 - 0
xxl-job-admin/src/main/java/com/xxl/job/admin/dao/impl/XxlJobGroupDaoImpl.java Näytä tiedosto

@@ -23,6 +23,11 @@ public class XxlJobGroupDaoImpl implements IXxlJobGroupDao {
23 23
     }
24 24
 
25 25
     @Override
26
+    public List<XxlJobGroup> findByAddressType(int addressType) {
27
+        return sqlSessionTemplate.selectList("XxlJobGroupMapper.findByAddressType", addressType);
28
+    }
29
+
30
+    @Override
26 31
     public int save(XxlJobGroup xxlJobGroup) {
27 32
         return sqlSessionTemplate.update("XxlJobGroupMapper.save", xxlJobGroup);
28 33
     }

+ 4 - 12
xxl-job-admin/src/main/java/com/xxl/job/admin/service/impl/XxlJobServiceImpl.java Näytä tiedosto

@@ -5,7 +5,6 @@ import com.xxl.job.admin.core.model.XxlJobGroup;
5 5
 import com.xxl.job.admin.core.model.XxlJobInfo;
6 6
 import com.xxl.job.admin.core.route.ExecutorRouteStrategyEnum;
7 7
 import com.xxl.job.admin.core.schedule.XxlJobDynamicScheduler;
8
-import com.xxl.job.admin.core.thread.JobRegistryMonitorHelper;
9 8
 import com.xxl.job.admin.dao.IXxlJobGroupDao;
10 9
 import com.xxl.job.admin.dao.IXxlJobInfoDao;
11 10
 import com.xxl.job.admin.dao.IXxlJobLogDao;
@@ -13,7 +12,6 @@ import com.xxl.job.admin.dao.IXxlJobLogGlueDao;
13 12
 import com.xxl.job.admin.service.IXxlJobService;
14 13
 import com.xxl.job.core.biz.model.ReturnT;
15 14
 import com.xxl.job.core.enums.ExecutorBlockStrategyEnum;
16
-import com.xxl.job.core.enums.RegistryConfig;
17 15
 import com.xxl.job.core.glue.GlueTypeEnum;
18 16
 import org.apache.commons.collections.CollectionUtils;
19 17
 import org.apache.commons.lang.StringUtils;
@@ -288,21 +286,15 @@ public class XxlJobServiceImpl implements IXxlJobService {
288 286
 		// executor count
289 287
 		Set<String> executerAddressSet = new HashSet<String>();
290 288
 		List<XxlJobGroup> groupList = xxlJobGroupDao.findAll();
289
+
291 290
 		if (CollectionUtils.isNotEmpty(groupList)) {
292 291
 			for (XxlJobGroup group: groupList) {
293
-				List<String> registryList = null;
294
-				if (group.getAddressType() == 0) {
295
-					registryList = JobRegistryMonitorHelper.discover(RegistryConfig.RegistType.EXECUTOR.name(), group.getAppName());
296
-				} else {
297
-					if (StringUtils.isNotBlank(group.getAddressList())) {
298
-						registryList = Arrays.asList(group.getAddressList().split(","));
299
-					}
300
-				}
301
-				if (CollectionUtils.isNotEmpty(registryList)) {
302
-					executerAddressSet.addAll(registryList);
292
+				if (CollectionUtils.isNotEmpty(group.getRegistryList())) {
293
+					executerAddressSet.addAll(group.getRegistryList());
303 294
 				}
304 295
 			}
305 296
 		}
297
+
306 298
 		int executorCount = executerAddressSet.size();
307 299
 
308 300
 		Map<String, Object> dashboardMap = new HashMap<String, Object>();

+ 8 - 1
xxl-job-admin/src/main/resources/mybatis-mapper/XxlJobGroupMapper.xml Näytä tiedosto

@@ -21,12 +21,19 @@
21 21
 		t.address_list
22 22
 	</sql>
23 23
 
24
-	<select id="findAll" parameterType="java.lang.Integer" resultMap="XxlJobGroup">
24
+	<select id="findAll" resultMap="XxlJobGroup">
25 25
 		SELECT <include refid="Base_Column_List" />
26 26
 		FROM XXL_JOB_QRTZ_TRIGGER_GROUP AS t
27 27
 		ORDER BY t.order ASC
28 28
 	</select>
29 29
 
30
+	<select id="findByAddressType" parameterType="java.lang.Integer" resultMap="XxlJobGroup">
31
+		SELECT <include refid="Base_Column_List" />
32
+		FROM XXL_JOB_QRTZ_TRIGGER_GROUP AS t
33
+		WHERE t.address_type = #{addressType}
34
+		ORDER BY t.order ASC
35
+	</select>
36
+
30 37
 	<insert id="save" parameterType="com.xxl.job.admin.core.model.XxlJobGroup" >
31 38
 		INSERT INTO XXL_JOB_QRTZ_TRIGGER_GROUP ( `app_name`, `title`, `order`, `address_type`, `address_list`)
32 39
 		values ( #{appName}, #{title}, #{order}, #{addressType}, #{addressList});