Nacos注册表是怎么解决读写并发冲突的

Posted 2022-07-15 抓手

Nacos在更新实例列表的时候，采用CopyOnWrite思想，将传入的实例列表List<Instance> ips与旧的实例列表比对，分别得出需要添加、更新、和删除的实例，然后做一些相关的操作，得到最终新实例列表并替换原来的旧实例列表。

更新过程中的旧实例列表不受影响，客户端依然可以读取。

源码在com.alibaba.nacos.naming.core.Cluster类下面：

    /**
     * 更新实例列表
     *
     * @param ips       实例列表
     * @param ephemeral 是否为临时实例
     */
    public void updateIps(List<Instance> ips, boolean ephemeral) 

        Set<Instance> toUpdateInstances = ephemeral ? ephemeralInstances : persistentInstances;

        // 旧的实例列表
        HashMap<String, Instance> oldIpMap = new HashMap<>(toUpdateInstances.size());
        for (Instance ip : toUpdateInstances) 
            oldIpMap.put(ip.getDatumKey(), ip);
        

        // 需要更新的实例列表
        List<Instance> 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<Instance> 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<Instance> 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);

        if (ephemeral) 
            ephemeralInstances = toUpdateInstances;
         else 
            persistentInstances = toUpdateInstances;