NacosSync雙向復制源碼分析

作者：梁勇 2021-09-06 10:34:48

開發前端

通過開源同步工具NacosSync的分析，對我們實現自定義的同步工具提供參考。文本就同步任務分發與Nacos集群之間、從zk到Nacos的同步源碼做個分析。

[[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

clusterAccessService.insert(clusterDO);

同步任務入庫

請求入口：TaskApi#taskAdd

入庫處理：TaskAccessService#addTask

taskAccessService.addTask(taskDO);

三、同步任務分發

同步任務入庫了，緊著需要任務進行分發。代碼翻到QuerySyncTaskTimer實現了springboot的CommandLineRunner接口。

定時任務調度

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

備注：定時任務每3秒鐘調度一次。

調度任務執行

private class CheckRunningStatusThread implements Runnable { 
 
    @Override 
    public void run() { 
 
        Long start = System.currentTimeMillis(); 
        try { 
           // 注解@1 
            Iterable<TaskDO> taskDOS = taskAccessService.findAll(); 
            taskDOS.forEach(taskDO -> { 
               // 注解@2  
                if ((null != skyWalkerCacheServices.getFinishedTask(taskDO))) { 
                    return; 
                } 
               // 注解@3 
                if (TaskStatusEnum.SYNC.getCode().equals(taskDO.getTaskStatus())) { 
                    eventBus.post(new SyncTaskEvent(taskDO)); 
                    log.info("從數據庫中查詢到一個同步任務，發出一個同步事件:" + taskDO); 
                } 
        // 注解@4 
                if (TaskStatusEnum.DELETE.getCode().equals(taskDO.getTaskStatus())) { 
                    eventBus.post(new DeleteTaskEvent(taskDO)); 
                    log.info("從數據庫中查詢到一個刪除任務，發出一個同步事件:" + taskDO); 
                } 
            }); 
 
        } catch (Exception e) { 
            log.warn("CheckRunningStatusThread Exception", e); 
        } 
    // 注解@5 
        metricsManager.record(MetricsStatisticsType.DISPATCHER_TASK, System.currentTimeMillis() - start); 
    } 
}

注解@1 查詢所有同步任務

注解@2 過濾已完成的任務

注解@3 發布一個同步任務事件SyncTaskEvent

注解@4 發布一個刪除任務事件DeleteTaskEvent

注解@5 通過metric統計本次調度任務執行的耗時情況

小結：當有新增任務或者刪除任務時通過Guava的EventBus發布一個同步事件或刪除事件，該檢測3秒執行一次。

四、同步事件處理

代碼EventListener#listenerSyncTaskEvent訂閱了同步事件SyncTaskEvent。

@Subscribe 
public void listenerSyncTaskEvent(SyncTaskEvent syncTaskEvent) { 
 
    try { 
        long start = System.currentTimeMillis(); 
       // 注解@6 
        if (syncManagerService.sync(syncTaskEvent.getTaskDO())) {     
           // 注解@7 
            skyWalkerCacheServices.addFinishedTask(syncTaskEvent.getTaskDO()); 
           // 注解@8 
            metricsManager.record(MetricsStatisticsType.SYNC_TASK_RT, System.currentTimeMillis() - start); 
        } else { 
            log.warn("listenerSyncTaskEvent sync failure"); 
        }                 
    } catch (Exception e) { 
        log.warn("listenerSyncTaskEvent process error", e); 
    } 
 
}

注解@6 執行同步任務

注解@7 標記該同步任務完成

注解@8 記錄任務執行時間

代碼EventListener#listenerDeleteTaskEvent訂閱了刪除任務事件DeleteTaskEvent。

@Subscribe 
public void listenerDeleteTaskEvent(DeleteTaskEvent deleteTaskEvent) { 
 
    try { 
        long start = System.currentTimeMillis(); 
        if (syncManagerService.delete(deleteTaskEvent.getTaskDO())) { 
            skyWalkerCacheServices.addFinishedTask(deleteTaskEvent.getTaskDO()); 
            metricsManager.record(MetricsStatisticsType.DELETE_TASK_RT, System.currentTimeMillis() - start); 
        } else { 
            log.warn("listenerDeleteTaskEvent delete failure"); 
        }                 
    } catch (Exception e) { 
        log.warn("listenerDeleteTaskEvent process error", e); 
    } 
 
}

小結： listenerSyncTaskEvent和listenerDeleteTaskEvent代碼結構一致，執行任務邏輯，執行完緩存已完成任務，最后記錄耗時情況。

五、Nacos集群之間同步邏輯

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

執行同步邏輯

@Override 
public boolean sync(TaskDO taskDO) { 
  String taskId = taskDO.getTaskId(); 
  try { 
    // 注解@7 
    NamingService sourceNamingService = 
      nacosServerHolder.get(taskDO.getSourceClusterId(), taskDO.getNameSpace()); 
 
    // 注解@8 
    NamingService destNamingService = nacosServerHolder.get(taskDO.getDestClusterId(), taskDO.getNameSpace()); 
 
 
    this.listenerMap.putIfAbsent(taskId, event -> { 
      if (event instanceof NamingEvent) { 
        try { 
          // 注解@9 
          List<Instance> sourceInstances = sourceNamingService.getAllInstances(taskDO.getServiceName(), 
                                                                               getGroupNameOrDefault(taskDO.getGroupName()), new ArrayList<>(), true); 
 
          // 注解@10 
          this.removeInvalidInstance(taskDO, destNamingService, sourceInstances); 
 
          // 注解@11 
          if (sourceInstances.isEmpty()) { 
            sourceInstanceSnapshot.remove(taskId); 
            return; 
          } 
 
          // 注解@12 
          this.syncNewInstance(taskDO, destNamingService, sourceInstances); 
        } catch (Exception e) { 
          log.error("event process fail, taskId:{}", taskId, e); 
          metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); 
        } 
      } 
    }); 
 
    sourceNamingService.subscribe(taskDO.getServiceName(), getGroupNameOrDefault(taskDO.getGroupName()), 
                                  listenerMap.get(taskId)); 
  } catch (Exception e) { 
    log.error("sync task from nacos to nacos was failed, taskId:{}", taskId, e); 
    metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); 
    return false; 
  } 
  return true; 
}

注解@7 創建源集群的NameService

注解@8 創建目標集群的NameService

注解@9 獲取服務注冊的實例

注解@10 先刪除已失效的節點

private void removeInvalidInstance(TaskDO taskDO, NamingService destNamingService, 
    List<Instance> sourceInstances) throws NacosException { 
 
    String taskId = taskDO.getTaskId(); 
    if (this.sourceInstanceSnapshot.containsKey(taskId)) { 
        // 注解@10.1 
        Set<String> oldInstanceKeys = this.sourceInstanceSnapshot.get(taskId); 
        List<String> newInstanceKeys = sourceInstances.stream().map(this::composeInstanceKey) 
            .collect(Collectors.toList()); 
        // 注解@10.2 
        Collection<String> instanceKeys = Collections.subtract(oldInstanceKeys, newInstanceKeys); 
        for (String instanceKey : instanceKeys) { 
            log.info("任務Id:{},移除無效同步實例:{}", taskId, instanceKey); 
            String[] split = instanceKey.split(":", -1); 
            // 注解@10.3 
            destNamingService 
                .deregisterInstance(taskDO.getServiceName(), getGroupNameOrDefault(taskDO.getGroupName()), split[0], 
                    Integer.parseInt(split[1])); 
 
        } 
    } 
}

注解@10.1 緩存的舊節點信息

注解@10.2 從舊節點中剝離出廢棄無效的節點

注解@10.3 將廢棄無效節點注銷

注解@11 如果同步實例已經為空代表該服務所有實例已經下線,清除本地持有快照

注解@12 同步新節實例到目標集群并更新緩存

private void syncNewInstance(TaskDO taskDO, NamingService destNamingService, 
    List<Instance> sourceInstances) throws NacosException { 
    Set<String> latestSyncInstance = new TreeSet<>(); 
    // 再次添加新實例 
    String taskId = taskDO.getTaskId(); 
    // 注解@12.1 
    Set<String> instanceKeys = sourceInstanceSnapshot.get(taskId); 
    // 注解@12.2 
    for (Instance instance : sourceInstances) { 
        if (needSync(instance.getMetadata())) { 
            String instanceKey = composeInstanceKey(instance); 
            // 注解@12.3 
            if (CollectionUtils.isEmpty(instanceKeys) || !instanceKeys.contains(instanceKey)) { 
                destNamingService.registerInstance(taskDO.getServiceName(), 
                    getGroupNameOrDefault(taskDO.getGroupName()), 
                    buildSyncInstance(instance, taskDO)); 
            } 
            // 注解@12.4 
            latestSyncInstance.add(instanceKey); 
        } 
    } 
    if (CollectionUtils.isNotEmpty(latestSyncInstance)) { 
 
        log.info("任務Id:{},已同步實例個數:{}", taskId, latestSyncInstance.size()); 
        // 注解@12.5 
        sourceInstanceSnapshot.put(taskId, latestSyncInstance); 
    } 
}

注解@12.1 緩存的舊節點信息

注解@12.2 遍歷新節點信息

注解@12.3 當新節點信息不為空并且舊節點不存在，則注冊到目標集群

注解@12.4 收集新節點

注解@12.5 更新緩存節點信息

小結：在兩個Nacos集群之間進行同步，同步任務在Service維度(AppId)建立。通過對源集群注冊監聽獲取注冊節點列表，通過剔除無效節點后，將新的節點注冊到目標集群的過程。

執行刪除任務邏輯

代碼翻到NacosSyncToNacosServiceImpl#delete部分

public boolean delete(TaskDO taskDO) { 
    try { 
        NamingService sourceNamingService = 
            nacosServerHolder.get(taskDO.getSourceClusterId(), taskDO.getNameSpace()); 
        NamingService destNamingService = nacosServerHolder.get(taskDO.getDestClusterId(), taskDO.getNameSpace()); 
        // 注解@13 
        sourceNamingService 
            .unsubscribe(taskDO.getServiceName(), getGroupNameOrDefault(taskDO.getGroupName()), 
                listenerMap.remove(taskDO.getTaskId())); 
        sourceInstanceSnapshot.remove(taskDO.getTaskId()); 
 
        // 注解@14 
        List<Instance> sourceInstances = sourceNamingService 
            .getAllInstances(taskDO.getServiceName(), getGroupNameOrDefault(taskDO.getGroupName()), 
                new ArrayList<>(), false); 
        for (Instance instance : sourceInstances) { 
            if (needSync(instance.getMetadata())) { 
               // 注銷操作 
                destNamingService 
                    .deregisterInstance(taskDO.getServiceName(), getGroupNameOrDefault(taskDO.getGroupName()), 
                        instance.getIp(), 
                        instance.getPort()); 
            } 
        } 
    } catch (Exception e) { 
        log.error("delete task from nacos to nacos was failed, taskId:{}", taskDO.getTaskId(), e); 
        metricsManager.recordError(MetricsStatisticsType.DELETE_ERROR); 
        return false; 
    } 
    return true; 
}

注解@13 移除該任務(service)源集群訂閱

注解@14 刪除目標集群中同步的實例列表

小結：刪除邏輯比較簡單，取消源集群訂閱，將目標集群的注冊節點移除。

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

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

@Override 
public boolean sync(TaskDO taskDO) { 
    try { 
        if (treeCacheMap.containsKey(taskDO.getTaskId())) { 
            return true; 
        } 
        // 注解@1 
        TreeCache treeCache = getTreeCache(taskDO); 
        // 注解@2 
        NamingService destNamingService = nacosServerHolder.get(taskDO.getDestClusterId(), null); 
        // 注解@3 
        registerAllInstances(taskDO, destNamingService); 
        // 注解@4 
        Objects.requireNonNull(treeCache).getListenable().addListener((client, event) -> { 
            try { 
                String path = event.getData().getPath(); 
                Map<String, String> queryParam = parseQueryString(path); 
                if (isMatch(taskDO, queryParam) && needSync(queryParam)) { 
                    processEvent(taskDO, destNamingService, event, path, queryParam); 
                } 
            } catch (Exception e) { 
                // ... 
            } 
        }); 
    } catch (Exception e) { 
        // ... 
        metricsManager.recordError(MetricsStatisticsType.SYNC_ERROR); 
        return false; 
    } 
    return true; 
}

注解@1 監聽zk源集群路徑為「/dubbo」

注解@2 目標Nacos集群構建

注解@3 初次執行任務統一注冊所有實例

private void registerAllInstances(TaskDO taskDO, NamingService destNamingService) throws Exception { 
    CuratorFramework zk =  zookeeperServerHolder.get(taskDO.getSourceClusterId(), ""); 
    // 注解@3.1 
    if(!ALL_SERVICE_NAME_PATTERN.equals(taskDO.getServiceName())) { 
        registerALLInstances0(taskDO, destNamingService, zk, taskDO.getServiceName()); 
    } else { 
        // 注解@3.2 
        List<String> serviceList = zk.getChildren().forPath(DUBBO_ROOT_PATH); 
        for(String serviceName : serviceList) { 
            registerALLInstances0(taskDO, destNamingService, zk, serviceName); 
        } 
    } 
}

注解@3.1 同步特定服務注冊節點(Dubbo)

注解@3.2 同步全部所有的zk節點到Nacos

注解@4 注冊zk監聽監聽新增和更新的同步

private void processEvent(TaskDO taskDO, NamingService destNamingService, TreeCacheEvent event, String path, 
                          Map<String, String> queryParam) throws NacosException { 
    if(!com.alibaba.nacossync.util.StringUtils.isDubboProviderPath(path)) { 
        return; 
    } 
 
    Map<String, String> ipAndPortParam = parseIpAndPortString(path); 
    Instance instance = buildSyncInstance(queryParam, ipAndPortParam, taskDO); 
    String serviceName = queryParam.get(INTERFACE_KEY); 
    switch (event.getType()) { 
        case NODE_ADDED: 
        case NODE_UPDATED: 
            // 注解@4.1 
            destNamingService.registerInstance( 
                getServiceNameFromCache(serviceName, queryParam), instance); 
            break; 
        case NODE_REMOVED: 
            // 注解@4.2 
            destNamingService.deregisterInstance( 
                getServiceNameFromCache(serviceName, queryParam), 
                ipAndPortParam.get(INSTANCE_IP_KEY), 
                Integer.parseInt(ipAndPortParam.get(INSTANCE_PORT_KEY))); 
            nacosServiceNameMap.remove(serviceName); 
            break; 
        default: 
            break; 
    } 
}

注解@4.1 同步節點新增更新到目標集群

注解@4.2 源集群節點被刪除同步注銷目標集群

小結： NacosSync從zk集群同步到Nacos只支持dubbo路徑，可參考基于二次改造。第一次先同步所有節點過去，再監聽源集群路徑變化，同步到目標集群。

本文轉載自微信公眾號「瓜農老梁」，可以通過以下二維碼關注。轉載本文請聯系瓜農老梁公眾號。

責任編輯：武曉燕來源：瓜農老梁

Nacos 復制源碼

成人免费xxxxx在线视频软件_久久精品久久久_亚洲国产精品久久久_天天色天天色_亚洲人成一区_欧美一级欧美三级在线观看