nacos源码级别集群启动及拉取数据同步流程

目录

nacos集群启动

配置数据库

配置集群文件

nacos启动配置

nacos加载节点信息

集群心跳健康检查机制

节点状态同步

数据新增及变更同步

总结 

---

nacos集群启动

配置数据库

首先在MySQL中创建数据库nacos（名称随意），在nacos数据库中执行config子工程中的nacos-db.sql文件

然后在console子工程中，打开数据库的配置，因为nacos集群需要时外部数据库。

#*************** Config Module Related Configurations ***************#
### If use MySQL as datasource:
spring.datasource.platform=mysql
 
### Count of DB:
db.num=1
 
### Connect URL of DB:
db.url.0=jdbc:mysql://127.0.0.1:3306/nacos?characterEncoding=utf8&connectTimeout=1000&socketTimeout=3000&autoReconnect=true&useUnicode=true&useSSL=false&serverTimezone=UTC
db.user.0=root
db.password.0=123456

配置集群文件

在某个目录中创建三个文件

文件目录及文件名称，注意在8846文件内部还要创建conf目录，不然启动报错：UnknowHostException。

由于要启动三个节点，所以配置三个节点，文件内容如下：

192.168.1.88:8846
192.168.1.88:8847
192.168.1.88:8848

nacos启动配置

配置三个Spring Boot启动项，配置内容如图。配置完成后，就可以直接启动了。

nacos加载节点信息

nacos启动过程中，会加载ServerMemberManager这个bean，而他只有一个构造方法，所以会被spring调用，init()方法中主要是3件事：初始化该bean（节点）的属性；注册事件监听器，用来监听其他节点信息的变更；集群模式下读取配置文件信息。

public ServerMemberManager(ServletContext servletContext) throws Exception {
        this.serverList = new ConcurrentSkipListMap<>();
        EnvUtil.setContextPath(servletContext.getContextPath());
        init();
    }
    
    protected void init() throws NacosException {
        //1、初始化该节点的属性
        Loggers.CORE.info("Nacos-related cluster resource initialization");
        this.port = EnvUtil.getProperty("server.port", Integer.class, 8848);
        this.localAddress = InetUtils.getSelfIP() + ":" + port;
        this.self = MemberUtil.singleParse(this.localAddress);
        this.self.setExtendVal(MemberMetaDataConstants.VERSION, VersionUtils.version);
        serverList.put(self.getAddress(), self);
        //2、注册事件监听器到NotifyManager
        // register NodeChangeEvent publisher to NotifyManager
        registerClusterEvent();
        //3、集群模式下读取配置文件信息（这里由几种方式）
        // Initializes the lookup mode
        initAndStartLookup();        
        if (serverList.isEmpty()) {
            throw new NacosException(NacosException.SERVER_ERROR, "cannot get serverlist, so exit.");
        }        Loggers.CORE.info("The cluster resource is initialized");
    }

主要分析initAndStartLookup();调用链为lookup.start();—>doStart();—>FileConfigMemberLookup.doStart()—>readClusterConfFromDisk()，到这里就开始读取我们配置的节点内容，加载到内存后，会被解析为Member集合，然后调用ServerMemberManager#memberChange()方法，将信息添加到这个bean的集合属性中，如memberAddressInfos：维护所有"UP"状态的节点信息，serverList：集群所有节点。

集群心跳健康检查机制

前面已经分析过单节点nacos心跳逻辑是在服务注册的过程中启动心跳任务ClientBeatCheckTask,如果是nacos集群环境，在执行心跳任务之前会有如下判断，目的就是一个service只会由一个节点执行心跳任务，而不是所有节点

if (!getDistroMapper().responsible(service.getName())) {
     return;
}

三个节点都会执行responsible()方法，但是在调用distroHash(serviceName)，对节点个数求余后，只会有一个节点返回true。然后才会执行心跳任务

    public boolean responsible(String serviceName) {
        final List servers = healthyList;
        
        if (!switchDomain.isDistroEnabled() || EnvUtil.getStandaloneMode()) {
            return true;
        }
        
        if (CollectionUtils.isEmpty(servers)) {
            // means distro config is not ready yet
            return false;
        }
        
        int index = servers.indexOf(EnvUtil.getLocalAddress());
        int lastIndex = servers.lastIndexOf(EnvUtil.getLocalAddress());
        if (lastIndex < 0 || index < 0) {
            return true;
        }
        
        int target = distroHash(serviceName) % servers.size();
        return target >= index && target <= lastIndex;
    }
 
    private int distroHash(String serviceName) {
        return Math.abs(serviceName.hashCode() % Integer.MAX_VALUE);
    }

Debug验证结果：注册两个不同的服务，在不重启的情况下，会分别固定由8846和8848这两个节点保持心跳。 debug断点是在ClientBeatCheckTask#run()中。

节点状态同步

在nacos启动中，会加载ServerListManager到spring容器中，它的init()方法被@PostConstruct注解了，所以会被执行。

    @PostConstruct
    public void init() {
        GlobalExecutor.registerServerStatusReporter(new ServerStatusReporter(), 2000);
        GlobalExecutor.registerServerInfoUpdater(new ServerInfoUpdater());
    }

创建线程池每隔2s执行ServerStatusReporter任务，代码摘抄了重要部分，主要先从serverMemberManager对象中获取所有节点，遍历排除自己，发送状态数据到各个节点的/operator/server/status接口中。

        @Override
        public void run() {
            try {
                ...
                int weight = Runtime.getRuntime().availableProcessors() / 2;
                if (weight <= 0) {
                    weight = 1;
                }
                long curTime = System.currentTimeMillis();
                String status = LOCALHOST_SITE + "#" + EnvUtil.getLocalAddress() + "#" + curTime + "#" + weight
                        + "\r\n";
                // 获取所有的节点信息
                List allServers = getServers();
                ...
                //遍历
                if (allServers.size() > 0 && !EnvUtil.getLocalAddress()
                        .contains(IPUtil.localHostIP())) {
                    for (Member server : allServers) {
                        //排除自己
                        if (Objects.equals(server.getAddress(), EnvUtil.getLocalAddress())) {
                            continue;
                        }
                        ...
                        Message msg = new Message();
                        msg.setData(status);
                        //向接口/operator/server/status发送数据
                        synchronizer.send(server.getAddress(), msg);
                    }
                }
            } catch (Exception e) {
                Loggers.SRV_LOG.error("[SERVER-STATUS] Exception while sending server status", e);
            } finally {
                GlobalExecutor
                        .registerServerStatusReporter(this, switchDomain.getServerStatusSynchronizationPeriodMillis());
            }
            
        }

数据新增及变更同步

在nacos服务启动中，会加载ServiceManager为spring的bean对象，执行init()方法，其中会创建定时任务线程池每隔1分钟执行ServiceReporter任务，他就是nacos各个节点间同步服务实例元数据的任务。一下是run()所有内容

                //从serviceMap中获取所有的serviceName,key:namespaceId,value:set
                Map> allServiceNames = getAllServiceNames();
                
                if (allServiceNames.size() <= 0) {
                    //ignore
                    return;
                }
                
                for (String namespaceId : allServiceNames.keySet()) {
                    //创建需要同步的数据对象，它封装了namespaceId对应的service所有的实例信息
                    ServiceChecksum checksum = new ServiceChecksum(namespaceId);
                    //遍历serviceName集合,获取每个service所对应的全部实例信息，
                    for (String serviceName : allServiceNames.get(namespaceId)) {
                        //只有维持心跳的节点才会向checksum中添加数据
                        if (!distroMapper.responsible(serviceName)) {
                            continue;
                        }
                        Service service = getService(namespaceId, serviceName);
                        if (service == null || service.isEmpty()) {
                            continue;
                        }
                        //拼接所有实例信息，解析为md5赋值给checksum属性
                        service.recalculateChecksum();
                        //添加到checksum中
                        checksum.addItem(serviceName, service.getChecksum());
                    }
                    //封装消息
                    Message msg = new Message();
                    msg.setData(JacksonUtils.toJson(checksum));
                    //拿到所有nacos节点地址
                    Collection sameSiteServers = memberManager.allMembers();
                    if (sameSiteServers == null || sameSiteServers.size() <= 0) {
                        return;
                    }
                    //将消息发送给除自身意外的所有nacos节点
                    for (Member server : sameSiteServers) {
                        if (server.getAddress().equals(NetUtils.localServer())) {
                            continue;
                        }
                        synchronizer.send(server.getAddress(), msg);
                    }
                }

大致可以总结为将namespaceId对应的所有实例元数据信息，对于serviceName下所有实例信息，只有维持该serviceName心跳的节点才会对这元数据信息进行处理，将他们都加到一个checksum对象中，然后封装为Message对象中，最后发送给其他所有nacos节点。直到所有namespaceId都遍历结束。

总结 

1、集群环境维持每个service心跳的算法，对于一个服务类型会对他的serviceName进行hash，然后对集群节点数量求余，得到一个节点，该节点就是维持该服务类型所对应的所有实例。

2、节点之间同步服务实例数据就是基于1中选出来的节点，每个节点会向其他节点同步自己维持心跳的服务的所有实例。