NacosSync双向复制源码分析

[[421895]]

引言

通过开源同步工具NacosSync的分析，对我们实现自定义的同步工具提供参考。文本就同步任务分发与Nacos集群之间、从zk到Nacos的同步源码做个分析。

站在用户的角度思考问题，与客户深入沟通，找到乌恰网站设计与乌恰网站推广的解决方案，凭借多年的经验，让设计与互联网技术结合，创造个性化、用户体验好的作品，建站类型包括：成都网站建设、成都网站制作、企业官网、英文网站、手机端网站、网站推广、域名申请、虚拟主机、企业邮箱。业务覆盖乌恰地区。

一、内容提要

任务和配置入库

  • 集群配置入库
  • 同步任务入库

同步任务分发

  • 每三秒调度一次任务列表
  • 新增任务发布同步任务事件SyncTaskEvent并由listenerSyncTaskEvent处理
  • 删除任务发布删除任务事件DeleteTaskEvent并由listenerDeleteTaskEvent处理
  • 任务的发布和订阅使用Guava的EventBus

Nacos集群之间同步逻辑

  • 两个Nacos集群之间进行同步，同步任务在Service维度(AppId)建立
  • 对源集群注册监听获取注册节点列表，通过剔除无效节点后，将新的节点注册到目标集群

从zk集群同步到Nacos集群

  • NacosSync从zk集群同步到Nacos只支持dubbo路径
  • 第一次先同步所有节点过去，再监听源集群路径变化，同步到目标集群

二、任务和配置入库

入库部分比较简单，只列出入口和处理类。

集群配置入库

请求入口：ClusterApi#clusterAdd

入库处理：ClusterAddProcessor#process

 
 
 
 
    
  
  
  
  
  1. clusterAccessService.insert(clusterDO); 
    2.

同步任务入库

请求入口：TaskApi#taskAdd

入库处理：TaskAccessService#addTask

 
 
 
 
    
  
  
  
  
  1. taskAccessService.addTask(taskDO); 
    2.

三、同步任务分发

同步任务入库了，紧着需要任务进行分发。代码翻到QuerySyncTaskTimer实现了springboot的CommandLineRunner接口。

定时任务调度

 
 
 
 
    
  
  
  
  
  1. public void run(String... args) { 
    2.   
  
  
  
  2.  scheduledExecutorService.scheduleWithFixedDelay(new CheckRunningStatusThread(), 0, 3000, 
    3.   
  
  
  
  3.  TimeUnit.MILLISECONDS); 
    4.   
  
  
  
  4. } 
    5.

备注： 定时任务每3秒钟调度一次。

调度任务执行

 
 
 
 
    
  
  
  
  
  1. private class CheckRunningStatusThread implements Runnable { 
    2.   
  
  
  
  2.  
    3.   
  
  
  
  3.     @Override 
    4.   
  
  
  
  4.     public void run() { 
    5.   
  
  
  
  5.  
    6.   
  
  
  
  6.         Long start = System.currentTimeMillis(); 
    7.   
  
  
  
  7.         try { 
    8.   
  
  
  
  8.            // 注解@1 
    9.   
  
  
  
  9.             Iterable taskDOS = taskAccessService.findAll(); 
    10.   
  
  
  
  10.             taskDOS.forEach(taskDO -> { 
    11.   
  
  
  
  11.                // 注解@2  
    12.   
  
  
  
  12.                 if ((null != skyWalkerCacheServices.getFinishedTask(taskDO))) { 
    13.   
  
  
  
  13.                     return; 
    14.   
  
  
  
  14.                 } 
    15.   
  
  
  
  15.                // 注解@3 
    16.   
  
  
  
  16.                 if (TaskStatusEnum.SYNC.getCode().equals(taskDO.getTaskStatus())) { 
    17.   
  
  
  
  17.                     eventBus.post(new SyncTaskEvent(taskDO)); 
    18.   
  
  
  
  18.                     log.info("从数据库中查询到一个同步任务，发出一个同步事件:" + taskDO); 
    19.   
  
  
  
  19.                 } 
    20.   
  
  
  
  20.         // 注解@4 
    21.   
  
  
  
  21.                 if (TaskStatusEnum.DELETE.getCode().equals(taskDO.getTaskStatus())) { 
    22.   
  
  
  
  22.                     eventBus.post(new DeleteTaskEvent(taskDO)); 
    23.   
  
  
  
  23.                     log.info("从数据库中查询到一个删除任务，发出一个同步事件:" + taskDO); 
    24.   
  
  
  
  24.                 } 
    25.   
  
  
  
  25.             }); 
    26.   
  
  
  
  26.  
    27.   
  
  
  
  27.         } catch (Exception e) { 
    28.   
  
  
  
  28.             log.warn("CheckRunningStatusThread Exception", e); 
    29.   
  
  
  
  29.         } 
    30.   
  
  
  
  30.     // 注解@5 
    31.   
  
  
  
  31.         metricsManager.record(MetricsStatisticsType.DISPATCHER_TASK, System.currentTimeMillis() - start); 
    32.   
  
  
  
  32.     } 
    33.   
  
  
  
  33. } 
    34.

注解@1 查询所有同步任务

注解@2 过滤已完成的任务

注解@3 发布一个同步任务事件SyncTaskEvent

注解@4 发布一个删除任务事件DeleteTaskEvent

注解@5 通过metric统计本次调度任务执行的耗时情况

小结： 当有新增任务或者删除任务时通过Guava的EventBus发布一个同步事件或删除事件，该检测3秒执行一次。

四、同步事件处理

代码EventListener#listenerSyncTaskEvent订阅了同步事件SyncTaskEvent。

 
 
 
 
    
  
  
  
  
  1. @Subscribe 
    2.   
  
  
  
  2. public void listenerSyncTaskEvent(SyncTaskEvent syncTaskEvent) { 
    3.   
  
  
  
  3.  
    4.   
  
  
  
  4.     try { 
    5.   
  
  
  
  5.         long start = System.currentTimeMillis(); 
    6.   
  
  
  
  6.        // 注解@6 
    7.   
  
  
  
  7.         if (syncManagerService.sync(syncTaskEvent.getTaskDO())) {     
    8.   
  
  
  
  8.            // 注解@7 
    9.   
  
  
  
  9.             skyWalkerCacheServices.addFinishedTask(syncTaskEvent.getTaskDO()); 
    10.   
  
  
  
  10.            // 注解@8 
    11.   
  
  
  
  11.             metricsManager.record(MetricsStatisticsType.SYNC_TASK_RT, System.currentTimeMillis() - start); 
    12.   
  
  
  
  12.         } else { 
    13.   
  
  
  
  13.             log.warn("listenerSyncTaskEvent sync failure"); 
    14.   
  
  
  
  14.         }                 
    15.   
  
  
  
  15.     } catch (Exception e) { 
    16.   
  
  
  
  16.         log.warn("listenerSyncTaskEvent process error", e); 
    17.   
  
  
  
  17.     } 
    18.   
  
  
  
  18.  
    19.   
  
  
  
  19. } 
    20.

注解@6 执行同步任务

注解@7 标记该同步任务完成

注解@8 记录任务执行时间

代码EventListener#listenerDeleteTaskEvent订阅了删除任务事件DeleteTaskEvent。

 
 
 
 
    
  
  
  
  
  1. @Subscribe 
    2.   
  
  
  
  2. public void listenerDeleteTaskEvent(DeleteTaskEvent deleteTaskEvent) { 
    3.   
  
  
  
  3.  
    4.   
  
  
  
  4.     try { 
    5.   
  
  
  
  5.         long start = System.currentTimeMillis(); 
    6.   
  
  
  
  6.         if (syncManagerService.delete(deleteTaskEvent.getTaskDO())) { 
    7.   
  
  
  
  7.             skyWalkerCacheServices.addFinishedTask(deleteTaskEvent.getTaskDO()); 
    8.   
  
  
  
  8.             metricsManager.record(MetricsStatisticsType.DELETE_TASK_RT, System.currentTimeMillis() - start); 
    9.   
  
  
  
  9.         } else { 
    10.   
  
  
  
  10.             log.warn("listenerDeleteTaskEvent delete failure"); 
    11.   
  
  
  
  11.         }                 
    12.   
  
  
  
  12.     } catch (Exception e) { 
    13.   
  
  
  
  13.         log.warn("listenerDeleteTaskEvent process error", e); 
    14.   
  
  
  
  14.     } 
    15.   
  
  
  
  15.  
    16.   
  
  
  
  16. } 
    17.

小结： listenerSyncTaskEvent和listenerDeleteTaskEvent代码结构一致，执行任务逻辑，执行完缓存已完成任务，最后记录耗时情况。

五、Nacos集群之间同步逻辑

先看下Nacos集群之间的同步，代码在NacosSyncToNacosServiceImpl#sync部分。

执行同步逻辑

 
 
 
 
    
  
  
  
  
  1. @Override 
    2.   
  
  
  
  2. public boolean sync(TaskDO taskDO) { 
    3.   
  
  
  
  3.   String taskId = taskDO.getTaskId(); 
    4.   
  
  
  
  4.   try { 
    5.   
  
  
  
  5.     // 注解@7 
    6.   
  
  
  
  6.     NamingService sourceNamingService = 
    7.   
  
  
  
  7.       nacosServerHolder.get(taskDO.getSourceClusterId(), taskDO.getNameSpace()); 
    8.   
  
  
  
  8.  
    9.   
  
  
  
  9.     // 注解@8 
    10.   
  
  
  
  10.     NamingService destNamingService = nacosServerHolder.get(taskDO.getDestClusterId(), taskDO.getNameSpace()); 
    11.   
  
  
  
  11.  
    12.   
  
  
  
  12.  
    13.   
  
  
  
  13.     this.listenerMap.putIfAbsent(taskId, event -> { 
    14.   
  
  
  
  14.       if (event instanceof NamingEvent) { 
    15.   
  
  
  
  15.         try { 
    16.   
  
  
  
  16.           // 注解@9 
    17.   
  
  
  
  17.           List sourceInstances = sourceNamingService.getAllInstances(taskDO.getServiceName(), 
    18.   
  
  
  
  18.                                                                                getGroupNameOrDefault(taskDO.getGroupName()), new ArrayList<>(), true); 
    19.   
  
  
  
  19.  
    20.   
  
  
  
  20.           // 注解@10 
    21.   
  
  
  
  21.           this.removeInvalidInstance(taskDO, destNamingService, sourceInstances); 
    22.   
  
  
  
  22.  
    23.   
  
  
  
  23.           // 注解@11 
    24.   
  
  
  
  24.           if (sourceInstances.isEmpty()) { 
    25.   
  
  
  
  25.             sourceInstanceSnapshot.remove(taskId); 
    26.   
  
  
  
  26.             return; 
    27.   
  
  
  
  27.           } 
    28.   
  
  
  
  28.  
    29.   
  
  
  
  29.           // 注解@12 
    30.   
  
  
  
  30.           this.syncNewInstance(taskDO, destNamingService, sourceInstances); 
    31.   
  
  
  
  31.         } catch (Exception e) { 
    32.   
  
  
  
  32.           log.error("event process fail, taskId:{}", taskId, e); 
    33.   
  
  
  
  33.           metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); 
    34.   
  
  
  
  34.         } 
    35.   
  
  
  
  35.       } 
    36.   
  
  
  
  36.     }); 
    37.   
  
  
  
  37.  
    38.   
  
  
  
  38.     sourceNamingService.subscribe(taskDO.getServiceName(), getGroupNameOrDefault(taskDO.getGroupName()), 
    39.   
  
  
  
  39.                                   listenerMap.get(taskId)); 
    40.   
  
  
  
  40.   } catch (Exception e) { 
    41.   
  
  
  
  41.     log.error("sync task from nacos to nacos was failed, taskId:{}", taskId, e); 
    42.   
  
  
  
  42.     metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); 
    43.   
  
  
  
  43.     return false; 
    44.   
  
  
  
  44.   } 
    45.   
  
  
  
  45.   return true; 
    46.   
  
  
  
  46. } 
    47.

注解@7 创建源集群的NameService

注解@8 创建目标集群的NameService

注解@9 获取服务注册的实例

注解@10 先删除已失效的节点

 
 
 
 
    
  
  
  
  
  1. private void removeInvalidInstance(TaskDO taskDO, NamingService destNamingService, 
    2.   
  
  
  
  2.     List sourceInstances) throws NacosException { 
    3.   
  
  
  
  3.  
    4.   
  
  
  
  4.     String taskId = taskDO.getTaskId(); 
    5.   
  
  
  
  5.     if (this.sourceInstanceSnapshot.containsKey(taskId)) { 
    6.   
  
  
  
  6.         // 注解@10.1 
    7.   
  
  
  
  7.         Set oldInstanceKeys = this.sourceInstanceSnapshot.get(taskId); 
    8.   
  
  
  
  8.         List newInstanceKeys = sourceInstances.stream().map(this::composeInstanceKey) 
    9.   
  
  
  
  9.             .collect(Collectors.toList()); 
    10.   
  
  
  
  10.         // 注解@10.2 
    11.   
  
  
  
  11.         Collection instanceKeys = Collections.subtract(oldInstanceKeys, newInstanceKeys); 
    12.   
  
  
  
  12.         for (String instanceKey : instanceKeys) { 
    13.   
  
  
  
  13.             log.info("任务Id:{},移除无效同步实例:{}", taskId, instanceKey); 
    14.   
  
  
  
  14.             String[] split = instanceKey.split(":", -1); 
    15.   
  
  
  
  15.             // 注解@10.3 
    16.   
  
  
  
  16.             destNamingService 
    17.   
  
  
  
  17.                 .deregisterInstance(taskDO.getServiceName(), getGroupNameOrDefault(taskDO.getGroupName()), split[0], 
    18.   
  
  
  
  18.                     Integer.parseInt(split[1])); 
    19.   
  
  
  
  19.  
    20.   
  
  
  
  20.         } 
    21.   
  
  
  
  21.     } 
    22.   
  
  
  
  22. } 
    23.

注解@10.1 缓存的旧节点信息

注解@10.2 从旧节点中剥离出废弃无效的节点

注解@10.3 将废弃无效节点注销

注解@11 如果同步实例已经为空代表该服务所有实例已经下线,清除本地持有快照

注解@12 同步新节实例到目标集群并更新缓存

 
 
 
 
    
  
  
  
  
  1. private void syncNewInstance(TaskDO taskDO, NamingService destNamingService, 
    2.   
  
  
  
  2.     List sourceInstances) throws NacosException { 
    3.   
  
  
  
  3.     Set latestSyncInstance = new TreeSet<>(); 
    4.   
  
  
  
  4.     // 再次添加新实例 
    5.   
  
  
  
  5.     String taskId = taskDO.getTaskId(); 
    6.   
  
  
  
  6.     // 注解@12.1 
    7.   
  
  
  
  7.     Set instanceKeys = sourceInstanceSnapshot.get(taskId); 
    8.   
  
  
  
  8.     // 注解@12.2 
    9.   
  
  
  
  9.     for (Instance instance : sourceInstances) { 
    10.   
  
  
  
  10.         if (needSync(instance.getMetadata())) { 
    11.   
  
  
  
  11.             String instanceKey = composeInstanceKey(instance); 
    12.   
  
  
  
  12.             // 注解@12.3 
    13.   
  
  
  
  13.             if (CollectionUtils.isEmpty(instanceKeys) || !instanceKeys.contains(instanceKey)) { 
    14.   
  
  
  
  14.                 destNamingService.registerInstance(taskDO.getServiceName(), 
    15.   
  
  
  
  15.                     getGroupNameOrDefault(taskDO.getGroupName()), 
    16.   
  
  
  
  16.                     buildSyncInstance(instance, taskDO)); 
    17.   
  
  
  
  17.             } 
    18.   
  
  
  
  18.             // 注解@12.4 
    19.   
  
  
  
  19.             latestSyncInstance.add(instanceKey); 
    20.   
  
  
  
  20.         } 
    21.   
  
  
  
  21.     } 
    22.   
  
  
  
  22.     if (CollectionUtils.isNotEmpty(latestSyncInstance)) { 
    23.   
  
  
  
  23.  
    24.   
  
  
  
  24.         log.info("任务Id:{},已同步实例个数:{}", taskId, latestSyncInstance.size()); 
    25.   
  
  
  
  25.         // 注解@12.5 
    26.   
  
  
  
  26.         sourceInstanceSnapshot.put(taskId, latestSyncInstance); 
    27.   
  
  
  
  27.     } 
    28.   
  
  
  
  28. } 
    29.

注解@12.1 缓存的旧节点信息

注解@12.2 遍历新节点信息

注解@12.3 当新节点信息不为空并且旧节点不存在，则注册到目标集群

注解@12.4 收集新节点

注解@12.5 更新缓存节点信息

小结： 在两个Nacos集群之间进行同步，同步任务在Service维度(AppId)建立。通过对源集群注册监听获取注册节点列表，通过剔除无效节点后，将新的节点注册到目标集群的过程。

执行删除任务逻辑

代码翻到NacosSyncToNacosServiceImpl#delete部分

 
 
 
 
    
  
  
  
  
  1. public boolean delete(TaskDO taskDO) { 
    2.   
  
  
  
  2.     try { 
    3.   
  
  
  
  3.         NamingService sourceNamingService = 
    4.   
  
  
  
  4.             nacosServerHolder.get(taskDO.getSourceClusterId(), taskDO.getNameSpace()); 
    5.   
  
  
  
  5.         NamingService destNamingService = nacosServerHolder.get(taskDO.getDestClusterId(), taskDO.getNameSpace()); 
    6.   
  
  
  
  6.         // 注解@13 
    7.   
  
  
  
  7.         sourceNamingService 
    8.   
  
  
  
  8.             .unsubscribe(taskDO.getServiceName(), getGroupNameOrDefault(taskDO.getGroupName()), 
    9.   
  
  
  
  9.                 listenerMap.remove(taskDO.getTaskId())); 
    10.   
  
  
  
  10.         sourceInstanceSnapshot.remove(taskDO.getTaskId()); 
    11.   
  
  
  
  11.  
    12.   
  
  
  
  12.         // 注解@14 
    13.   
  
  
  
  13.         List sourceInstances = sourceNamingService 
    14.   
  
  
  
  14.             .getAllInstances(taskDO.getServiceName(), getGroupNameOrDefault(taskDO.getGroupName()), 
    15.   
  
  
  
  15.                 new ArrayList<>(), false); 
    16.   
  
  
  
  16.         for (Instance instance : sourceInstances) { 
    17.   
  
  
  
  17.             if (needSync(instance.getMetadata())) { 
    18.   
  
  
  
  18.                // 注销操作 
    19.   
  
  
  
  19.                 destNamingService 
    20.   
  
  
  
  20.                     .deregisterInstance(taskDO.getServiceName(), getGroupNameOrDefault(taskDO.getGroupName()), 
    21.   
  
  
  
  21.                         instance.getIp(), 
    22.   
  
  
  
  22.                         instance.getPort()); 
    23.   
  
  
  
  23.             } 
    24.   
  
  
  
  24.         } 
    25.   
  
  
  
  25.     } catch (Exception e) { 
    26.   
  
  
  
  26.         log.error("delete task from nacos to nacos was failed, taskId:{}", taskDO.getTaskId(), e); 
    27.   
  
  
  
  27.         metricsManager.recordError(MetricsStatisticsType.DELETE_ERROR); 
    28.   
  
  
  
  28.         return false; 
    29.   
  
  
  
  29.     } 
    30.   
  
  
  
  30.     return true; 
    31.   
  
  
  
  31. } 
    32.

注解@13 移除该任务(service)源集群订阅

注解@14 删除目标集群中同步的实例列表

小结： 删除逻辑比较简单，取消源集群订阅，将目标集群的注册节点移除。

六、从zk集群同步到Nacos集群

再看从zk集群同步到Nacos集群，代码翻到ZookeeperSyncToNacosServiceImpl#sync()

 
 
 
 
    
  
  
  
  
  1. @Override 
    2.   
  
  
  
  2. public boolean sync(TaskDO taskDO) { 
    3.   
  
  
  
  3.     try { 
    4.   
  
  
  
  4.         if (treeCacheMap.containsKey(taskDO.getTaskId())) { 
    5.   
  
  
  
  5.             return true; 
    6.   
  
  
  
  6.         } 
    7.   
  
  
  
  7.         // 注解@1 
    8.   
  
  
  
  8.         TreeCache treeCache = getTreeCache(taskDO); 
    9.   
  
  
  
  9.         // 注解@2 
    10.   
  
  
  
  10.         NamingService destNamingService = nacosServerHolder.get(taskDO.getDestClusterId(), null); 
    11.   
  
  
  
  11.         // 注解@3 
    12.   
  
  
  
  12.         registerAllInstances(taskDO, destNamingService); 
    13.   
  
  
  
  13.         // 注解@4 
    14.   
  
  
  
  14.         Objects.requireNonNull(treeCache).getListenable().addListener((client, event) -> { 
    15.   
  
  
  
  15.             try { 
    16.   
  
  
  
  16.                 String path = event.getData().getPath(); 
    17.   
  
  
  
  17.                 Map queryParam = parseQueryString(path); 
    18.   
  
  
  
  18.                 if (isMatch(taskDO, queryParam) && needSync(queryParam)) { 
    19.   
  
  
  
  19.                     processEvent(taskDO, destNamingService, event, path, queryParam); 
    20.   
  
  
  
  20.                 } 
    21.   
  
  
  
  21.             } catch (Exception e) { 
    22.   
  
  
  
  22.                 // ... 
    23.   
  
  
  
  23.             } 
    24.   
  
  
  
  24.         }); 
    25.   
  
  
  
  25.     } catch (Exception e) { 
    26.   
  
  
  
  26.         // ... 
    27.   
  
  
  
  27.         metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); 
    28.   
  
  
  
  28.         return false; 
    29.   
  
  
  
  29.     } 
    30.   
  
  
  
  30.     return true; 
    31.   
  
  
  
  31. } 
    32.

注解@1 监听zk源集群 路径为「/dubbo」

注解@2 目标Nacos集群构建

注解@3 初次执行任务统一注册所有实例

 
 
 
 
    
  
  
  
  
  1. private void registerAllInstances(TaskDO taskDO, NamingService destNamingService) throws Exception { 
    2.   
  
  
  
  2.     CuratorFramework zk =  zookeeperServerHolder.get(taskDO.getSourceClusterId(), ""); 
    3.   
  
  
  
  3.     // 注解@3.1 
    4.   
  
  
  
  4.     if(!ALL_SERVICE_NAME_PATTERN.equals(taskDO.getServiceName())) { 
    5.   
  
  
  
  5.         registerALLInstances0(taskDO, destNamingService, zk, taskDO.getServiceName()); 
    6.   
  
  
  
  6.     } else { 
    7.   
  
  
  
  7.         // 注解@3.2 
    8.   
  
  
  
  8.         List serviceList = zk.getChildren().forPath(DUBBO_ROOT_PATH); 
    9.   
  
  
  
  9.         for(String serviceName : serviceList) { 
    10.   
  
  
  
  10.             registerALLInstances0(taskDO, destNamingService, zk, serviceName); 
    11.   
  
  
  
  11.         } 
    12.   
  
  
  
  12.     } 
    13.   
  
  
  
  13. } 
    14.

注解@3.1 同步特定服务注册节点(Dubbo)

注解@3.2 同步全部所有的zk节点到Nacos

注解@4 注册zk监听监听新增和更新的同步

 
 
 
 
    
  
  
  
  
  1. private void processEvent(TaskDO taskDO, NamingService destNamingService, TreeCacheEvent event, String path, 
    2.   
  
  
  
  2.                           Map queryParam) throws NacosException { 
    3.   
  
  
  
  3.     if(!com.alibaba.nacossync.util.StringUtils.isDubboProviderPath(path)) { 
    4.   
  
  
  
  4.         return; 
    5.   
  
  
  
  5.     } 
    6.   
  
  
  
  6.  
    7.   
  
  
  
  7.     Map ipAndPortParam = parseIpAndPortString(path); 
    8.   
  
  
  
  8.     Instance instance = buildSyncInstance(queryParam, ipAndPortParam, taskDO); 
    9.   
  
  
  
  9.     String serviceName = queryParam.get(INTERFACE_KEY); 
    10.   
  
  
  
  10.     switch (event.getType()) { 
    11.   
  
  
  
  11.         case NODE_ADDED: 
    12.   
  
  
  
  12.         case NODE_UPDATED: 
    13.   
  
  
  
  13.             // 注解@4.1 
    14.   
  
  
  
  14.             destNamingService.registerInstance( 
    15.   
  
  
  
  15.                 getServiceNameFromCache(serviceName, queryParam), instance); 
    16.   
  
  
  
  16.             break; 
    17.   
  
  
  
  17.         case NODE_REMOVED: 
    18.   
  
  
  
  18.             // 注解@4.2 
    19.   
  
  
  
  19.             destNamingService.deregisterInstance( 
    20.   
  
  
  
  20.                 getServiceNameFromCache(serviceName, queryParam), 
    21.   
  
  
  
  21.                 ipAndPortParam.get(INSTANCE_IP_KEY), 
    22.   
  
  
  
  22.                 Integer.parseInt(ipAndPortParam.get(INSTANCE_PORT_KEY))); 
    23.   
  
  
  
  23.             nacosServiceNameMap.remove(serviceName); 
    24.   
  
  
  
  24.             break; 
    25.   
  
  
  
  25.         default: 
    26.   
  
  
  
  26.             break; 
    27.   
  
  
  
  27.     } 
    28.   
  
  
  
  28. } 
    29.

注解@4.1 同步节点新增更新到目标集群

注解@4.2 源集群节点被删除同步注销目标集群

小结： NacosSync从zk集群同步到Nacos只支持dubbo路径，可参考基于二次改造。第一次先同步所有节点过去，再监听源集群路径变化，同步到目标集群。

本文转载自微信公众号「瓜农老梁」，可以通过以下二维码关注。转载本文请联系瓜农老梁公众号。

 

新闻名称：NacosSync双向复制源码分析
当前路径：http://www.shufengxianlan.com/qtweb/news11/210911.html

网站建设、网络推广公司-创新互联，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联

猜你还喜欢下面的内容

全网营销推广知识

各行业网站

资阳网站建设　　 　网络推广公司　　 　成都微信小程序开发　　 　绵阳网站建设　　 　名片　　 　四川乐山服务器托管　　 　四川艺考培训　　 　乐山网站建设　　 　酒柜书柜鞋柜定制　　 　成都网站设计　　 　四川平武网站建设　　 　不锈钢防护栏　　 　重庆APP开发　　 　成都高电机柜租用　　 　中国（西部）云计算中心　　 　等级保护测评　　 　腾讯云香港免备案空间　　 　西信服务器托管　　 　广汉锦华建材　　 　成都网站建设公司