feat: support dual-discovery

Author: fishtailfu

polaris-plugins/polaris-plugins-connector/connector-nacos/src/main/java/com/tencent/polaris/plugins/connector/nacos/NacosConnector.java

@@ -264,12 +264,7 @@ public CommonProviderResponse registerInstance(CommonProviderRequest req,
 
             try {
                 Instance instance = buildRegisterNacosInstance(req);
-                // 优先设置成nacos的service name，如没有再设置成req的service name
-                String serviceName = req.getService();
-                if (StringUtils.isNotEmpty(nacosContext.getServiceName())) {
-                    serviceName = nacosContext.getServiceName();
-                }
-                namingService.registerInstance(serviceName,
+                namingService.registerInstance(instance.getServiceName(),
                         nacosContext.getGroupName(), instance);
                 response.setInstanceID(instance.getInstanceId());
             } catch (NacosException e) {
@@ -376,7 +371,12 @@ public void addLongRunningTask(ServiceUpdateTask serviceUpdateTask) {
     @Override
     protected void doDestroy() {
         if (initialized.compareAndSet(true, false)) {
-            // shutdown naming service
+            // 先unsubscribe listener
+            if (CollectionUtils.isNotEmpty(nacosServices)) {
+                nacosServices.forEach((s, nacosService) -> {
+                    nacosService.destroy();
+                });
+            }
             if (CollectionUtils.isNotEmpty(namingServices)) {
                 namingServices.forEach((s, namingService) -> {
                     try {
@@ -385,11 +385,7 @@ protected void doDestroy() {
                     }
                 });
             }
-            if (CollectionUtils.isNotEmpty(nacosServices)) {
-                nacosServices.forEach((s, nacosService) -> {
-                    nacosService.destroy();
-                });
-            }
+
         }
     }
 

polaris-plugins/polaris-plugins-connector/connector-nacos/src/main/java/com/tencent/polaris/plugins/connector/nacos/NacosService.java

@@ -54,6 +54,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import org.slf4j.Logger;
@@ -70,11 +71,14 @@ public class NacosService extends Destroyable {
 
     private static final int NACOS_SERVICE_PAGESIZE = 10;
 
+    private Map<String, EventListener> eventListeners;
+
     public NacosService(NamingService namingService, NacosContext nacosContext) {
         this.namingService = namingService;
         this.nacosContext = nacosContext;
         NamedThreadFactory threadFactory = new NamedThreadFactory("nacos-service");
         this.refreshExecutor = Executors.newFixedThreadPool(8, threadFactory);
+        this.eventListeners = new ConcurrentHashMap<>();
     }
 
 
@@ -144,7 +148,8 @@ public void asyncGetInstances(ServiceUpdateTask serviceUpdateTask) {
                     newDiscoverResponseBuilder.addAllInstances(polarisInstanceList);
                     int code = ServerCodes.EXECUTE_SUCCESS;
                     newDiscoverResponseBuilder.setCode(UInt32Value.of(code));
-                    LOG.debug("[NacosConnector] Subscribe instances of {} success. ", serviceUpdateTask.getServiceEventKey().getService());
+                    LOG.debug("[NacosConnector] Subscribe instances of {} success. ",
+                            serviceUpdateTask.getServiceEventKey().getService());
                     ServerEvent serverEvent = new ServerEvent(serviceUpdateTask.getServiceEventKey(),
                             newDiscoverResponseBuilder.build(), null, SERVER_CONNECTOR_NACOS);
                     boolean svcDeleted = serviceUpdateTask.notifyServerEvent(serverEvent);
@@ -173,6 +178,8 @@ public void asyncGetInstances(ServiceUpdateTask serviceUpdateTask) {
         try {
             namingService.subscribe(serviceUpdateTask.getServiceEventKey().getService(), nacosContext.getGroupName(),
                     serviceListener);
+            eventListeners.put(serviceUpdateTask.getServiceEventKey().getService(), serviceListener);
+
         } catch (NacosException e) {
             LOG.error("Get nacos service instances of {} failed. ", serviceUpdateTask.getServiceEventKey().getService(),
                     e);
@@ -241,7 +248,8 @@ private void syncGetService(ServiceUpdateTask serviceUpdateTask) {
 
             int code = ServerCodes.EXECUTE_SUCCESS;
             newDiscoverResponseBuilder.setCode(UInt32Value.of(code));
-            LOG.debug("[NacosConnector] get service of {} success. ", serviceUpdateTask.getServiceEventKey().getService());
+            LOG.debug("[NacosConnector] get service of {} success. ",
+                    serviceUpdateTask.getServiceEventKey().getService());
             ServerEvent serverEvent = new ServerEvent(serviceUpdateTask.getServiceEventKey(),
                     newDiscoverResponseBuilder.build(), null, SERVER_CONNECTOR_NACOS);
             boolean svcDeleted = serviceUpdateTask.notifyServerEvent(serverEvent);
@@ -269,6 +277,13 @@ private String buildRevision(List<Instance> instances) throws Exception {
 
     @Override
     protected void doDestroy() {
+        for (Map.Entry<String, EventListener> entry : eventListeners.entrySet()) {
+            try {
+                namingService.unsubscribe(entry.getKey(), nacosContext.getGroupName(), entry.getValue());
+            } catch (NacosException e) {
+                LOG.error("[NacosConnector] unsubscribe service {} in group {} failed. ", entry.getKey(), nacosContext.getGroupName());
+            }
+        }
         ThreadPoolUtils.waitAndStopThreadPools(new ExecutorService[]{refreshExecutor});
     }