nacos数据一致性服务执行流程

数据一致性

其实nacos内部提供两种数据同步方案AP和CP，而且是混用的，只要你的实例是临时的默认用AP，如果是永久的要就用CP。两个数据一致性服务的处理器类结构：

可以看到，左边的RaftConsistencyServiceImpl就是CP的实现类，右边的DistroConsistencyServiceImpl就是AP的实现类，我们注册实例的时候通常是DelegateConsistencyServiceImpl来帮助我们判断该用临时的还是永久的服务，其实他内部就是代理这两个：

具体怎么同步的，我们先来讲AP的TaskScheduler任务同步。

TaskScheduler的run临时数据同步

只要有服务实例的改变，就会让最终一致性服务去做数据同步，其实就是放进一个队列里，然后同步任务从队列中获取要同步的服务key，然后判断数量是不是超过1000或者超时2秒了，是的话就创建SyncTask同步任务并提交，给每一个服务器集群去同步，其实这个属于最终一致性的同步方案。

     @Override
            public void run() {
    
                List<String> keys = new ArrayList<>();
                while (true) {
    
                    try {
                        //阻塞2秒获取key
                        String key = queue.poll(partitionConfig.getTaskDispatchPeriod(),
                            TimeUnit.MILLISECONDS);
                        //不为空，表示获取到了
                        if (Loggers.DISTRO.isDebugEnabled() && StringUtils.isNotBlank(key)) {
                            Loggers.DISTRO.debug("got key: {}", key);
                        }
                        //没有服务器继续
                        if (dataSyncer.getServers() == null || dataSyncer.getServers().isEmpty()) {
                            continue;
                        }
                        //为空继续
                        if (StringUtils.isBlank(key)) {
                            continue;
                        }
                        //还没数据
                        if (dataSize == 0) {
                            keys = new ArrayList<>();
                        }
    
                        keys.add(key);
                        dataSize++;//数据+1
                        //达到分区批量同步数，默认1000个或者同步时间超时，默认2秒
                        if (dataSize == partitionConfig.getBatchSyncKeyCount() ||
                            (System.currentTimeMillis() - lastDispatchTime) > partitionConfig.getTaskDispatchPeriod()) {
                            //获取所有集群服务器
                            for (Server member : dataSyncer.getServers()) {
                                if (NetUtils.localServer().equals(member.getKey())) {//本机不同步
                                    continue;
                                }
                                SyncTask syncTask = new SyncTask();
                               	syncTask.setKeys(keys);//发送的服务key
                                syncTask.setTargetServer(member.getKey());//发送的目标，IP:port
    
                                if (Loggers.DISTRO.isDebugEnabled() && StringUtils.isNotBlank(key)) {
                                    Loggers.DISTRO.debug("add sync task: {}", JSON.toJSONString(syncTask));
                                }
                                //提交同步任务
                                dataSyncer.submit(syncTask, 0);
                            }
                            lastDispatchTime = System.currentTimeMillis();//重置
                            dataSize = 0;
                        }
    
                    } catch (Exception e) {
                        Loggers.DISTRO.error("dispatch sync task failed.", e);
                    }
                }
            }

DataSyncer的submit

提交一个同步任务，如果是新任务的话要去重，删除新的key，因为有可能会有相同的key的任务在重试或者等待重试。然后进行相关key的服务实例集合获取，序列化后同步到目标服务器，如果失败的话还要重试，成功就删除任务返回。

    public void submit(SyncTask task, long delay) {
    
            // If it's a new task:
            if (task.getRetryCount() == 0) {//新的任务
                Iterator<String> iterator = task.getKeys().iterator();
                while (iterator.hasNext()) {//去重
                    String key = iterator.next();
                    if (StringUtils.isNotBlank(taskMap.putIfAbsent(buildKey(key, task.getTargetServer()), key))) {//已经存在的
                        // associated key already exist:
                        if (Loggers.DISTRO.isDebugEnabled()) {
                            Loggers.DISTRO.debug("sync already in process, key: {}", key);
                        }
                        iterator.remove();//删除新的任务里的key，理论上应该是删除旧的，但是此时旧的可能已经在执行了，所以删除新的
                    }
                }
            }
    
            if (task.getKeys().isEmpty()) {
                // all keys are removed:
                return;
            }
            //全局执行器执行数据同步任务
            GlobalExecutor.submitDataSync(() -> {
                // 1. check the server
                if (getServers() == null || getServers().isEmpty()) {
                    Loggers.SRV_LOG.warn("try to sync data but server list is empty.");
                    return;
                }
    
                List<String> keys = task.getKeys();
    
                if (Loggers.SRV_LOG.isDebugEnabled()) {
                    Loggers.SRV_LOG.debug("try to sync data for this keys {}.", keys);
                }
                // 这里又会去重一次，应该keys里可能有重复的，但是里面获取的时候去重了
                Map<String, Datum> datumMap = dataStore.batchGet(keys);
                if (datumMap == null || datumMap.isEmpty()) {//服务实例集合空了就删除相应任务
                    // clear all flags of this task:
                    for (String key : keys) {
                        taskMap.remove(buildKey(key, task.getTargetServer()));
                    }
                    return;
                }
    
                byte[] data = serializer.serialize(datumMap);//服务实例集合序列化
    
                long timestamp = System.currentTimeMillis();
                boolean success = NamingProxy.syncData(data, task.getTargetServer());//发送
                if (!success) {//失败就重试
                    SyncTask syncTask = new SyncTask();
                    syncTask.setKeys(task.getKeys());
                    syncTask.setRetryCount(task.getRetryCount() + 1);
                    syncTask.setLastExecuteTime(timestamp);
                    syncTask.setTargetServer(task.getTargetServer());
                    retrySync(syncTask);
                } else {//成功了就删除任务
                    // clear all flags of this task:
                    for (String key : task.getKeys()) {
                        taskMap.remove(buildKey(key, task.getTargetServer()));
                    }
                }
            }, delay);
        }

好了，今天就到这里了，希望对学习理解有帮助，大神看见勿喷，仅为自己的学习理解，能力有限，请多包涵。