执行器自动注册逻辑更新

xxl-job-admin/src/main/java/com/xxl/job/admin/core/jobbean/RemoteHttpJobBean.java

@@ -3,8 +3,8 @@ package com.xxl.job.admin.core.jobbean;
 import com.xxl.job.admin.core.callback.XxlJobLogCallbackServer;
 import com.xxl.job.admin.core.model.XxlJobInfo;
 import com.xxl.job.admin.core.model.XxlJobLog;
-import com.xxl.job.admin.core.model.XxlJobRegistry;
 import com.xxl.job.admin.core.thread.JobMonitorHelper;
+import com.xxl.job.admin.core.thread.JobRegistryHelper;
 import com.xxl.job.admin.core.util.DynamicSchedulerUtil;
 import com.xxl.job.core.registry.RegistHelper;
 import com.xxl.job.core.router.HandlerRouter.ActionRepository;
@@ -60,17 +60,12 @@ public class RemoteHttpJobBean extends QuartzJobBean {
 		List<String> addressList = new ArrayList<String>();
 		String parseAddressMsg = null;
 		if (StringUtils.isNotBlank(jobInfo.getExecutorAppname())) {
-			List<XxlJobRegistry> xxlJobRegistryList = DynamicSchedulerUtil.xxlJobRegistryDao.findRegistrys(RegistHelper.RegistType.EXECUTOR.name(), jobInfo.getExecutorAppname());
+			addressList = JobRegistryHelper.discover(RegistHelper.RegistType.EXECUTOR.name(), jobInfo.getExecutorAppname());
-			if (xxlJobRegistryList!=null && xxlJobRegistryList.size()>0) {
+			parseAddressMsg = MessageFormat.format("Parse Address (Appname注册方式) <br>>>>[address list] : {0}<br><hr>", addressList);
-				for (XxlJobRegistry item: xxlJobRegistryList) {
-					addressList.add(item.getRegistryValue());
-				}
-			}
-			parseAddressMsg = MessageFormat.format("Parse Address (Appname注册方式) <br>>>>[address list] : {0}<br><hr>", addressList.toArray());
 		} else {
 			List<String> addressArr = Arrays.asList(jobInfo.getExecutorAddress().split(","));
 			addressList.addAll(addressArr);
-			parseAddressMsg = MessageFormat.format("Parse Address (地址配置方式) <br>>>>[address list] : {0}<br><hr>", addressList.toArray());
+			parseAddressMsg = MessageFormat.format("Parse Address (地址配置方式) <br>>>>[address list] : {0}<br><hr>", addressList);
 		}
 
 		// failover trigger
@@ -97,7 +92,21 @@ public class RemoteHttpJobBean extends QuartzJobBean {
 	 * @return
 	 */
 	public ResponseModel failoverTrigger(List<String> addressList, RequestModel requestModel, XxlJobLog jobLog){
-		if (addressList.size() > 1) {
+		 if (addressList==null || addressList.size() < 1) {
+			ResponseModel result = new ResponseModel();
+			result.setStatus(ResponseModel.FAIL);
+			result.setMsg( "Trigger error, <br>>>>address list is null <br><hr>" );
+			return result;
+		} else if (addressList.size() == 1) {
+			 String address = addressList.get(0);
+			 // store real address
+			 jobLog.setExecutorAddress(address);
+
+			 ResponseModel triggerCallback = XxlJobNetCommUtil.postHex(XxlJobNetCommUtil.addressToUrl(address), requestModel);
+			 String failoverMessage = MessageFormat.format("Trigger running, <br>>>>[address] : {0}, <br>>>>[status] : {1}, <br>>>>[msg] : {2} <br><hr>", address, triggerCallback.getStatus(), triggerCallback.getMsg());
+			 triggerCallback.setMsg(failoverMessage);
+			 return triggerCallback;
+		 } else {
 			
 			// for ha
 			Collections.shuffle(addressList);
@@ -133,20 +142,6 @@ public class RemoteHttpJobBean extends QuartzJobBean {
 			result.setStatus(ResponseModel.FAIL);
 			result.setMsg(failoverMessage);
 			return result;
-		} else if (addressList.size() == 1) {
-			String address = addressList.get(0);
-			// store real address
-			jobLog.setExecutorAddress(address);
-
-			ResponseModel triggerCallback = XxlJobNetCommUtil.postHex(XxlJobNetCommUtil.addressToUrl(address), requestModel);
-			String failoverMessage = MessageFormat.format("Trigger running, <br>>>>[address] : {0}, <br>>>>[status] : {1}, <br>>>>[msg] : {2} <br><hr>", address, triggerCallback.getStatus(), triggerCallback.getMsg());
-			triggerCallback.setMsg(failoverMessage);
-			return triggerCallback;
-		} else {
-			ResponseModel result = new ResponseModel();
-			result.setStatus(ResponseModel.FAIL);
-			result.setMsg( "Trigger error, <br>>>>address list is null <br><hr>" );
-			return result;
 		}
 	}
 

xxl-job-admin/src/main/java/com/xxl/job/admin/core/thread/JobRegistryHelper.java

@@ -0,0 +1,74 @@
+package com.xxl.job.admin.core.thread;
+
+import com.xxl.job.admin.core.model.XxlJobRegistry;
+import com.xxl.job.admin.core.util.DynamicSchedulerUtil;
+import com.xxl.job.core.registry.RegistHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * job registry helper
+ * @author xuxueli 2016-10-02 19:10:24
+ */
+public class JobRegistryHelper {
+	private static Logger logger = LoggerFactory.getLogger(JobRegistryHelper.class);
+
+	private static JobRegistryHelper helper = new JobRegistryHelper();
+	private ConcurrentHashMap<String, List<String>> registMap = new ConcurrentHashMap<String, List<String>>();
+
+	public JobRegistryHelper(){
+		Thread registryThread = new Thread(new Runnable() {
+			@Override
+			public void run() {
+				int timeout = 15;
+				while (true) {
+					try {
+						ConcurrentHashMap<String, List<String>> temp = new ConcurrentHashMap<String, List<String>>();
+						// do biz
+						DynamicSchedulerUtil.xxlJobRegistryDao.removeDead(RegistHelper.TIMEOUT*2);
+						List<XxlJobRegistry> list = DynamicSchedulerUtil.xxlJobRegistryDao.findAll(RegistHelper.TIMEOUT*2);
+						if (list != null) {
+							for (XxlJobRegistry item: list) {
+								String groupKey = makeGroupKey(item.getRegistryGroup(), item.getRegistryKey());
+								List<String> dataSet = temp.get(groupKey);
+								if (dataSet == null) {
+									dataSet = new ArrayList<String>();
+								}
+								dataSet.add(item.getRegistryValue());
+								temp.put(groupKey, dataSet);
+							}
+						}
+						// gresh registry
+						registMap = temp;
+						logger.error("job registry :{}", list);
+					} catch (Exception e) {
+						logger.error("job registry helper error:{}", e);
+					}
+					try {
+						TimeUnit.SECONDS.sleep(RegistHelper.TIMEOUT);
+					} catch (InterruptedException e) {
+						logger.error("job registry helper error:{}", e);
+					}
+				}
+			}
+		});
+		registryThread.setDaemon(true);
+		registryThread.start();
+
+	}
+
+	private static String makeGroupKey(String registryGroup, String registryKey){
+		return registryGroup.concat("_").concat(registryKey);
+	}
+	
+	public static List<String> discover(String registryGroup, String registryKey){
+		String groupKey = makeGroupKey(registryGroup, registryKey);
+		return helper.registMap.get(groupKey);
+	}
+	
+}

xxl-job-admin/src/main/java/com/xxl/job/admin/core/util/DynamicSchedulerUtil.java

@@ -3,6 +3,7 @@ package com.xxl.job.admin.core.util;
 import com.xxl.job.admin.core.callback.XxlJobLogCallbackServer;
 import com.xxl.job.admin.core.jobbean.RemoteHttpJobBean;
 import com.xxl.job.admin.core.model.XxlJobInfo;
+import com.xxl.job.admin.core.thread.JobRegistryHelper;
 import com.xxl.job.admin.dao.IXxlJobInfoDao;
 import com.xxl.job.admin.dao.IXxlJobLogDao;
 import com.xxl.job.admin.dao.IXxlJobRegistryDao;
@@ -49,6 +50,9 @@ public final class DynamicSchedulerUtil implements ApplicationContextAware, Init
 		} catch (Exception e) {
 			e.printStackTrace();
 		}
+
+		// init JobRegistryHelper
+        JobRegistryHelper.discover("g", "k");
     }
     
     // destroy

xxl-job-admin/src/main/java/com/xxl/job/admin/dao/IXxlJobRegistryDao.java

@@ -8,5 +8,7 @@ import java.util.List;
  * Created by xuxueli on 16/9/30.
  */
 public interface IXxlJobRegistryDao {
-    List<XxlJobRegistry> findRegistrys(String registryGroup, String registryKey);
+    public int removeDead(int timeout);
+
+    public List<XxlJobRegistry> findAll(int timeout);
 }

xxl-job-admin/src/main/java/com/xxl/job/admin/dao/impl/XxlJobRegistryDaoImpl.java

@@ -6,9 +6,7 @@ import org.mybatis.spring.SqlSessionTemplate;
 import org.springframework.stereotype.Repository;
 
 import javax.annotation.Resource;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 
 /**
  * Created by xuxueli on 16/9/30.
@@ -20,11 +18,13 @@ public class XxlJobRegistryDaoImpl implements IXxlJobRegistryDao {
     public SqlSessionTemplate sqlSessionTemplate;
 
     @Override
-    public List<XxlJobRegistry> findRegistrys(String registryGroup, String registryKey) {
+    public int removeDead(int timeout) {
-        Map<String, String> params = new HashMap<String, String>();
+        return sqlSessionTemplate.delete("XxlJobRegistryMapper.removeDead", timeout);
-        params.put("registryGroup", registryGroup);
+    }
-        params.put("registryKey", registryKey);
+
-        return sqlSessionTemplate.selectList("XxlJobRegistryMapper.findRegistrys", params);
+    @Override
+    public List<XxlJobRegistry> findAll(int timeout) {
+        return sqlSessionTemplate.selectList("XxlJobRegistryMapper.findAll", timeout);
     }
 
 }

xxl-job-admin/src/main/resources/mybatis-mapper/XxlJobRegistryMapper.xml

@@ -19,17 +19,15 @@
 		t.update_time
 	</sql>
 	
-	<select id="findRegistrys" parameterType="java.util.HashMap" resultMap="XxlJobRegistry">
+	<delete id="removeDead" parameterType="java.lang.Integer" >
+		DELETE FROM XXL_JOB_QRTZ_TRIGGER_REGISTRY
+		WHERE update_time <![CDATA[ < ]]> DATE_ADD(NOW(),INTERVAL -#{timeout} SECOND)
+	</delete>
+
+	<select id="findAll" parameterType="java.lang.Integer" resultMap="XxlJobRegistry">
 		SELECT <include refid="Base_Column_List" />
 		FROM XXL_JOB_QRTZ_TRIGGER_REGISTRY AS t
-		WHERE t.registry_group = #{registryGroup}
+		WHERE t.update_time <![CDATA[ > ]]> DATE_ADD(NOW(),INTERVAL -#{timeout} SECOND)
-			AND t.registry_key = #{registryKey}
-			AND t.update_time <![CDATA[ > ]]> DATE_ADD(NOW(),INTERVAL -30 SECOND)
 	</select>
 
-	<delete id="refresh" >
-		delete from XXL_JOB_QRTZ_TRIGGER_REGISTRY
-		WHERE update_time <![CDATA[ < ]]> DATE_ADD(NOW(),INTERVAL -30 SECOND)
-	</delete>
-	
 </mapper>

xxl-job-core/src/main/java/com/xxl/job/core/executor/jetty/XxlJobExecutor.java

@@ -96,7 +96,7 @@ public class XxlJobExecutor implements ApplicationContextAware {
                     try {
                         String address = IpUtil.getIp().concat(":").concat(String.valueOf(port));
                         registHelper.registry(RegistHelper.RegistType.EXECUTOR.name(), appName, address);
-                        TimeUnit.SECONDS.sleep(15);
+                        TimeUnit.SECONDS.sleep(RegistHelper.TIMEOUT);
                     } catch (Exception e) {
                         e.printStackTrace();
                     }

xxl-job-core/src/main/java/com/xxl/job/core/registry/RegistHelper.java

@@ -5,6 +5,7 @@ package com.xxl.job.core.registry;
  */
 public interface RegistHelper {
 
+    public static final int TIMEOUT = 15;
     public enum RegistType{ EXECUTOR, ADMIN }
 
     public int registry(String registGroup, String registryKey, String registryValue);