createServiceIfAbsent()方法主要作用在于第一次注册进来，从注册表里获取命名空间，肯定是为null，所以需要构建一个命名空间，设置nameSpace等信息并保存到缓存中。这个方法里值得注意的是putServiceAndInit()方法，可以点进来看一下这个方法：

private void putServiceAndInit(Service service) throws NacosException {        //构建注册表双层map，初始化serviceMap --> Map> serviceMap        putService(service);        //初始化service，开启心跳检测的线程        service.init();        //实现数据一致性监听        consistencyService                .listen(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), true), service);        consistencyService                .listen(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), false), service);        Loggers.SRV_LOG.info("[NEW-SERVICE] {}", service.toJson());    }

这里我着重putService(service)方法，这里实际是将注册的实例缓存到内存的注册表中

/** * 通过putService()方法将服务缓存到内存 * * @param service service */public void putService(Service service) {    if (!serviceMap.containsKey(service.getNamespaceId())) {        //双检索防止并发，为了防止同一个服务多个地方同时注册        synchronized (putServiceLock) {            if (!serviceMap.containsKey(service.getNamespaceId())) {                //构建NamespaceId，Serivce对象放到了ServiceMap里面了，也就是说下次我们再调用getService(namespaceId)的时候就可以获取到一个Service对象了                serviceMap.put(service.getNamespaceId(), new ConcurrentSkipListMap<>());            }        }    }    //构建 service name    serviceMap.get(service.getNamespaceId()).put(service.getName(), service);}

接下来我们看一下 putServiceAndInit(Service service)方法中的，init()初始化方法是怎么保持心跳连接的。

/** * service.init()建立心跳机制 */public void init() {    //客户端心跳检查任务，每隔5s执行一次，clientBeatCheckTask是一个线程的方法    HealthCheckReactor.scheduleCheck(clientBeatCheckTask);    for (Map.Entry entry : clusterMap.entrySet()) {        entry.getValue().setService(this);        entry.getValue().init();    }}/** * Schedule client beat check task with a delay. * * @param task client beat check task */public static void scheduleCheck(ClientBeatCheckTask task) {    //客户端的心跳任务，这里并没有嵌套调用，而是开启延迟5s的任务，然后每隔5秒钟执行一次    futureMap.putIfAbsent(task.taskKey(), GlobalExecutor.scheduleNamingHealth(task, 5000, 5000, TimeUnit.MILLISECONDS));}public class ClientBeatCheckTask implements Runnable {        private Service service;        public ClientBeatCheckTask(Service service) {        this.service = service;    }        @JsonIgnore    public PushService getPushService() {        return ApplicationUtils.getBean(PushService.class);    }        @JsonIgnore    public DistroMapper getDistroMapper() {        return ApplicationUtils.getBean(DistroMapper.class);    }        public GlobalConfig getGlobalConfig() {        return ApplicationUtils.getBean(GlobalConfig.class);    }        public SwitchDomain getSwitchDomain() {        return ApplicationUtils.getBean(SwitchDomain.class);    }        public String taskKey() {        return KeyBuilder.buildServiceMetaKey(service.getNamespaceId(), service.getName());    }        @Override    public void run() {        try {            /**             * nacos心跳在集群架构下只允许在一台机器上执行健康检查任务             *             * 集群中有多台机器，本方法在于对服务名称做hash运算再对机器数量取模后，那么             * 这里每次只有定位到一台机器，其他机器都直接return了             *             * 疑问：如果一台机器挂了会怎么办？这里取模会不会乱掉？那这里会不会要做一致性hash？             * 在nacos集群中每台机器之间也是存在状态同步的，每台机器之间都有集群节点同步任务，详见com.alibaba.nacos.naming.cluster.ServerListManager.ServerStatusReporter             *             */            if (!getDistroMapper().responsible(service.getName())) {                return;            }                        if (!getSwitchDomain().isHealthCheckEnabled()) {                return;            }            //获取服务端所有实例            List instances = service.allIPs(true);                        // first set health status of instances:            /**             *  for循环对每个实例都做健康检查             *  在这个方法里面主要是循环当前service的每一个临时实例 用当前时间减去最后一次心跳时间 是否大于心跳超时时间来判断心跳是否超时，             *  如果大于这个时间会执行instance.setHealthy(false)将实例的健康状态改为false；但是这个定时任务不会立即执行，会每5秒执行一次：             */            for (Instance instance : instances) {                //判断心跳是否超时：当前时间 - 实例上次心跳时间 > 心跳的超时时间【默认是15秒】？                if (System.currentTimeMillis() - instance.getLastBeat() > instance.getInstanceHeartBeatTimeOut()) {                    if (!instance.isMarked()) {                        if (instance.isHealthy()) {                            //如果大于心跳默认时间，把实例的 healthy 设置为false【服务列表一开始不会删掉，一开始会变成false】                            instance.setHealthy(false);                            Loggers.EVT_LOG                                    .info("{POS} {IP-DISABLED} valid: {}:{}@{}@{}, region: {}, msg: client timeout after {}, last beat: {}",                                            instance.getIp(), instance.getPort(), instance.getClusterName(),                                            service.getName(), UtilsAndCommons.LOCALHOST_SITE,                                            instance.getInstanceHeartBeatTimeOut(), instance.getLastBeat());                            getPushService().serviceChanged(service);                            ApplicationUtils.publishEvent(new InstanceHeartbeatTimeoutEvent(this, instance));                        }                    }                }            }                        if (!getGlobalConfig().isExpireInstance()) {                return;            }                        // then remove obsolete instances:            for (Instance instance : instances) {                                if (instance.isMarked()) {                    continue;                }                //当前时间 - 实例上一次心跳时间 > 实例的删除时间【默认30s】                if (System.currentTimeMillis() - instance.getLastBeat() > instance.getIpDeleteTimeout()) {                    // delete instance                    Loggers.SRV_LOG.info("[AUTO-DELETE-IP] service: {}, ip: {}", service.getName(),                            JacksonUtils.toJson(instance));                    //直接删除实例                    deleteIp(instance);                }            }                    } catch (Exception e) {            Loggers.SRV_LOG.warn("Exception while processing client beat time out.", e);        }            }        private void deleteIp(Instance instance) {                try {            NamingProxy.Request request = NamingProxy.Request.newRequest();            request.appendParam("ip", instance.getIp()).appendParam("port", String.valueOf(instance.getPort()))                    .appendParam("ephemeral", "true").appendParam("clusterName", instance.getClusterName())                    .appendParam("serviceName", service.getName()).appendParam("namespaceId", service.getNamespaceId());            //调用本地服务            String url = "http://" + IPUtil.localHostIP() + IPUtil.IP_PORT_SPLITER + EnvUtil.getPort() + EnvUtil.getContextPath()                    + UtilsAndCommons.NACOS_NAMING_CONTEXT + "/instance?" + request.toUrl();  //  /v/ns/instance                        // delete instance asynchronously:            HttpClient.asyncHttpDelete(url, null, null, new Callback() {                @Override                public void onReceive(RestResult result) {                    if (!result.ok()) {                        Loggers.SRV_LOG                                .error("[IP-DEAD] failed to delete ip automatically, ip: {}, caused {}, resp code: {}",                                        instance.toJson(), result.getMessage(), result.getCode());                    }                }                    @Override                public void onError(Throwable throwable) {                    Loggers.SRV_LOG                            .error("[IP-DEAD] failed to delete ip automatically, ip: {}, error: {}", instance.toJson(),                                    throwable);                }                    @Override                public void onCancel() {                        }            });                    } catch (Exception e) {            Loggers.SRV_LOG                    .error("[IP-DEAD] failed to delete ip automatically, ip: {}, error: {}", instance.toJson(), e);        }    }}

可以看出init方法是开启了一个异步线程ClientBeatCheckTask去做了个周期性发送心跳的机制，方法中客户端心跳检查任务，开启延迟5s的任务，然后每隔5秒钟执行一次。

service.init()方法主要通过定时任务不断检测当前服务下所有实例最后发送心跳包的时间。在这个方法里面主要是循环当前service的每一个临时实例，用当前时间减去最后一次心跳时间是否大于15s来判断心跳是否超时，如果大于这个时间会执行instance.setHealthy(false)将实例的健康状态改为false，但是这个定时任务不会立即执行，会每5秒执行一次；当前时间 - 实例上一次心跳时间 > 实例的删除时间【默认30s】就会删除实例。

那么服务实例的最后心跳包更新时间是谁来触发的呢?实际上前面在说客户端注册时有说到, Nacos客户端注册服务的同时也建立了心跳机制。

上文中registerInstance注册实例方法中还有一个最最重要的方法就是addInstance()方法，其本质上就是把当前注册的服务实例保存到Service中。

public void addInstance(String namespaceId, String serviceName, boolean ephemeral, Instance... ips)            throws NacosException {                String key = KeyBuilder.buildInstanceListKey(namespaceId, serviceName, ephemeral);                Service service = getService(namespaceId, serviceName);                synchronized (service) {            //将需要注册的实例全部放到Cluster，再将Cluster放在Service里            List instanceList = addIpAddresses(service, ephemeral, ips);                        Instances instances = new Instances();            instances.setInstanceList(instanceList);            //看一下 consistencyService 对象初始化的地方就知道走的是哪个实现            consistencyService.put(key, instances);        }    }    public static String buildInstanceListKey(String namespaceId, String serviceName, boolean ephemeral) {        //根据 ephemeral 取值默认是 true为临时实例，临时实例是存放在内存的；false即为永久实例写到文件的，可以通过此参数区分nacos是AP还是CP架构        return ephemeral                ? buildEphemeralInstanceListKey(namespaceId, serviceName)                : buildPersistentInstanceListKey(namespaceId, serviceName);    }

这里着重看一下这个put方法，put方法主要做了两件事，第一对对客户端的请求过来的实例进行注册，第二是Nacos集群架构下的数据同步，Nacos默认用的是临时实例，也就是ephemeral = true，也就是本文的重点AP架构的Nacos注册原理。

public void put(String key, Record value) throws NacosException {        //注册逻辑：实际就是把实例注册任务放到内存阻塞队列中        onPut(key, value);        //AP 架构下的节点数据同步        distroProtocol.sync(new DistroKey(key, KeyBuilder.INSTANCE_LIST_KEY_PREFIX), DataOperation.CHANGE,                globalConfig.getTaskDispatchPeriod() / 2);    }    /**    * 注册逻辑    */    public void onPut(String key, Record value) {                if (KeyBuilder.matchEphemeralInstanceListKey(key)) {            Datum datum = new Datum<>();            datum.value = (Instances) value;            datum.key = key;            datum.timestamp.incrementAndGet();            //把客户端信息注册信息更新到注册表            dataStore.put(key, datum);        }                if (!listeners.containsKey(key)) {            return;        }        //这里放的是DataOperation.CHANGE        notifier.addTask(key, DataOperation.CHANGE);    }

先来看一下onPut()方法，不难发现当注册实例数据有改变时，就无脑将这个实例扔到这个task内存阻塞队列中去，具体可以看一下addTask()方法。

public class Notifier implements Runnable {                private ConcurrentHashMap services = new ConcurrentHashMap<>(10 * 1024);        //用于存放即将要注册实例信息的内存阻塞队列        private BlockingQueue> tasks = new ArrayBlockingQueue<>(1024 * 1024);                /**         * Add new notify task to queue.         *         * @param datumKey data key         * @param action   action for data         */        public void addTask(String datumKey, DataOperation action) {                        if (services.containsKey(datumKey) && action == DataOperation.CHANGE) {                return;            }            if (action == DataOperation.CHANGE) {                services.put(datumKey, StringUtils.EMPTY);            }            /**             * 把客户端的参数封装成pair对象后，放进了一个内存队列中，注册就结束了，看这里并没有把客户端的注册信息写进双层map中             * 凭经验能猜到，这里把客户端对象放进内存队列，后续肯定是通过异步起线程的方式去注册             */            tasks.offer(Pair.with(datumKey, action));        }                public int getTaskSize() {            return tasks.size();        }                @Override        public void run() {            Loggers.DISTRO.info("distro notifier started");            /**             * Spring启动时就会开启一个线程加载 Notifier任务，这里就会死循环一直从内存队列中拿取实例信息实现异步注册             *             * 问   题1：这里的for循环会占用cpu资源吗？             *          不会占用，因为tasks是个阻塞队列，如果tasks中没有实例信息，这里就会阻塞在这，不会无脑死循环             *             * 问   题2：为什么要把实例信息都无脑先放在内存阻塞队列中，然后另起一个线程去异步注册呢？阿里这里为什么要这么设计？             * 个人理解：nacos是在阿里内部使用的中间件，一般是需要满足三高特性【高并发、高性能、高可扩展】，阿里内部就有几十万台机器，             * 如果不能实现高并发注册那么肯定会有很多问题。比如订单服务A需要注册到nacos时，是在订单A启动时就需要注册，服务注册到nacos的逻辑             * 还是比较复杂的【详见com.alibaba.nacos.naming.core.Service#updateIPs】，假如这里不用异步注册而是用同步注册的方式，那么             * 服务注册到nacos需要花费很多时间，这才是一个注册到nacos的行为就花费了大量时间，那么如果多几个中间需要加载的话，那得浪费多少时间？             * 所以这里采用异步注册。             *             * 问   题3：内存阻塞队列 tasks 会不会有堆积的情况呢？             * 实际上看了com.alibaba.nacos.naming.core.Cluster#updateIps注册方法可以发现，注册实际上就是把实例信息写进一个内存集合Set中             * 【com.alibaba.nacos.naming.core.Cluster#ephemeralInstances】这样的操作其实是很快的，假如真有个运维写了个批量注册的脚本             * 把一堆机器同时注册进来，那这样确实有可能会造成内存阻塞队列tasks的堆积现象，但是这种情况并没什么关系，Eureka有时候实例注册都会感知             * 几十秒，对当前的nacos架构而言，既然要实现高并发那么只能牺牲一点实例注册的即使响应时间。正常情况下，即使有几十台几百台机器同时注册，             * 由于注册是内存操作，速度也很快，可以说是准实时，基本上正常情况下注册信息1s就能感知到。             *             */            for (; ; ) {                try {                    //从内存队列中拿出任务                    Pair pair = tasks.take();                    //拿出pair对象中的客户端信息去注册                    handle(pair);                } catch (Throwable e) {                    //这个线程即使抛异常也不终止                    Loggers.DISTRO.error("[NACOS-DISTRO] Error while handling notifying task", e);                }            }        }                private void handle(Pair pair) {            try {                //前面拼接的参数 "com.alibaba.nacos.naming.iplist.ephemeral.namespaceId##serviceName"                String datumKey = pair.getValue0();                //前面传的DataOperation.CHANGE。                DataOperation action = pair.getValue1();                                services.remove(datumKey);                                int count = 0;                                if (!listeners.containsKey(datumKey)) {                    return;                }                                for (RecordListener listener : listeners.get(datumKey)) {                                        count++;                                        try {                        if (action == DataOperation.CHANGE) {                            /**                             * 拿到前面放的map中的客户端信息dataStore.get，这里的key就是前面的                             * 拼接的参数 "com.alibaba.nacos.naming.iplist.ephemeral.namespaceId##serviceName"                             * 前面放的是DataOperation.CHANGE。                             *                             */                            listener.onChange(datumKey, dataStore.get(datumKey).value);                            continue;                        }                                                if (action == DataOperation.DELETE) {                            listener.onDelete(datumKey);                            continue;                        }                    } catch (Throwable e) {                        Loggers.DISTRO.error("[NACOS-DISTRO] error while notifying listener of key: {}", datumKey, e);                    }                }                                if (Loggers.DISTRO.isDebugEnabled()) {                    Loggers.DISTRO                            .debug("[NACOS-DISTRO] datum change notified, key: {}, listener count: {}, action: {}",                                    datumKey, count, action.name());                }            } catch (Throwable e) {                Loggers.DISTRO.error("[NACOS-DISTRO] Error while handling notifying task", e);            }        }    }

当有实例需要注册时，直接调用addTask()方法将这个实例信息无脑扔进内存阻塞队列中去，注册就结束了。这个应该算是Nacos注册的一个精髓吧，Nacos为了提高性能其源码使用了大量的异步任务、异步线程等操作，用这些方式对提升Nacos性能有很大帮助。不难猜到，这里把客户端实例对象放进内存队列，后续肯定是通过异步起线程的方式去注册。

不难发现addTask()方法是Notifier类的方法，Notifier实现了Runnable接口，很明显这就是一个异步线程，这里跟上面的猜想一致，Nacos就是通过开启了一个异步线程实现注册的，具体的注册方法直接可以看Notifier线程的run()方法即可。那么这个Notifier线程是啥时候开启的呢？

@DependsOn("ProtocolManager")@org.springframework.stereotype.Service("distroConsistencyService")public class DistroConsistencyServiceImpl implements EphemeralConsistencyService, DistroDataProcessor {        //本类DistroConsistencyServiceImpl注入到了Spring容器中，所以项目启动时类加载的时候就会初始化这个方法    @PostConstruct    public void init() {        //线程池执行器，执行notifier线程的任务        GlobalExecutor.submitDistroNotifyTask(notifier);    }}

了解Spring的应该都知道这个是Spring加载的一种初始化方式，Spring启动时加载这个init方法初始化数据，就会开启一个线程加载Notifier任务。

看了Notifier线程的run()方法，不免会有几个疑问。第一、这里的for循环会占用cpu资源吗？第二、为什么要把实例信息都无脑先放在内存阻塞队列中，然后另起一个线程去异步注册呢？第三、阿里这里为什么要这么设计？这样设计好处是什么呢？

这里第一个问题不会占用cpu资源，因为tasks是个阻塞队列，如果tasks中没有实例信息，这里就会阻塞在这，不会无脑死循环，所以是不会占用cpu资源的；

第二个问题个人理解：Nacos是在阿里内部使用的中间件，肯定是需要满足高并发、高性能、高可扩展，阿里内部估计就有几十万台机器，如果不能实现高并发注册那么肯定会有很多问题。比如订单服务需要注册到nacos时，那么订单启动时就需要注册，服务注册到Nacos的逻辑还是比较复杂的【详见com.alibaba.nacos.naming.core.Service#updateIPs】，假如这里不用异步注册而是用同步注册的方式，那么服务注册到Nacos需要花费很多时间，这才是一个注册到Nacos的行为就花费了大量时间，那么如果多几个中间需要加载的话，那会浪费很多时间，所以这里采用异步注册。

那么我们再看一下异步写注册表的方法：

public void onChange(String key, Instances value) throws Exception {                Loggers.SRV_LOG.info("[NACOS-RAFT] datum is changed, key: {}, value: {}", key, value);                for (Instance instance : value.getInstanceList()) {                        if (instance == null) {                // Reject this abnormal instance list:                throw new RuntimeException("got null instance " + key);            }            //设置权重默认值啥的            if (instance.getWeight() > 10000.0D) {                instance.setWeight(10000.0D);            }                        if (instance.getWeight() < 0.01D && instance.getWeight() > 0.0D) {                instance.setWeight(0.01D);            }        }        //真正的注册实例的方法        updateIPs(value.getInstanceList(), KeyBuilder.matchEphemeralInstanceListKey(key));                recalculateChecksum();      }        public void updateIPs(Collection instances, boolean ephemeral) {        Map> ipMap = new HashMap<>(clusterMap.size());        for (String clusterName : clusterMap.keySet()) {            ipMap.put(clusterName, new ArrayList<>());        }                for (Instance instance : instances) {            try {                if (instance == null) {                    Loggers.SRV_LOG.error("[NACOS-DOM] received malformed ip: null");                    continue;                }                                if (StringUtils.isEmpty(instance.getClusterName())) {                    instance.setClusterName(UtilsAndCommons.DEFAULT_CLUSTER_NAME);                }                                if (!clusterMap.containsKey(instance.getClusterName())) {                    Loggers.SRV_LOG                            .warn("cluster: {} not found, ip: {}, will create new cluster with default configuration.",                                    instance.getClusterName(), instance.toJson());                    Cluster cluster = new Cluster(instance.getClusterName(), this);                    cluster.init();                    getClusterMap().put(instance.getClusterName(), cluster);                }                                List clusterIPs = ipMap.get(instance.getClusterName());                if (clusterIPs == null) {                    clusterIPs = new LinkedList<>();                    ipMap.put(instance.getClusterName(), clusterIPs);                }                                clusterIPs.add(instance);            } catch (Exception e) {                Loggers.SRV_LOG.error("[NACOS-DOM] failed to process ip: " + instance, e);            }        }                for (Map.Entry> entry : ipMap.entrySet()) {            //make every ip mine            List entryIPs = entry.getValue();            /**             * 注册逻辑updateIps，更新注册表信息：             *             * 这个方法里面会将已经注册过的实例列表复制一份，将新的实例和老的实例都更新到一个集合中，             * 最终再将这个集合更新到真正的实例列表，是一种写时复制的思想，主要时为了解决并发冲突，             * 在写的过程中，其他线程读到的还是旧数据，等真正写完之后再将数据更新回去。             *             * 思考一下，正常情况下在写之前都要加锁，不然可能会有读写并发问题，这里为什么不在加锁？             * 假如这里加了一个锁之后，相当于把读和写操作排队串行化执行了，就是读写不能同时进行了，             * 这里并发肯定会很低，所以这里用copy on write机制，将原来的注册表复制出一个副本，然后进             * 行修改，此时读请求进来还是读老的注册表，这样读写就能并发执行。             *             * 那这里用读写分离和加锁串行执行有什么优劣势吗？             * 读写分离：写的时候写的是副本，读的是老得数据，这样可能读到读不是最新数据，只有当副本写完将             * 老数据替换，此时读的才是最新数据，读写分离虽然提高了读写并发但是对数据的一致性稍有妥协，但是             * 对于此时注册的场景而言影响不大，即使是没有读到最新数据也没关系，最多就当此服务启动的慢一点而             * 已，当前这个注册场景，还是提高并发注册能力稍重要些，若是对读写数据一致场景要求很高时，就必须得             * 加锁串行执行             *             * 加锁执行：读写都是加锁执行，写完后再去读，读的一定是最新的数据，读写数据强一致，但是这里根本不需要数据强一致             *             * 这里有个疑问注册时每个微服务都去复制一个副本，然后将副本替换回原件时会不会有覆盖的问题？             * 不会，这里是开的一个线程去拿内存队列的数据进行注册的，所以不可能存在覆盖并发问题。详见             * com.alibaba.nacos.naming.consistency.ephemeral.distro.DistroConsistencyServiceImpl#init()方法             *             */            clusterMap.get(entry.getKey()).updateIps(entryIPs, ephemeral);        }                setLastModifiedMillis(System.currentTimeMillis());        //上面已经更新过了注册表后，这里需要发布事件，主动通知客户端        getPushService().serviceChanged(this);        StringBuilder stringBuilder = new StringBuilder();                for (Instance instance : allIPs()) {            stringBuilder.append(instance.toIpAddr()).append("_").append(instance.isHealthy()).append(",");        }                Loggers.EVT_LOG.info("[IP-UPDATED] namespace: {}, service: {}, ips: {}", getNamespaceId(), getName(),                stringBuilder.toString());             }    /**     * Update instance list.     *     * @param ips       instance list    需要注册的实例     * @param ephemeral whether these instances are ephemeral  是哪种实例，永久实例还是临时实例     */    public void updateIps(List ips, boolean ephemeral) {        //根据传进来的 ephemeral 判断是哪种实例，直接把这种实例复制一份toUpdateInstances        Set toUpdateInstances = ephemeral                ? ephemeralInstances    //临时实例 【这里这个实例是注册表已有的实例，这里直接将原有的实例再次复制了一份，就是toUpdateInstances】                : persistentInstances;  //永久实例 【这里这个实例是注册表已有的实例，这里直接将原有的实例再次复制了一份，就是toUpdateInstances】        HashMap oldIpMap = new HashMap<>(toUpdateInstances.size());        //循环将上面复制的实例toUpdateInstances放oldIpMap中        for (Instance ip : toUpdateInstances) {            oldIpMap.put(ip.getDatumKey(), ip);        }        /**         * 下面就是对比新注册的实例与副本实例的差异，判断新的实例是新增、删除、还是修改，这里使用了copyOnWrite机制，上面的toUpdateInstances就是         * 复制出来的副本，新增、删除、修改都是在副本上进行，之后再将原注册表覆盖         * 新增：直接新增到副本中         * 修改：直接修改副本的实例信息         * 删除：删除副本的实例         *         */        //更新注册表        List updatedIPs = updatedIps(ips, oldIpMap.values());        if (updatedIPs.size() > 0) {            for (Instance ip : updatedIPs) {                Instance oldIP = oldIpMap.get(ip.getDatumKey());                                // do not update the ip validation status of updated ips                // because the checker has the most precise result                // Only when ip is not marked, don"t we update the health status of IP:                if (!ip.isMarked()) {                    ip.setHealthy(oldIP.isHealthy());                }                                if (ip.isHealthy() != oldIP.isHealthy()) {                    // ip validation status updated                    Loggers.EVT_LOG.info("{} {SYNC} IP-{} {}:{}@{}", getService().getName(),                            (ip.isHealthy() ? "ENABLED" : "DISABLED"), ip.getIp(), ip.getPort(), getName());                }                                if (ip.getWeight() != oldIP.getWeight()) {                    // ip validation status updated                    Loggers.EVT_LOG.info("{} {SYNC} {IP-UPDATED} {}->{}", getService().getName(), oldIP.toString(),                            ip.toString());                }            }        }        //删除注册表        List newIPs = subtract(ips, oldIpMap.values());        if (newIPs.size() > 0) {            Loggers.EVT_LOG                    .info("{} {SYNC} {IP-NEW} cluster: {}, new ips size: {}, content: {}", getService().getName(),                            getName(), newIPs.size(), newIPs.toString());                        for (Instance ip : newIPs) {                HealthCheckStatus.reset(ip);            }        }                List deadIPs = subtract(oldIpMap.values(), ips);                if (deadIPs.size() > 0) {            Loggers.EVT_LOG                    .info("{} {SYNC} {IP-DEAD} cluster: {}, dead ips size: {}, content: {}", getService().getName(),                            getName(), deadIPs.size(), deadIPs.toString());                        for (Instance ip : deadIPs) {                HealthCheckStatus.remv(ip);            }        }                toUpdateInstances = new HashSet<>(ips);

第三个问题：可以看上面这个写注册表的源码，当服务A需要注册到Nacos时，并不是直接写进Nacos的注册表里，实际上是先拷贝了一个副本，订单服务注册写注册表时直接写副本的注册表，副本写完后才会替换原来Nacos中的注册表，所以当库存服务需要从Nacos拉取服务时，拉取的是Nacos实际注册表中的信息，这种设计方式能够大大提高Nacos的注册性能。

类似于CopyOnWriteArrayList的copy on write机制，也就是写时复制、读写分离设计思想。这种读写分离对于客户端注册感知实时性可能会稍差点，但是这种情况并没什么关系，Eureka有时候实例注册都会感知几十秒，对当前的nacos架构而言，既然要实现高并发那么只能牺牲一点实例注册的及时响应时间。

当Nacos注册成功后，就需要发布事件，主动通知客户端，接下来可以看一下发布事件的源码：

public void serviceChanged(Service service) {        // merge some change events to reduce the push frequency:        if (futureMap                .containsKey(UtilsAndCommons.assembleFullServiceName(service.getNamespaceId(), service.getName()))) {            return;        }        //时间发布，发布事件，通知客户端        this.applicationContext.publishEvent(new ServiceChangeEvent(this, service));    }

这里Nacos会通过udp的方式将服务变动通知给订阅的客户端。Nacos的这种推送模式相对于zk那种利用tcp长连接而言还是会节约很多资源，即使有大量节点更新也不会使Nacos出现性能瓶颈。

当Nacos客户端接收到了udp消息后会给服务端返回一个ack，如果Nacos超时未收到ack，还会有重发机制，超过了这个超时时间就不再重发了。虽然udp是个不可靠协议不能保证消息一定能推送到客户端，但是Nacos客户端还是有定时轮训做兜底定时查询Nacos注册表。Nacas采用了这两种机制，既保证了实时性，又保证了数据更新不会被漏掉。

Nacos数据同步分为全量同步和增量同步，全量同步就是初始化数据一次性同步，而增量同步是指有数据增加的时候，只同步增加的数据。

7.3.5.1 Nacos集群全量数据同步

Nacos集群有新的节点启动时，DistroProtocol类就会在Spring加载时调用构造方法，同时开启一个数据同步任务，该方法会执行startVerifyTask()和startLoadTask()，我们重点关注startLoadTask(),具体代码如下：

@Componentpublic class DistroProtocol {    /**     * 当nacos集群只有两台机器时，此时若又新增一台机器，此时需要将原来两台机器的数据同步到新的nacos机器上     *     * @param memberManager     * @param distroComponentHolder     * @param distroTaskEngineHolder     * @param distroConfig     */    public DistroProtocol(ServerMemberManager memberManager, DistroComponentHolder distroComponentHolder,            DistroTaskEngineHolder distroTaskEngineHolder, DistroConfig distroConfig) {        this.memberManager = memberManager;        this.distroComponentHolder = distroComponentHolder;        this.distroTaskEngineHolder = distroTaskEngineHolder;        this.distroConfig = distroConfig;        //本项目启动时DistroProtocol类加载时需要加载本构造方法，开启一个数据同步任务        startDistroTask();    }        private void startDistroTask() {        if (EnvUtil.getStandaloneMode()) {            isInitialized = true;            return;        }        //启动startVerifyTask，做数据同步校验        startVerifyTask();        //加载任务        startLoadTask();    }        private void startLoadTask() {        ///处理状态回调对象        DistroCallback loadCallback = new DistroCallback() {            //处理成功            @Override            public void onSuccess() {                isInitialized = true;            }            //处理失败            @Override            public void onFailed(Throwable throwable) {                isInitialized = false;            }        };        GlobalExecutor.submitLoadDataTask(                new DistroLoadDataTask(memberManager, distroComponentHolder, distroConfig, loadCallback));    } }

上面方法会调用DistroLoadDataTask对象，而该对象其实是个线程，因此会执行它的run方法，run方法会调用load()方法实现数据全量加载，代码如下：

public class DistroLoadDataTask implements Runnable {        private final ServerMemberManager memberManager;        private final DistroComponentHolder distroComponentHolder;        private final DistroConfig distroConfig;        private final DistroCallback loadCallback;        private final Map loadCompletedMap;        public DistroLoadDataTask(ServerMemberManager memberManager, DistroComponentHolder distroComponentHolder,            DistroConfig distroConfig, DistroCallback loadCallback) {        this.memberManager = memberManager;        this.distroComponentHolder = distroComponentHolder;        this.distroConfig = distroConfig;        this.loadCallback = loadCallback;        loadCompletedMap = new HashMap<>(1);    }    /**     * 数据加载过程     */    @Override    public void run() {        try {            //加载数据            load();            if (!checkCompleted()) {                GlobalExecutor.submitLoadDataTask(this, distroConfig.getLoadDataRetryDelayMillis());            } else {                loadCallback.onSuccess();                Loggers.DISTRO.info("[DISTRO-INIT] load snapshot data success");            }        } catch (Exception e) {            loadCallback.onFailed(e);            Loggers.DISTRO.error("[DISTRO-INIT] load snapshot data failed. ", e);        }    }}

数据同步会通过Http请求从远程服务器获取数据，并同步到当前服务的缓存中。执行流程如下：

首先，loadAllDataSnapshotFromRemote()从远程加载所有数据，并处理同步到本机；

第二，transportAgent.getDatumSnapshot()远程加载数据，通过Http请求执行远程加载；

第三，dataProcessor.processSnapshot()处理数据同步到本地

/**     * 加载数据，并同步     *     * @throws Exception     */    private void load() throws Exception {        while (memberManager.allMembersWithoutSelf().isEmpty()) {            Loggers.DISTRO.info("[DISTRO-INIT] waiting server list init...");            TimeUnit.SECONDS.sleep(1);        }        while (distroComponentHolder.getDataStorageTypes().isEmpty()) {            Loggers.DISTRO.info("[DISTRO-INIT] waiting distro data storage register...");            TimeUnit.SECONDS.sleep(1);        }        for (String each : distroComponentHolder.getDataStorageTypes()) {            if (!loadCompletedMap.containsKey(each) || !loadCompletedMap.get(each)) {                //从远端机器拉取数据，从远程加载所有数据，并处理同步到本机                loadCompletedMap.put(each, loadAllDataSnapshotFromRemote(each));            }        }    }    /**     * 从远端机器拉取数据     *     * @param resourceType     * @return     */    private boolean loadAllDataSnapshotFromRemote(String resourceType) {        DistroTransportAgent transportAgent = distroComponentHolder.findTransportAgent(resourceType);        DistroDataProcessor dataProcessor = distroComponentHolder.findDataProcessor(resourceType);        if (null == transportAgent || null == dataProcessor) {            Loggers.DISTRO.warn("[DISTRO-INIT] Can"t find component for type {}, transportAgent: {}, dataProcessor: {}",                    resourceType, transportAgent, dataProcessor);            return false;        }        //拉取不包含自己的机器        for (Member each : memberManager.allMembersWithoutSelf()) {            try {                Loggers.DISTRO.info("[DISTRO-INIT] load snapshot {} from {}", resourceType, each.getAddress());                //调取接口获取除自己外的所有机器                DistroData distroData = transportAgent.getDatumSnapshot(each.getAddress());                //同步数据                boolean result = dataProcessor.processSnapshot(distroData);                Loggers.DISTRO                        .info("[DISTRO-INIT] load snapshot {} from {} result: {}", resourceType, each.getAddress(),                                result);                //同步成功直接return，从一台机器上同步                if (result) {                    return true;                }            } catch (Exception e) {                Loggers.DISTRO.error("[DISTRO-INIT] load snapshot {} from {} failed.", resourceType, each.getAddress(), e);            }        }        return false;    }        private boolean checkCompleted() {        if (distroComponentHolder.getDataStorageTypes().size() != loadCompletedMap.size()) {            return false;        }        for (Boolean each : loadCompletedMap.values()) {            if (!each) {                return false;            }        }        return true;    }

到这为止实现数据全量同步，其实全量同步最终还是互相调用Nacos提供的api。总结一下全量数据同步的过程：

7.3.5.2 Nacos集群增量数据同步

当服务注册完成后，Nacos需要将客户端实例信息同步到Nacos集群其他节点，可以看一下Nacos底层是怎么实现的。我们再次回到put方法：

public void put(String key, Record value) throws NacosException {        //注册逻辑：实际就是把实例注册任务放到内存阻塞队列中        onPut(key, value);        //AP 架构下的节点数据同步        distroProtocol.sync(new DistroKey(key, KeyBuilder.INSTANCE_LIST_KEY_PREFIX), DataOperation.CHANGE,                globalConfig.getTaskDispatchPeriod() / 2);    }

上文中已经解释过put方法中的 onPut(key, value)方法，接下来我们再了解一下AP结构Nacos下的节点数据是同步，也就是distroProtocol.sync方法。

** * Start to sync data to all remote server. * * @param distroKey distro key of sync data * @param action    the action of data operation */public void sync(DistroKey distroKey, DataOperation action, long delay) {    //循环将新增实例同步到除自己外的所有实例，单机for循环都不会走，集群架构就会走本方法    for (Member each : memberManager.allMembersWithoutSelf()) {        //先把每台机器都数据封装称distroKeyWithTarget对象        DistroKey distroKeyWithTarget = new DistroKey(distroKey.getResourceKey(), distroKey.getResourceType(),                each.getAddress());        DistroDelayTask distroDelayTask = new DistroDelayTask(distroKeyWithTarget, action, delay);        //添加到task任务中        distroTaskEngineHolder.getDelayTaskExecuteEngine().addTask(distroKeyWithTarget, distroDelayTask);        if (Loggers.DISTRO.isDebugEnabled()) {            Loggers.DISTRO.debug("[DISTRO-SCHEDULE] {} to {}", distroKey, each.getAddress());    }}/** * 将集群中除自己外其他需要同步的机器信息添加到一个tasks任务中，由前面的知识可以猜到这里也是 * 用异步开启一个线程去拿tasks进行同步新增实例信息到其他nacos机器中 * * @param key  key of task * @param newTask */@Overridepublic void addTask(Object key, AbstractDelayTask newTask) {    lock.lock();    try {        AbstractDelayTask existTask = tasks.get(key);        if (null != existTask) {            newTask.merge(existTask);        }        //将集群中除自己外其他需要同步的机器信息添加到一个tasks任务中,task是个保存信息的ConcurrentHashMap        tasks.put(key, newTask);    } finally {        lock.unlock();    }}

这里直接把需要同步的信息放在了内存的ConcurrentHashMap中，我们看一下这里具体看一下怎么同步其他节点。

@Componentpublic class DistroTaskEngineHolder {        //延迟任务执行器    private final DistroDelayTaskExecuteEngine delayTaskExecuteEngine = new DistroDelayTaskExecuteEngine();        //任务执行引擎器    private final DistroExecuteTaskExecuteEngine executeWorkersManager = new DistroExecuteTaskExecuteEngine();        public DistroTaskEngineHolder(DistroComponentHolder distroComponentHolder) {        DistroDelayTaskProcessor defaultDelayTaskProcessor = new DistroDelayTaskProcessor(this, distroComponentHolder);        delayTaskExecuteEngine.setDefaultTaskProcessor(defaultDelayTaskProcessor);    }        public DistroDelayTaskExecuteEngine getDelayTaskExecuteEngine() {        return delayTaskExecuteEngine;    }        public DistroExecuteTaskExecuteEngine getExecuteWorkersManager() {        return executeWorkersManager;    }        public void registerNacosTaskProcessor(Object key, NacosTaskProcessor nacosTaskProcessor) {        this.delayTaskExecuteEngine.addProcessor(key, nacosTaskProcessor);    }}

这个类中会创建一个任务执行引擎，代码如下：

public class DistroDelayTaskExecuteEngine extends NacosDelayTaskExecuteEngine {        public DistroDelayTaskExecuteEngine() {        super(DistroDelayTaskExecuteEngine.class.getName(), Loggers.DISTRO);    }        @Override    public void addProcessor(Object key, NacosTaskProcessor taskProcessor) {        Object actualKey = getActualKey(key);        super.addProcessor(actualKey, taskProcessor);    }        @Override    public NacosTaskProcessor getProcessor(Object key) {        Object actualKey = getActualKey(key);        return super.getProcessor(actualKey);    }        private Object getActualKey(Object key) {        return key instanceof DistroKey ? ((DistroKey) key).getResourceType() : key;    }}//忽略一些中间省略的代码public NacosDelayTaskExecuteEngine(String name, int initCapacity, Logger logger, long processInterval) {    super(logger);    tasks = new ConcurrentHashMap(initCapacity);    processingExecutor = ExecutorFactory.newSingleScheduledExecutorService(new NameThreadFactory(name));    //这里又执行了一个任务    processingExecutor            .scheduleWithFixedDelay(new ProcessRunnable(), processInterval, processInterval, TimeUnit.MILLISECONDS);}/** * 执行了一个任务 */private class ProcessRunnable implements Runnable {        @Override    public void run() {        try {            processTasks();        } catch (Throwable e) {            getEngineLog().error(e.toString(), e);        }    }}

public boolean process(NacosTask task) {    if (!(task instanceof DistroDelayTask)) {        return true;    }    DistroDelayTask distroDelayTask = (DistroDelayTask) task;    DistroKey distroKey = distroDelayTask.getDistroKey();    if (DataOperation.CHANGE.equals(distroDelayTask.getAction())) {        //将延迟任务变更成异步任务，异步任务对象是一个线程        DistroSyncChangeTask syncChangeTask = new DistroSyncChangeTask(distroKey, distroComponentHolder);        //将前面封装到任务拿出来放在一个队列中        distroTaskEngineHolder.getExecuteWorkersManager().addTask(distroKey, syncChangeTask);        return true;    }    return false;}public void addTask(Object tag, AbstractExecuteTask task) {    //拿前面其他实例到任务    NacosTaskProcessor processor = getProcessor(tag);    if (null != processor) {        processor.process(task);        return;    }    TaskExecuteWorker worker = getWorker(tag);    //将同步数据到其他nacos实例到tasks任务放进一个queue中 在InnerWorker.run()方法中从queue队列中拿任务执行    worker.process(task);}public boolean process(NacosTask task) {    if (task instanceof AbstractExecuteTask) {        putTask((Runnable) task);    }    return true;}/** * 把任务同步放进内存队列中 * * @param task */private void putTask(Runnable task) {    try {        queue.put(task);    } catch (InterruptedException ire) {        log.error(ire.toString(), ire);    }}

将同步数据到其他Nacos实例到tasks任务放进一个queue中，然后在InnerWorker.run()方法中从queue队列中拿任务执行。看一下具体是怎么执行同步任务的：

/** * Inner execute worker. */private class InnerWorker extends Thread {        InnerWorker(String name) {        setDaemon(false);        setName(name);    }        @Override    public void run() {        while (!closed.get()) {            try {                //从queue中拿同步任务                Runnable task = queue.take();                long begin = System.currentTimeMillis();                //实际就是执行异步同步任务 DistroSyncChangeTask 的run()                task.run();                long duration = System.currentTimeMillis() - begin;                if (duration > 1000L) {                    log.warn("distro task {} takes {}ms", task, duration);                }            } catch (Throwable e) {                log.error("[DISTRO-FAILED] " + e.toString(), e);            }        }    }}

这里从队列里面拿出来任务执行，不难发现这里的任务执行的具体方法就是DistroSyncChangeTask类的run方法：

public class DistroSyncChangeTask extends AbstractDistroExecuteTask {        private final DistroComponentHolder distroComponentHolder;        public DistroSyncChangeTask(DistroKey distroKey, DistroComponentHolder distroComponentHolder) {        super(distroKey);        this.distroComponentHolder = distroComponentHolder;    }        @Override    public void run() {        Loggers.DISTRO.info("[DISTRO-START] {}", toString());        try {            //获取各种参数            String type = getDistroKey().getResourceType();            DistroData distroData = distroComponentHolder.findDataStorage(type).getDistroData(getDistroKey());            distroData.setType(DataOperation.CHANGE);            //调用http接口同步任务            boolean result = distroComponentHolder.findTransportAgent(type).syncData(distroData, getDistroKey().getTargetServer());            //同步失败会继续重试            if (!result) {                handleFailedTask();            }            Loggers.DISTRO.info("[DISTRO-END] {} result: {}", toString(), result);        } catch (Exception e) {            Loggers.DISTRO.warn("[DISTRO] Sync data change failed.", e);            handleFailedTask();        }    }    /**     * 同步失败会继续重试     */    private void handleFailedTask() {        String type = getDistroKey().getResourceType();        DistroFailedTaskHandler failedTaskHandler = distroComponentHolder.findFailedTaskHandler(type);        if (null == failedTaskHandler) {            Loggers.DISTRO.warn("[DISTRO] Can"t find failed task for type {}, so discarded", type);            return;        }        failedTaskHandler.retry(getDistroKey(), DataOperation.CHANGE);    }        @Override    public String toString() {        return "DistroSyncChangeTask for " + getDistroKey().toString();    }}

每个Nacos服务端实例都会提供这样的一个api接口供其他Nacos实例调用，从而同步注册实例数据。DistroSyncChangeTask类的run方法，就是调用http接口同步任务接口，将本节点的注册实例数据同步到其他节点机器上。

总结一下上面增量数据同步方法：

DistroProtocol 使用 sync() 方法处理AP 架构下的节点数据同步

向其他节点发布广播任务调用 distroTaskEngineHolder 发布延迟任务

调用 DistroDelayTaskProcessor.process() 方法进行任务投递：将延迟任务转换为异步变更任务

执行变更任务 DistroSyncChangeTask.run() 方法：向指定节点发送消息

Nacos避免并发读写的冲突：Nacos在更新实例列表时，会采用CopyOnWrite技术，首先将老得实例列表拷贝一份，然后更新拷贝的实例列表，再用更新后的实例列表来覆盖旧的实例列表。

Nacos提高注册并发：为了应对阿里巴巴内部数十万服务的并发写请求Nacos内部会将服务注册的任务放入阻塞队列，采用线程池异步来完成实例更新，从而提高并发写能力。

Nacos的服务发现分为两种模式：主动拉取模式，消费者定期主动从Nacos服务端拉取服务列表并缓存起来，当服务调用时优先读取本地缓存中的服务列表。订阅模式，消费者订阅Nacos中的服务列表，并基于UDP协议来接收服务变更通知。当Nacos中的服务列表更新时，会发送UDP广播给所有订阅者。与Eureka相比，Nacos的订阅模式服务状态更新更及时，消费者更容易及时发现服务列表的变化，剔除故障服务。

关键词： 定时任务 服务列表 不难发现 命名空间