【Nacos】—客户端与服务端源码解析

Nacos系列

Nacos—简述、注册中心、配置中心
Nacos安装教程
SpringBoot项目与Nacos配置

一、背景介绍

Nacos（Naming and Configuration Service）是阿里巴巴开源的服务发现和配置管理工具，它是一个全面的微服务基础设施组件，提供了服务注册与发现、配置管理、动态DNS服务等功能。基于之前发表的博客已经对Nacos的演变以及项目如何集成应用做了说明，今天我们一起来探索探索Nacos源码。

二、正文

①、服务注册与发现

nacos客户端

思想：服务注册过程其实就是nacos客户端发起http请求调用，把客户端的信息发给nacos服务端，将服务注册到nacos服务端



技术：引用谷歌的grpc（长连接协议），这样的好处是减少了http请求频繁的连接创建和销毁过程，大幅度提升性能，节约资源





完成注册后会调用nacos客户端的bind方法，这个方法的作用是将服务实例和指定的服务组或命名空间进行关联，设置服务的访问权限等操作
注册完成后还会调用start方法，发布一个事件通知，通知其他服务，我这个服务注册到nacos里面了

### nacos服务端

服务注册到nacos中之后nacos服务端需要将服务实例信息保存起来，那是如何保存的呢？
Nacos Server接收到注册请求后，就会把这些元数据信息存储在一个双层的内存Map中，俗称注册表


namespace：区分环境：dev、test……

group：分组
cluster-name：多机房部署，就近访问

注册接口controller

    /**
     * Register new instance.
     *
     * @param request http request
     * @return 'ok' if success
     * @throws Exception any error during register
     */
    @CanDistro
    @PostMapping
    @Secured(action = ActionTypes.WRITE)
    public String register(HttpServletRequest request) throws Exception {
        //获取namespaceId，默认为public
        final String namespaceId = WebUtils
                .optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);
        //获取service-name
        final String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);
        //根据分割规则，分割服务名，不符合规则的则抛出异常
        NamingUtils.checkServiceNameFormat(serviceName);
        //获取服务实例
        final Instance instance = HttpRequestInstanceBuilder.newBuilder()
                .setDefaultInstanceEphemeral(switchDomain.isDefaultInstanceEphemeral()).setRequest(request).build();
        //注册实例
        getInstanceOperator().registerInstance(namespaceId, serviceName, instance);
        //在注册表中添加客户端注册信息，并发版订阅事件
        NotifyCenter.publishEvent(new RegisterInstanceTraceEvent(System.currentTimeMillis(), "", false, namespaceId,
                NamingUtils.getGroupName(serviceName), NamingUtils.getServiceName(serviceName), instance.getIp(),
                instance.getPort()));
        return "ok";
    }

注册流程

    /**
     * This method creates {@code IpPortBasedClient} if it don't exist.
     */
    @Override
    public void registerInstance(String namespaceId, String serviceName, Instance instance) throws NacosException {
        NamingUtils.checkInstanceIsLegal(instance);
        //是否是短暂实例信息
        boolean ephemeral = instance.isEphemeral();
        //获取客户端ip信息：192.168.0.1:9002#true
        String clientId = IpPortBasedClient.getClientId(instance.toInetAddr(), ephemeral);
        //像clientManager中注册客户的那信息
        createIpPortClientIfAbsent(clientId);
        //获取服务
        Service service = getService(namespaceId, serviceName, ephemeral);
        //注册实例
        clientOperationService.registerInstance(service, instance, clientId);
    }

---

②、心跳机制

nacos提供了两种服务类型：
永久实例（持久化）：一直存储在注册表中，服务断开也不会删除，不会主动向注册中心发送心跳，服务端反向探测
临时实例：服务断开后会从注册表中删除，服务再次重启重新注册。客户端会每5秒发送一个心跳包，与注册中心实时保持心跳，注册中心如果15秒都没有收到客户端发来的心跳将客户端实例设置为不健康，30秒没收到心跳时将这个临时实例摘除


心跳机制的作用？
实时获取服务下线情况


nacos不同版本中心跳机制实现方式不同
1.x：http协议定时发送
2.x：grpc（nacos服务端收到实例数据后通过grpc推送给订阅者）

nacos客户端

思想：使用定时器定时执行任务，每隔5s发送一次心跳

beatInfo中包括了心跳续约的对象信息（端口、ip、服务名、分组、权重等）

心跳是通过线程池ScheduledThreadPoolExecutor，默认每5秒执行一次

nacos服务端

在InstanceController中beat方法为发送心跳接口
服务端心跳处理流程

处理流程：

1. 获取集群名称（没有指定则使用默认的）
2. 获取续约对象的信息（ip、port、namespaceId、service-name等）
3. 处理心跳逻辑
   a. 判断clientManager中是否包含客户端注册实例
   ⅰ. 是：处理客户端心跳
   ⅱ. 否：判断是否传入beantInfo续约信息
   1. 是：创建instance实例，将实例注册到ServiceManager中
   2. 否：返回给客户端 20404状态码，表示“实例不存在”
   b. 判断ServiceManager是否包含客户端实例，不包含则返回提示“服务不存在”
   c. 判断是否心跳信息，如果不存在则创建心跳信息
   d. 立即执行心跳机制
   e. 返回心跳请求处理结果（true表示成功）

源码见下图解析：

/**
     * Create a beat for instance.
     *
     * @param request http request
     * @return detail information of instance
     * @throws Exception any error during handle
     */
    @CanDistro
    @PutMapping("/beat")
    @Secured(action = ActionTypes.WRITE)
    public ObjectNode beat(HttpServletRequest request) throws Exception {

        ObjectNode result = JacksonUtils.createEmptyJsonNode();
        result.put(SwitchEntry.CLIENT_BEAT_INTERVAL, switchDomain.getClientBeatInterval());
        //获取续约对象的信息
        String beat = WebUtils.optional(request, "beat", StringUtils.EMPTY);
        RsInfo clientBeat = null;
        //判断是否传入续约对象信息：是返回true；否返回false
        if (StringUtils.isNotBlank(beat)) {
            clientBeat = JacksonUtils.toObj(beat, RsInfo.class);
        }
        //从http请求中获取集群名称（如果请求中没有指定集群名则使用默认的集群名称）
        //HTTP请求对象request，集群名称的参数名称CommonParams.CLUSTER_NAME，以及默认的集群名称UtilsAndCommons.DEFAULT_CLUSTER_NAME
        String clusterName = WebUtils
                .optional(request, CommonParams.CLUSTER_NAME, UtilsAndCommons.DEFAULT_CLUSTER_NAME);
        //从http请求中获取ip
        String ip = WebUtils.optional(request, "ip", StringUtils.EMPTY);
        //获取端口号
        int port = Integer.parseInt(WebUtils.optional(request, "port", "0"));

        if (clientBeat != null) {
            if (StringUtils.isNotBlank(clientBeat.getCluster())) {
                clusterName = clientBeat.getCluster();
            } else {
                // fix #2533
                clientBeat.setCluster(clusterName);
            }
            ip = clientBeat.getIp();
            port = clientBeat.getPort();
        }
        //获取namespaceId（默认public）
        String namespaceId = WebUtils.optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);
        //获取服务名
        String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);
        //检查服务名
        NamingUtils.checkServiceNameFormat(serviceName);
        Loggers.SRV_LOG.debug("[CLIENT-BEAT] full arguments: beat: {}, serviceName: {}, namespaceId: {}", clientBeat,
                serviceName, namespaceId);
        BeatInfoInstanceBuilder builder = BeatInfoInstanceBuilder.newBuilder();
        builder.setRequest(request);
        //处理心跳信息逻辑并返回结果
        int resultCode = getInstanceOperator()
                .handleBeat(namespaceId, serviceName, ip, port, clusterName, clientBeat, builder);
        result.put(CommonParams.CODE, resultCode);
        result.put(SwitchEntry.CLIENT_BEAT_INTERVAL,
                getInstanceOperator().getHeartBeatInterval(namespaceId, serviceName, ip, port, clusterName));
        result.put(SwitchEntry.LIGHT_BEAT_ENABLED, switchDomain.isLightBeatEnabled());
        return result;
    }

处理心跳逻辑

//处理心跳
    @Override
    public int handleBeat(String namespaceId, String serviceName, String ip, int port, String cluster,
            RsInfo clientBeat, BeatInfoInstanceBuilder builder) throws NacosException {
        //获取续约对象信息
        Service service = getService(namespaceId, serviceName, true);
        //获取客户端ip信息：192.168.0.1:9002#true
        String clientId = IpPortBasedClient.getClientId(ip + InternetAddressUtil.IP_PORT_SPLITER + port, true);
        //从clientManager中获取当前客户端注册实例
        //补充：clientManager存储了所有与客户端建立连接的实例信息
        IpPortBasedClient client = (IpPortBasedClient) clientManager.getClient(clientId);
        //如果注册实例为空 或者 发布者信息集合当中不包含客户端实例
        if (null == client || !client.getAllPublishedService().contains(service)) {
            if (null == clientBeat) {
                return NamingResponseCode.RESOURCE_NOT_FOUND;
            }
            //创建新实例
            Instance instance = builder.setBeatInfo(clientBeat).setServiceName(serviceName).build();
            registerInstance(namespaceId, serviceName, instance);
            //将客户端实例信息添加到clientManager中
            client = (IpPortBasedClient) clientManager.getClient(clientId);
        }
        //检查ServiceManager是否包含当前服务实例：没有则抛出异常
        if (!ServiceManager.getInstance().containSingleton(service)) {
            throw new NacosException(NacosException.SERVER_ERROR,
                    "service not found: " + serviceName + "@" + namespaceId);
        }
        //判断是否存在心跳信息，如果没有则创建心跳信息
        if (null == clientBeat) {
            clientBeat = new RsInfo();
            clientBeat.setIp(ip);
            clientBeat.setPort(port);
            clientBeat.setCluster(cluster);
            clientBeat.setServiceName(serviceName);
        }
        //
        ClientBeatProcessorV2 beatProcessor = new ClientBeatProcessorV2(namespaceId, clientBeat, client);
        //立即执行心跳机制
        HealthCheckReactor.scheduleNow(beatProcessor);
        client.setLastUpdatedTime();
        //表示心跳请求处理成功
        return NamingResponseCode.OK;
    }

---

③、持久化

nacos注册表放到内存中（临时）、数据库（持久化）

1. 配置存储的持久化：
Nacos支持多种数据源进行配置的持久化，包括文件系统、数据库等。默认情况下，Nacos使用嵌入式数据库（Derby）进行持久化，但可以通过配置切换到其他数据库。


使用MySQL等数据库作为配置存储： 在Nacos的application.properties或application.yml中配置数据库相关信息。例如，使用MySQL：

spring.datasource.platform=mysql
db.num=1
db.url.0=jdbc:mysql://localhost:3306/nacos_config?characterEncoding=utf8&connectTimeout=1000&socketTimeout=3000&autoReconnect=true
db.user=root
db.password=root

这将使得Nacos将配置信息存储到MySQL数据库中。

使用文件系统作为配置存储： 配置文件系统存储也是可能的，通过修改Nacos配置文件中的nacos.standalone.data-stand 或 nacos.standalone.data-dir 属性，指定存储路径。


2. 注册中心数据的持久化：
Nacos注册中心数据（服务实例信息）的持久化同样可以使用数据库进行存储。


使用MySQL等数据库作为注册中心数据存储： 在Nacos的application.properties或application.yml中配置数据库相关信息。例如，使用MySQL：

spring.datasource.platform=mysql
db.num=1
db.url.0=jdbc:mysql://localhost:3306/nacos_registry?characterEncoding=utf8&connectTimeout=1000&socketTimeout=3000&autoReconnect=true
db.user=root
db.password=root

这将使得Nacos将注册中心数据存储到MySQL数据库中。


使用文件系统作为注册中心数据存储： 配置文件系统存储同样是可能的，通过修改Nacos配置文件中的nacos.standalone.data-stand 或 nacos.standalone.data-dir 属性，指定存储路径。

在实际生产环境中，大家可以根据具体的需求和环境来选择合适的持久化方式。通过数据库进行持久化可以提供更好的可维护性和扩展性，但需要配置数据库连接等信息。如果规模较小，也可以考虑使用文件系统进行持久化。

Nacos的客户SDK在本地生成配置的快照。当客户端无法连接到Nacos Server时，可以使用配置快照显示系统的整体容灾能力。配置快照类似于Git中的本地commit,也类似于缓存，会在适当的时机更新，但是并没有缓存过期(expiration)的概念。

---

④、读写冲突问题

注册/查询实例TPS达到13000以上，接口达到预期

注册表读写并发冲突问题是什么？
在多线程环境下，多个线程同时对注册表进行修改和读取时，会产生读写并发冲突问题

如何解决冲突问题？
CopyOnWrite思想，读写分离。在修改注册表的时候会根据原来的注册表内存结构复制一个新的注册列表，线程读取数据时读取的是原来的注册表，修改完之后会将注册列表与原来的旧注册表进行对比，替换成最终的数据，实现高并发，不加锁。
注意：这里复制出来的新注册表是针对同一个服务同一台及其是单线程注册的，只复制对应服务的结构信息，粒度很小，修改完后替换

---

三、总结

Nacos在服务注册与发现、配置管理、动态DNS等方面提供了全面而灵活的解决方案，为微服务架构的搭建和管理提供了强有力的支持。我们除了知道Nacos如何使用之外，也要知其所以然。

如果有想要交流的内容欢迎在评论区进行留言，如果这篇文档受到了您的喜欢那就留下你点赞+收藏+评论脚印支持一下博主~