support batchRegisterInstance of nacos2.x for develop branch

README.md

@@ -99,7 +99,7 @@ The path to the target file:
 
 ``` xml
 
-nacos-sync/nacossync-distribution/target/nacos-sync-0.4.8.tar.gz
+nacos-sync/nacossync-distribution/target/nacos-sync-0.4.9-batchRegister.tar.gz
 
 ```
 

nacossync-console/src/main/resources/static/console-fe/src/containers/Layout/Layout.js

@@ -13,7 +13,7 @@ class Layout extends React.Component {
         <Header />
         <Row className="layout">
           <Col fixedSpan="9" className="nav-bar">
-            <h1 className="title">Nacos-Sync 0.4.8</h1>
+            <h1 className="title">Nacos-Sync 0.4.9-batchRegister</h1>
             <Menu />
           </Col>
           <Col className="main-panel">{this.props.children}</Col>

nacossync-worker/src/main/java/com/alibaba/nacossync/dao/ClusterAccessService.java

@@ -107,7 +107,12 @@ private List<Predicate> getPredicates(Root<ClusterDO> root, CriteriaBuilder crit
     public int findClusterLevel(String sourceClusterId){
         ClusterDO clusterDO = clusterRepository.findByClusterId(sourceClusterId);
         if (clusterDO != null) {
-            return clusterDO.getClusterLevel();
+            Integer level = clusterDO.getClusterLevel();
+            if (level == null) {
+                //此字段未设置时取默认0按普通集群处理（目前控制台新增集群的代码并未设置此字段）
+                return 0;
+            }
+            return level;
         }
         return -1;
     }

nacossync-worker/src/main/java/com/alibaba/nacossync/extension/holder/NacosServerHolder.java

@@ -13,24 +13,30 @@
 package com.alibaba.nacossync.extension.holder;
 
 import com.alibaba.nacos.api.PropertyKeyConst;
+import com.alibaba.nacos.api.config.ConfigFactory;
+import com.alibaba.nacos.api.config.ConfigService;
 import com.alibaba.nacos.api.exception.NacosException;
 import com.alibaba.nacos.api.naming.NamingFactory;
 import com.alibaba.nacos.api.naming.NamingService;
+import com.alibaba.nacos.client.naming.NacosNamingService;
+import com.alibaba.nacos.client.naming.remote.NamingClientProxyDelegate;
+import com.alibaba.nacos.client.naming.remote.http.NamingHttpClientProxy;
 import com.alibaba.nacossync.constant.SkyWalkerConstants;
 import com.alibaba.nacossync.dao.ClusterAccessService;
 import com.alibaba.nacossync.dao.TaskAccessService;
 import com.alibaba.nacossync.pojo.model.ClusterDO;
 import com.alibaba.nacossync.pojo.model.TaskDO;
 import com.google.common.base.Joiner;
+import java.lang.reflect.Field;
 import java.util.List;
 import java.util.Optional;
 import java.util.Properties;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Supplier;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.logging.log4j.util.Strings;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
+import org.springframework.util.ReflectionUtils;
 
 /**
  * @author paderlol
@@ -44,7 +50,11 @@ public class NacosServerHolder extends AbstractServerHolderImpl<NamingService> {
 
     private final TaskAccessService taskAccessService;
 
-    private static ConcurrentHashMap<String,NamingService> globalNameService = new ConcurrentHashMap<>(16);
+    private static ConcurrentHashMap<String,NamingService> globalNamingService = new ConcurrentHashMap<>(16);
+
+    private static ConcurrentHashMap<String,NamingHttpClientProxy> globalNamingHttpProxy = new ConcurrentHashMap<>(16);
+
+    private static ConcurrentHashMap<String,ConfigService> globalConfigService = new ConcurrentHashMap<>(16);
 
     public NacosServerHolder(ClusterAccessService clusterAccessService, TaskAccessService taskAccessService) {
         this.clusterAccessService = clusterAccessService;
@@ -68,29 +78,68 @@ NamingService createServer(String clusterId, Supplier<String> serverAddressSuppl
         String serverList = Joiner.on(",").join(allClusterConnectKey);
         Properties properties = new Properties();
         properties.setProperty(PropertyKeyConst.SERVER_ADDR, serverList);
-        properties.setProperty(PropertyKeyConst.NAMESPACE, Optional.ofNullable(clusterDO.getNamespace()).orElse(
-            Strings.EMPTY));
         Optional.ofNullable(clusterDO.getUserName()).ifPresent(value ->
             properties.setProperty(PropertyKeyConst.USERNAME, value)
         );
 
         Optional.ofNullable(clusterDO.getPassword()).ifPresent(value ->
             properties.setProperty(PropertyKeyConst.PASSWORD, value)
         );
-        NamingService namingService = NamingFactory.createNamingService(properties);
-        globalNameService.put(clusterId,namingService);
-        return namingService;
+
+        // configService不能设置namespace,否则获取不到dubbo服务元数据
+        globalConfigService.computeIfAbsent(newClusterId, id -> {
+            try {
+                return ConfigFactory.createConfigService(properties);
+            } catch (NacosException e) {
+                log.error("start config service fail,clusterId:{}", id, e);
+                return null;
+            }
+        });
+
+        properties.setProperty(PropertyKeyConst.NAMESPACE, Optional.ofNullable(clusterDO.getNamespace()).orElse(
+                Strings.EMPTY));
+        return globalNamingService.computeIfAbsent(newClusterId, id -> {
+            try {
+                NacosNamingService namingService = (NacosNamingService) NamingFactory.createNamingService(properties);
+
+                // clientProxy
+                final Field clientProxyField = ReflectionUtils.findField(NacosNamingService.class, "clientProxy");
+                assert clientProxyField != null;
+                ReflectionUtils.makeAccessible(clientProxyField);
+                NamingClientProxyDelegate clientProxy = (NamingClientProxyDelegate) ReflectionUtils.getField(clientProxyField, namingService);
+
+                // httpClientProxy
+                final Field httpClientProxyField = ReflectionUtils.findField(NamingClientProxyDelegate.class, "httpClientProxy");
+                assert httpClientProxyField != null;
+                ReflectionUtils.makeAccessible(httpClientProxyField);
+                NamingHttpClientProxy httpClientProxy = (NamingHttpClientProxy) ReflectionUtils.getField(httpClientProxyField, clientProxy);
+                globalNamingHttpProxy.put(id, httpClientProxy);
+
+                return namingService;
+            } catch (NacosException e) {
+                log.error("start naming service fail,clusterId:{}", id, e);
+                return null;
+            }
+        });
     }
 
     /**
      * Get NamingService for different clients
      * @param clusterId clusterId
      * @return Returns Naming Service objects for different clusters
      */
-    public NamingService getNameService(String clusterId){
-        return globalNameService.get(clusterId);
+    public NamingService getNamingService(String clusterId){
+        return globalNamingService.get(clusterId);
     }
-
+
+    public NamingHttpClientProxy getNamingHttpProxy(String clusterId){
+        return globalNamingHttpProxy.get(clusterId);
+    }
+
+    public ConfigService getConfigService(String clusterId) {
+        return globalConfigService.get(clusterId);
+    }
+
     public NamingService getSourceNamingService(String taskId, String sourceClusterId) {
         String key = taskId + sourceClusterId;
         return serviceMap.computeIfAbsent(key, k->{

nacossync-worker/src/main/java/com/alibaba/nacossync/extension/impl/ConsulSyncToNacosServiceImpl.java

@@ -15,6 +15,7 @@
 import com.alibaba.nacos.api.exception.NacosException;
 import com.alibaba.nacos.api.naming.NamingService;
 import com.alibaba.nacos.api.naming.pojo.Instance;
+import com.alibaba.nacos.common.utils.CollectionUtils;
 import com.alibaba.nacossync.cache.SkyWalkerCacheServices;
 import com.alibaba.nacossync.constant.ClusterTypeEnum;
 import com.alibaba.nacossync.constant.MetricsStatisticsType;
@@ -32,6 +33,8 @@
 import com.ecwid.consul.v1.QueryParams;
 import com.ecwid.consul.v1.Response;
 import com.ecwid.consul.v1.health.model.HealthService;
+
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -77,16 +80,20 @@ public boolean delete(TaskDO taskDO) {
             specialSyncEventBus.unsubscribe(taskDO);
 
             NamingService destNamingService = nacosServerHolder.get(taskDO.getDestClusterId());
-            List<Instance> allInstances = destNamingService.getAllInstances(taskDO.getServiceName(),
-                NacosUtils.getGroupNameOrDefault(taskDO.getGroupName()));
+            String servideName = taskDO.getServiceName();
+            String groupName = NacosUtils.getGroupNameOrDefault(taskDO.getGroupName());
+            List<Instance> allInstances = destNamingService.getAllInstances(servideName, groupName);
+            List<Instance> needDeregisterInstances = new ArrayList<>();
             for (Instance instance : allInstances) {
                 if (needDelete(instance.getMetadata(), taskDO)) {
-
-                    destNamingService.deregisterInstance(taskDO.getServiceName(),
-                        NacosUtils.getGroupNameOrDefault(taskDO.getGroupName()), instance.getIp(), instance.getPort());
+                    NacosSyncToNacosServiceImpl.removeUnwantedAttrsForNacosRedo(instance);
+                    log.debug("需要反注册的实例: {}", instance);
+                    needDeregisterInstances.add(instance);
                 }
             }
-
+            if (CollectionUtils.isNotEmpty(needDeregisterInstances)) {
+                NacosSyncToNacosServiceImpl.doDeregisterInstance(taskDO, destNamingService, servideName, groupName, needDeregisterInstances);
+            }
         } catch (Exception e) {
             log.error("delete task from consul to nacos was failed, taskId:{}", taskDO.getTaskId(), e);
             metricsManager.recordError(MetricsStatisticsType.DELETE_ERROR);
@@ -117,26 +124,47 @@ public boolean sync(TaskDO taskDO, Integer index) {
 
     private void cleanAllOldInstance(TaskDO taskDO, NamingService destNamingService, Set<String> instanceKeys)
         throws NacosException {
-        List<Instance> allInstances = destNamingService.getAllInstances(taskDO.getServiceName());
+        String serviceName = taskDO.getServiceName();
+        String groupName = NacosUtils.getGroupNameOrDefault(taskDO.getGroupName());
+        List<Instance> allInstances = destNamingService.getAllInstances(serviceName, groupName);
+        List<Instance> needDeregisterInstances = new ArrayList<>();
         for (Instance instance : allInstances) {
             if (needDelete(instance.getMetadata(), taskDO)
                 && !instanceKeys.contains(composeInstanceKey(instance.getIp(), instance.getPort()))) {
-
-                destNamingService.deregisterInstance(taskDO.getServiceName(),
-                    NacosUtils.getGroupNameOrDefault(taskDO.getGroupName()), instance.getIp(), instance.getPort());
+                NacosSyncToNacosServiceImpl.removeUnwantedAttrsForNacosRedo(instance);
+                log.debug("需要反注册的实例: {}", instance);
+                needDeregisterInstances.add(instance);
             }
         }
+        if (CollectionUtils.isNotEmpty(needDeregisterInstances)) {
+            NacosSyncToNacosServiceImpl.doDeregisterInstance(taskDO, destNamingService, serviceName, groupName, needDeregisterInstances);
+        }
     }
 
     private void overrideAllInstance(TaskDO taskDO, NamingService destNamingService,
         List<HealthService> healthServiceList, Set<String> instanceKeys) throws NacosException {
+        String serviceName = taskDO.getServiceName();
+        String groupName = NacosUtils.getGroupNameOrDefault(taskDO.getGroupName());
+        List<Instance> needRegisterInstances = new ArrayList<>();
         for (HealthService healthService : healthServiceList) {
             if (needSync(ConsulUtils.transferMetadata(healthService.getService().getTags()))) {
-                destNamingService.registerInstance(taskDO.getServiceName(),
-                    NacosUtils.getGroupNameOrDefault(taskDO.getGroupName()),
-                    buildSyncInstance(healthService, taskDO));
+                Instance syncInstance = buildSyncInstance(healthService, taskDO);
+                log.debug("需要从源集群同步到目标集群的实例：{}", syncInstance);
+                needRegisterInstances.add(syncInstance);
                 instanceKeys.add(composeInstanceKey(healthService.getService().getAddress(),
-                    healthService.getService().getPort()));
+                        healthService.getService().getPort()));
+            }
+        }
+        if (CollectionUtils.isNotEmpty(needRegisterInstances)) {
+            if (needRegisterInstances.get(0).isEphemeral()) {
+                //批量注册
+                log.debug("将源集群指定service的临时实例全量同步到目标集群: {}", taskDO);
+                destNamingService.batchRegisterInstance(serviceName, groupName, needRegisterInstances);
+            } else {
+                for (Instance instance : needRegisterInstances) {
+                    log.debug("从源集群同步到目标集群的持久实例：{}", instance);
+                    destNamingService.registerInstance(serviceName, groupName, instance);
+                }
             }
         }
     }

nacossync-worker/src/main/java/com/alibaba/nacossync/extension/impl/EurekaSyncToNacosServiceImpl.java

@@ -29,6 +29,8 @@
 import com.alibaba.nacossync.pojo.model.TaskDO;
 import com.alibaba.nacossync.util.NacosUtils;
 import com.netflix.appinfo.InstanceInfo;
+
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -118,13 +120,29 @@ public boolean sync(TaskDO taskDO,Integer index) {
 
     private void addValidInstance(TaskDO taskDO, NamingService destNamingService, List<InstanceInfo> eurekaInstances)
         throws NacosException {
+        String serviceName = taskDO.getServiceName();
+        String groupName = NacosUtils.getGroupNameOrDefault(taskDO.getGroupName());
+        List<Instance> needRegisterInstances = new ArrayList<>();
         for (InstanceInfo instance : eurekaInstances) {
             if (needSync(instance.getMetadata())) {
                 log.info("Add service instance from Eureka, serviceName={}, Ip={}, port={}",
-                    instance.getAppName(), instance.getIPAddr(), instance.getPort());
-                destNamingService.registerInstance(taskDO.getServiceName(),
-                    NacosUtils.getGroupNameOrDefault(taskDO.getGroupName()), buildSyncInstance(instance,
-                        taskDO));
+                        instance.getAppName(), instance.getIPAddr(), instance.getPort());
+                Instance syncInstance = buildSyncInstance(instance, taskDO);
+                log.debug("需要从源集群同步到目标集群的实例：{}", syncInstance);
+                needRegisterInstances.add(syncInstance);
+            }
+        }
+        if (CollectionUtils.isEmpty(needRegisterInstances)) {
+            return;
+        }
+        if (needRegisterInstances.get(0).isEphemeral()) {
+            // 批量注册
+            log.debug("将源集群指定service的临时实例全量同步到目标集群: {}", taskDO);
+            destNamingService.batchRegisterInstance(serviceName, groupName, needRegisterInstances);
+        } else {
+            for (Instance instance : needRegisterInstances) {
+                log.debug("从源集群同步到目标集群的持久实例：{}", instance);
+                destNamingService.registerInstance(serviceName, groupName, instance);
             }
         }
     }
@@ -135,26 +153,40 @@ private void deleteAllInstanceFromEureka(TaskDO taskDO, NamingService destNaming
         if (CollectionUtils.isEmpty(eurekaInstances)) {
             return;
         }
+        List<Instance> needDeregisterInstances = new ArrayList<>();
         for (InstanceInfo instance : eurekaInstances) {
             if (needSync(instance.getMetadata())) {
                 log.info("Delete service instance from Eureka, serviceName={}, Ip={}, port={}",
                     instance.getAppName(), instance.getIPAddr(), instance.getPort());
-                destNamingService.deregisterInstance(taskDO.getServiceName(),
-                    NacosUtils.getGroupNameOrDefault(taskDO.getGroupName()), buildSyncInstance(instance, taskDO));
+                Instance needDeregisterInstance = buildSyncInstance(instance, taskDO);
+                log.debug("需要反注册的实例: {}", needDeregisterInstance);
+                needDeregisterInstances.add(needDeregisterInstance);
             }
         }
+        if (CollectionUtils.isEmpty(needDeregisterInstances)) {
+            return;
+        }
+        NacosSyncToNacosServiceImpl.doDeregisterInstance(taskDO, destNamingService, taskDO.getServiceName(),
+                NacosUtils.getGroupNameOrDefault(taskDO.getGroupName()), needDeregisterInstances);
     }
 
     private void removeInvalidInstance(TaskDO taskDO, NamingService destNamingService,
         List<InstanceInfo> eurekaInstances, List<Instance> nacosInstances) throws NacosException {
+        List<Instance> needDeregisterInstances = new ArrayList<>();
         for (Instance instance : nacosInstances) {
             if (!isExistInEurekaInstance(eurekaInstances, instance) && needDelete(instance.getMetadata(), taskDO)) {
                 log.info("Remove invalid service instance from Nacos, serviceName={}, Ip={}, port={}",
                     instance.getServiceName(), instance.getIp(), instance.getPort());
-                destNamingService.deregisterInstance(taskDO.getServiceName(),
-                    NacosUtils.getGroupNameOrDefault(taskDO.getGroupName()), instance.getIp(), instance.getPort());
+                NacosSyncToNacosServiceImpl.removeUnwantedAttrsForNacosRedo(instance);
+                log.debug("需要反注册的实例: {}", instance);
+                needDeregisterInstances.add(instance);
             }
         }
+        if (CollectionUtils.isEmpty(needDeregisterInstances)) {
+            return;
+        }
+        NacosSyncToNacosServiceImpl.doDeregisterInstance(taskDO, destNamingService, taskDO.getServiceName(),
+                NacosUtils.getGroupNameOrDefault(taskDO.getGroupName()), needDeregisterInstances);
     }
 
     private boolean isExistInEurekaInstance(List<InstanceInfo> eurekaInstances, Instance nacosInstance) {
@@ -165,20 +197,26 @@ private boolean isExistInEurekaInstance(List<InstanceInfo> eurekaInstances, Inst
 
     private void deleteAllInstance(TaskDO taskDO, NamingService destNamingService, List<Instance> allInstances)
         throws NacosException {
+        List<Instance> needDeregisterInstances = new ArrayList<>();
         for (Instance instance : allInstances) {
             if (needDelete(instance.getMetadata(), taskDO)) {
-                destNamingService.deregisterInstance(taskDO.getServiceName(),
-                    NacosUtils.getGroupNameOrDefault(taskDO.getGroupName()), instance);
+                NacosSyncToNacosServiceImpl.removeUnwantedAttrsForNacosRedo(instance);
+                log.debug("需要反注册的实例: {}", instance);
+                needDeregisterInstances.add(instance);
             }
-
         }
+        if (CollectionUtils.isEmpty(needDeregisterInstances)) {
+            return;
+        }
+        NacosSyncToNacosServiceImpl.doDeregisterInstance(taskDO, destNamingService, taskDO.getServiceName(),
+                NacosUtils.getGroupNameOrDefault(taskDO.getGroupName()), needDeregisterInstances);
     }
 
     private Instance buildSyncInstance(InstanceInfo instance, TaskDO taskDO) {
         Instance temp = new Instance();
         temp.setIp(instance.getIPAddr());
         temp.setPort(instance.getPort());
-        temp.setServiceName(instance.getAppName());
+        //查询nacos集群实例返回的serviceName含组名前缀，但Nacos2服务端检查批量注册请求serviceName参数时不能包含组名前缀，因此注册实例到目标集群时不再设置serviceName。
         temp.setHealthy(true);
 
         Map<String, String> metaData = new HashMap<>(instance.getMetadata());