[alibaba#359] Use SPI to make health scalable

Author: XCXCXCXCX

api/src/main/java/com/alibaba/nacos/api/naming/pojo/AbstractHealthChecker.java

@@ -16,6 +16,7 @@
 package com.alibaba.nacos.api.naming.pojo;
 
 import com.alibaba.fastjson.annotation.JSONField;
+import com.alibaba.fastjson.serializer.SerializeWriter;
 import com.alibaba.nacos.api.common.Constants;
 import org.apache.commons.lang3.StringUtils;
 
@@ -45,8 +46,16 @@ public void setType(String type) {
      * @return Another instance with exactly the same fields.
      * @throws CloneNotSupportedException
      */
+    @Override
     public abstract AbstractHealthChecker clone() throws CloneNotSupportedException;
 
+    /**
+     * used to JsonAdapter
+     */
+    public void jsonAdapterCallback(SerializeWriter writer){
+        // do nothing
+    }
+
     public static class None extends AbstractHealthChecker {
 
         public static final String TYPE = "NONE";
@@ -116,6 +125,17 @@ public Map<String, String> getCustomHeaders() {
             return headerMap;
         }
 
+        /**
+         * used to JsonAdapter
+         *
+         * @param writer
+         */
+        @Override
+        public void jsonAdapterCallback(SerializeWriter writer) {
+            writer.writeFieldValue(',', "path", getPath());
+            writer.writeFieldValue(',', "headers", getHeaders());
+        }
+
         @Override
         public int hashCode() {
             return Objects.hash(path, headers, expectedResponseCode);
@@ -215,6 +235,18 @@ public void setPwd(String pwd) {
             this.pwd = pwd;
         }
 
+        /**
+         * used to JsonAdapter
+         *
+         * @param writer
+         */
+        @Override
+        public void jsonAdapterCallback(SerializeWriter writer) {
+            writer.writeFieldValue(',', "user", getUser());
+            writer.writeFieldValue(',', "pwd", getPwd());
+            writer.writeFieldValue(',', "cmd", getCmd());
+        }
+
         @Override
         public int hashCode() {
             return Objects.hash(user, pwd, cmd);

distribution/plugins/health/nacos-health-plugin-example-1.0.jar

@@ -15,48 +15,45 @@
  */
 package com.alibaba.nacos.naming.healthcheck;
 
+import com.alibaba.nacos.naming.healthcheck.extend.HealthCheckExtendProvider;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Collectors;
+
 /**
  * @author nacos
  */
 @Component("healthCheckDelegate")
 public class HealthCheckProcessorDelegate implements HealthCheckProcessor {
 
-    @Autowired
-    private HttpHealthCheckProcessor httpProcessor;
-
-    @Autowired
-    private TcpSuperSenseProcessor tcpProcessor;
+    private Map<String, HealthCheckProcessor> healthCheckProcessorMap
+        = new HashMap<>();
 
-    @Autowired
-    private MysqlHealthCheckProcessor mysqlProcessor;
+    public HealthCheckProcessorDelegate(HealthCheckExtendProvider provider) {
+        provider.init();
+    }
 
     @Autowired
-    private NoneHealthCheckProcessor noneProcessor;
+    public void addProcessor(Collection<HealthCheckProcessor> processors){
+        healthCheckProcessorMap.putAll(processors.stream()
+            .filter(processor -> processor.getType() != null)
+            .collect(Collectors.toMap(HealthCheckProcessor::getType, processor -> processor)));
+    }
 
     @Override
     public void process(HealthCheckTask task) {
 
         String type = task.getCluster().getHealthChecker().getType();
-
-        if (type.equals(httpProcessor.getType())) {
-            httpProcessor.process(task);
-            return;
-        }
-
-        if (type.equals(tcpProcessor.getType())) {
-            tcpProcessor.process(task);
-            return;
-        }
-
-        if (type.equals(mysqlProcessor.getType())) {
-            mysqlProcessor.process(task);
-            return;
+        HealthCheckProcessor processor = healthCheckProcessorMap.get(type);
+        if(processor == null){
+            processor = healthCheckProcessorMap.get("none");
         }
 
-        noneProcessor.process(task);
+        processor.process(task);
     }
 
     @Override

naming/src/main/java/com/alibaba/nacos/naming/healthcheck/HealthCheckProcessorDelegate.java

@@ -15,24 +15,49 @@
  */
 package com.alibaba.nacos.naming.healthcheck;
 
+import com.alibaba.nacos.api.naming.pojo.AbstractHealthChecker;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
 /**
  * @author nkorange
  */
 public enum HealthCheckType {
     /**
      * TCP type
      */
-    TCP,
+    TCP("tcp", AbstractHealthChecker.Tcp.class),
     /**
      * HTTP type
      */
-    HTTP,
+    HTTP("http", AbstractHealthChecker.Http.class),
     /**
      * MySQL type
      */
-    MYSQL,
+    MYSQL("mysql", AbstractHealthChecker.Mysql.class),
     /**
      * No check
      */
-    NONE
+    NONE("none", AbstractHealthChecker.None.class);
+
+    private String name;
+
+    private Class healthCheckerClass;
+
+    private static Map<String, Class> EXTEND =
+        new ConcurrentHashMap<>();
+
+    HealthCheckType(String name, Class healthCheckerClass) {
+        this.name = name;
+        this.healthCheckerClass = healthCheckerClass;
+    }
+
+    public static void registerHealthChecker(String type, Class healthCheckerClass){
+        EXTEND.putIfAbsent(type, healthCheckerClass);
+    }
+
+    public static Class ofHealthCheckerClass(String type){
+        return valueOf(type) == null ? EXTEND.get(type) : valueOf(type).healthCheckerClass;
+    }
 }

naming/src/main/java/com/alibaba/nacos/naming/healthcheck/HealthCheckType.java

@@ -42,25 +42,16 @@ public static JsonAdapter getInstance() {
         return INSTANCE;
     }
 
+    @SuppressWarnings("unchecked")
     @Override
     public <T> T deserialze(DefaultJSONParser parser, Type type, Object fieldName) {
         JSONObject jsonObj = (JSONObject) parser.parse();
         String checkType = jsonObj.getString("type");
 
-        if (StringUtils.equals(checkType, AbstractHealthChecker.Http.TYPE)) {
-            return (T) JSON.parseObject(jsonObj.toJSONString(), AbstractHealthChecker.Http.class);
-        }
-
-        if (StringUtils.equals(checkType, AbstractHealthChecker.Tcp.TYPE)) {
-            return (T) JSON.parseObject(jsonObj.toJSONString(), AbstractHealthChecker.Tcp.class);
-        }
-
-        if (StringUtils.equals(checkType, AbstractHealthChecker.None.TYPE)) {
-            return (T) JSON.parseObject(jsonObj.toJSONString(), AbstractHealthChecker.None.class);
-        }
+        Class target = HealthCheckType.ofHealthCheckerClass(checkType);
 
-        if (StringUtils.equals(checkType, AbstractHealthChecker.Mysql.TYPE)) {
-            return (T) JSON.parseObject(jsonObj.toJSONString(), AbstractHealthChecker.Mysql.class);
+        if(target != null){
+            return (T) JSON.parseObject(jsonObj.toJSONString(), target);
         }
 
         return null;
@@ -83,21 +74,6 @@ public void write(JSONSerializer jsonSerializer, Object o, Object o1, Type type,
 
         writer.writeFieldValue(',', "type", config.getType());
 
-        if (StringUtils.equals(config.getType(), HealthCheckType.HTTP.name())) {
-            AbstractHealthChecker.Http httpCheckConfig = (AbstractHealthChecker.Http) config;
-            writer.writeFieldValue(',', "path", httpCheckConfig.getPath());
-            writer.writeFieldValue(',', "headers", httpCheckConfig.getHeaders());
-        }
-
-        if (StringUtils.equals(config.getType(), HealthCheckType.TCP.name())) {
-            // nothing sepcial to handle
-        }
-
-        if (StringUtils.equals(config.getType(), HealthCheckType.MYSQL.name())) {
-            AbstractHealthChecker.Mysql mysqlCheckConfig = (AbstractHealthChecker.Mysql) config;
-            writer.writeFieldValue(',', "user", mysqlCheckConfig.getUser());
-            writer.writeFieldValue(',', "pwd", mysqlCheckConfig.getPwd());
-            writer.writeFieldValue(',', "cmd", mysqlCheckConfig.getCmd());
-        }
+        config.jsonAdapterCallback(writer);
     }
 }

naming/src/main/java/com/alibaba/nacos/naming/healthcheck/JsonAdapter.java

@@ -0,0 +1,77 @@
+package com.alibaba.nacos.naming.healthcheck.extend;
+
+import com.alibaba.nacos.api.naming.pojo.AbstractHealthChecker;
+import com.alibaba.nacos.naming.healthcheck.HealthCheckProcessor;
+import com.alibaba.nacos.naming.healthcheck.HealthCheckType;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.beans.factory.config.SingletonBeanRegistry;
+import org.springframework.stereotype.Component;
+
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.ServiceLoader;
+import java.util.Set;
+
+/**
+ * @author XCXCXCXCX
+ */
+@Component
+public class HealthCheckExtendProvider implements BeanFactoryAware{
+
+    private ServiceLoader<HealthCheckProcessor> processorLoader
+        = ServiceLoader.load(HealthCheckProcessor.class);
+
+    private ServiceLoader<AbstractHealthChecker> checkerLoader
+        = ServiceLoader.load(AbstractHealthChecker.class);
+
+    private SingletonBeanRegistry registry;
+
+    private static final char LOWER_A = 'A';
+    private static final char LOWER_Z = 'Z';
+
+    public void init(){
+        loadExtend();
+    }
+
+    private void loadExtend() {
+        Iterator<HealthCheckProcessor> processorIt = processorLoader.iterator();
+        Iterator<AbstractHealthChecker> healthCheckerIt = checkerLoader.iterator();
+
+        Set<String> processorType = new HashSet<>();
+        Set<String> healthCheckerType = new HashSet<>();
+        while(processorIt.hasNext()){
+            HealthCheckProcessor processor = processorIt.next();
+            processorType.add(processor.getType());
+            registry.registerSingleton(lowerFirstChar(processor.getClass().getSimpleName()), processor);
+        }
+
+        while(healthCheckerIt.hasNext()){
+            AbstractHealthChecker checker = healthCheckerIt.next();
+            healthCheckerType.add(checker.getType());
+            HealthCheckType.registerHealthChecker(checker.getType(), checker.getClass());
+        }
+        if(!processorType.equals(healthCheckerType)){
+            throw new RuntimeException("An unmatched processor and healthChecker are detected in the extension package.");
+        }
+    }
+
+    private String lowerFirstChar(String simpleName) {
+        if(simpleName == null || "".equals(simpleName)){
+            throw new IllegalArgumentException("can't find extend processor class name");
+        }
+        char[] chars = simpleName.toCharArray();
+        if(chars[0] >= LOWER_A && chars[0] <= LOWER_Z){
+            chars[0] = (char)(chars[0] + 32);
+        }
+        return String.valueOf(chars);
+    }
+
+    @Override
+    public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
+        if(beanFactory instanceof SingletonBeanRegistry){
+            this.registry = (SingletonBeanRegistry) beanFactory;
+        }
+    }
+}