Merge pull request #8 from qq254963746/develop

Develop

Author: qq254963746

job-core/src/main/java/com/lts/job/core/cluster/Config.java

@@ -146,6 +146,13 @@ private Map<String, Number> getNumbers() {
         }
         return numbers;
     }
+    public boolean getParameter(String key, boolean defaultValue) {
+        String value = getParameter(key);
+        if (value == null || value.length() == 0) {
+            return defaultValue;
+        }
+        return Boolean.parseBoolean(value);
+    }
     public int getParameter(String key, int defaultValue) {
         Number n = getNumbers().get(key);
         if (n != null) {

job-core/src/main/java/com/lts/job/core/cluster/Node.java

@@ -1,5 +1,7 @@
 package com.lts.job.core.cluster;
 
+import com.lts.job.core.registry.NodeRegistryUtils;
+
 import java.util.ArrayList;
 import java.util.List;
 
@@ -11,7 +13,7 @@ public class Node {
 
     // 是否可用
     private boolean isAvailable = true;
-
+    private String clusterName;
     private NodeType nodeType;
     private String ip;
     private Integer port;
@@ -25,6 +27,8 @@ public class Node {
     // 自己关注的节点类型
     private List<NodeType> listenNodeTypes;
 
+    private String fullString;
+
     public boolean isAvailable() {
         return isAvailable;
     }
@@ -104,6 +108,14 @@ public void addListenNodeType(NodeType nodeType) {
         this.listenNodeTypes.add(nodeType);
     }
 
+    public String getClusterName() {
+        return clusterName;
+    }
+
+    public void setClusterName(String clusterName) {
+        this.clusterName = clusterName;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;
@@ -124,10 +136,18 @@ public String getAddress() {
         return ip + ":" + port;
     }
 
+    public String toFullString() {
+        if (fullString == null) {
+            fullString = NodeRegistryUtils.getFullPath(this);
+        }
+        return fullString;
+    }
+
     @Override
     public String toString() {
         return "Node{" +
                 "identity='" + identity + '\'' +
+                ", clusterName='" + clusterName + '\'' +
                 ", nodeType=" + nodeType +
                 ", ip='" + ip + '\'' +
                 ", port=" + port +

job-core/src/main/java/com/lts/job/core/constant/Constants.java

@@ -40,6 +40,9 @@ public interface Constants {
 
     public static final String UNREGISTER = "unregister";
 
+    public static final String SUBSCRIBE = "subscribe";
+
+    public static final String UNSUBSCRIBE = "unsubscribe";
     /**
      * 注册中心失败事件重试事件
      */

job-core/src/main/java/com/lts/job/core/extension/ExtensionLoader.java

@@ -569,7 +569,7 @@ private String createAdaptiveExtensionClassCode() {
                         break;
                     }
                 }
-                // 有类型为URL的参数
+                // 有类型为Config的参数
                 if (configTypeIndex != -1) {
                     // Null Point check
                     String s = String.format("\nif (arg%d == null) throw new IllegalArgumentException(\"config == null\");",
@@ -579,11 +579,11 @@ private String createAdaptiveExtensionClassCode() {
                     s = String.format("\n%s config = arg%d;", Config.class.getName(), configTypeIndex);
                     code.append(s);
                 }
-                // 参数没有URL类型
+                // 参数没有Config类型
                 else {
                     String attribMethod = null;
 
-                    // 找到参数的URL属性
+                    // 找到参数的Config属性
                     LBL_PTS:
                     for (int i = 0; i < pts.length; ++i) {
                         Method[] ms = pts[i].getMethods();

job-core/src/main/java/com/lts/job/core/factory/NodeFactory.java

@@ -18,6 +18,7 @@ public static <T extends Node> T create(Class<T> clazz, Config config) {
             node.setThreads(config.getWorkThreads());
             node.setPort(config.getListenPort());
             node.setIdentity(config.getIdentity());
+            node.setClusterName(config.getClusterName());
             return node;
         } catch (InstantiationException e) {
             throw new RuntimeException(e);

job-core/src/main/java/com/lts/job/core/registry/NodeRegistryUtils.java

@@ -65,10 +65,10 @@ public static Node parse(String fullPath) {
         return node;
     }
 
-    public static String getFullPath(String clusterName, Node node) {
+    public static String getFullPath(Node node) {
         StringBuilder path = new StringBuilder();
 
-        path.append(getRootPath(clusterName))
+        path.append(getRootPath(node.getClusterName()))
                 .append("/")
                 .append(node.getNodeType())
                 .append("/")
@@ -106,13 +106,14 @@ public static void main(String[] args) {
         node.setNodeType(NodeType.JOB_TRACKER);
         node.setCreateTime(new Date().getTime());
         node.setPort(2313);
+        node.setClusterName("lts");
         node.setIp(NetUtils.getLocalHost());
-        String fullPath = NodeRegistryUtils.getFullPath("lts", node);
+        String fullPath = NodeRegistryUtils.getFullPath(node);
         System.out.println(fullPath);
 
         node = NodeRegistryUtils.parse(fullPath);
         node.setNodeType(NodeType.JOB_CLIENT);
-        fullPath = NodeRegistryUtils.getFullPath("lts", node);
+        fullPath = NodeRegistryUtils.getFullPath(node);
         System.out.println(fullPath);
 
         node = NodeRegistryUtils.parse(fullPath);

job-core/src/main/java/com/lts/job/core/registry/RegistryFactory.java

@@ -26,6 +26,11 @@ public static Registry getRegistry(Config config) {
                     address.replace("redis://", "")
             );
             return new RedisRegistry(config);
+        } else if(address.startsWith("multicast://")){
+//            config.setRegistryAddress(
+//                    address.replace("multicast://", "")
+//            );
+//            return new MulticastRegistry(config);
         }
         throw new IllegalArgumentException("illegal address protocol");
     }