皮皮网

1.nacosԴ?码下?????
2.微服务架构系列之Nacos 集群环境搭建
3.在Linux中进行nacos集群搭建（一台服务器）
4.2024最新版Nacos安装教程（史上最详细保姆级教程）
5.图解+源码讲解 Nacos 客户端动态监听配置机制
6.Nacos服务端源码分析(四): 拉取服务信息

nacosԴ??????

       Nacos源码剖析

       深入学习Nacos，解析源码，码下重点关注以下两点：

       源码环境搭建

       从官方项目克隆Nacos源码，码下检出1.4.1版本，码下导入IDEA。码下

       在本地MySQL中创建nacos-config数据库，码下大盗无极指标源码执行resources/META-INF/nacos-db.sql脚本创建表。码下

       修改console模块下的码下application.properties文件，配置相关参数。码下

       启动console模块的码下启动类，非集群模式启动Nacos服务端。码下

       访问本地Nacos服务：/alibaba/naco... 下载最新稳定版本的码下 nacos-server 包。

       三、码下配置集群配置文件

       解压安装包。码下

       在 Nacos 解压目录 nacos/conf 下，码下复制配置文件 cluster.conf.example 并重命名为 cluster.conf，串口示波器 源码每行配置成 ip:port。（请配置 3 个或 3 个以上节点）

       四、配置 MySQL 数据库

       Nacos 在 0.7 版本之前，默认使用嵌入式数据库 Apache Derby 存储数据（内嵌的数据库会随 Nacos 一同启动，无需额外安装）；0.7 版本及以后，增加了对 MySQL 数据源的支持。

       五、MySQL 数据源

       环境要求：MySQL 5.6.5+（生产使用建议至少主备模式，或采用高可用数据库）

       1、初始化 MySQL 数据库

       创建数据库 nacos_config。

       SQL 源文件地址：github.com/alibaba/naco...，或在 nacos-server 解压目录 conf 下找到 nacos-mysql.sql 文件，运行该文件，结果如下：

       2、application.properties 配置

       修改 nacos/conf/application.properties 文件中的brjtag 2.05源码以下内容。

       最终修改结果如下：

       如果你和我一样使用的是 MySQL 8.0+ 版本，那么启动 Nacos 时肯定会报错。莫慌，在 Nacos 安装目录下新建 plugins/mysql 文件夹，并放入 8.0+ 版本的 mysql-connector-java-8.0.xx.jar，重启 Nacos 即可，启动时会提示更换了 MySQL 的 driver-class 类。

       六、启动服务器 Linux/Unix/Mac

       在 Nacos 的解压目录 nacos/bin 下启动。

       启动命令（无参数模式，为集群模式）：

       七、查看启动记录

       可通过 /nacos/logs/nacos.log（详细日志）或 /nacos/conf/start.out（启动记录）的输出内容查看是否启动成功。

       查看命令：

       启动成功输出结果：

       八、访问

       访问以下链接，默认用户名/密码是openssl aes 源码 nacos/nacos：

       从下图可以看出，集群节点共有三个，其中 ...: 为 leader。

       九、关闭服务器

       Linux/Unix/Mac

       十、测试

       1、直连 ip 模式

       发布配置

       选择配置管理的配置列表页面，点击最右侧 + 按钮新建配置。

       获取配置

       bootstrap.yml

       使用之前的控制层代码，访问：/alibaba/nacos，Release页面为github.com/alibaba/nacos/releases。

       下载后，解压文件并进行必要的配置。在2.2.0.1和2.2.1版本中，需修改conf目录下的application.properties文件，设置nacos.core.auth.plugin.nacos.token.secret.key，spark的源码使用默认值SecretKey和VGhpc0lzTXlDdXN0bTZWNyZXRLZXkwMTIzNDU2Nzg=（用于测试，详情参阅鉴权插件）。

       Nacos默认使用端口，如果该端口被占用，需关闭占用进程或在conf目录中调整server.port。Nacos的本地数据库默认为Derby内存数据库，可修改为MySQL等其他数据库，并执行mysql-schema.sql脚本来创建表。在启动Nacos前，需进入bin目录并使用cmd执行命令，最后通过浏览器访问.0.0.1:/nacos，使用默认的nacos账号和密码登录。

       若要在SpringBoot项目中集成Nacos，需添加配置中心依赖，并在application.yml中调整，注意新版Nacos的鉴权需求，注册和配置中心的username和password通常与控制台登录密码（默认nacos/nacos）保持一致。

图解+源码讲解 Nacos 客户端动态监听配置机制

       图解+源码讲解 Nacos 客户端动态监听配置机制

       在人生中第一要紧的是发现自己。为了这个目的，各位时常需要孤独和深思 —— 南森 Nacos 源码分析系列相关文章

       从零开始看 Nacos 源码环境搭建

       图解+源码讲解 Nacos 客户端发起注册流程

       图解+源码讲解 Nacos 服务端处理注册请求逻辑

       图解+源码讲解 Nacos 客户端下线流程

       图解+源码讲解 Nacos 服务端处理下线请求

       图解+源码讲解 Nacos 客户端发起心跳请求

       图解+源码讲解 Nacos 服务端处理心跳请求

       图解+源码讲解 Nacos 服务端处理配置获取请求

       图解+源码讲解 Nacos 客户端动态监听配置机制

NacosConfigAutoConfiguration

       我们看到这里面其实注入了一个 Nacos 配置刷新的关键 NacosContextRefresherBean

@Configuration@ConditionalOnProperty(name?=?"spring.cloud.nacos.config.enabled",?matchIfMissing?=?true)public?class?NacosConfigAutoConfiguration?{ //?Nacos?配置属性@Beanpublic?NacosConfigProperties?nacosConfigProperties(ApplicationContext?context)?{ if?(context.getParent()?!=?null&&?BeanFactoryUtils.beanNamesForTypeIncludingAncestors(context.getParent(),?NacosConfigProperties.class).length?>?0)?{ return?BeanFactoryUtils.beanOfTypeIncludingAncestors(context.getParent(),NacosConfigProperties.class);}return?new?NacosConfigProperties();}//?Nacos?配置刷新属性@Beanpublic?NacosRefreshProperties?nacosRefreshProperties()?{ return?new?NacosRefreshProperties();}//?Nacos?刷新历史@Beanpublic?NacosRefreshHistory?nacosRefreshHistory()?{ return?new?NacosRefreshHistory();}//?Nacos?配置管理@Beanpublic?NacosConfigManager?nacosConfigManager(NacosConfigProperties?nacosConfigProperties)?{ return?new?NacosConfigManager(nacosConfigProperties);}//?Nacos?配置刷新@Beanpublic?NacosContextRefresher?nacosContextRefresher(NacosConfigManager?nacosConfigManager,NacosRefreshHistory?nacosRefreshHistory)?{ return?new?NacosContextRefresher(nacosConfigManager,?nacosRefreshHistory);}}NacosContextRefresher 配置中心刷新public?NacosContextRefresher(NacosConfigManager?nacosConfigManager,NacosRefreshHistory?refreshHistory)?{ //?获取配置属性信息this.nacosConfigProperties?=?nacosConfigManager.getNacosConfigProperties();//?刷新历史this.nacosRefreshHistory?=?refreshHistory;//?获取配置服务this.configService?=?nacosConfigManager.getConfigService();//?是否开启刷新，是truethis.isRefreshEnabled?=?this.nacosConfigProperties.isRefreshEnabled();}获取配置服务 getConfigService

       nacosConfigManager.getConfigService()，这行代码其实就是为了创建 NcaosConfigService 对象，我们看看你是怎么创建的，其实核心代码就是通过 NacosFactory 反射创建的 NcaosConfigService 对象，这个对象是一个核心对象后续会讲到的

public?static?ConfigService?createConfigService(Properties?properties)?throws?NacosException?{ try?{ //?加载?NacosConfigService?类Class<?>?driverImplClass?=?Class.forName("com.alibaba.nacos.client.config.NacosConfigService");//?获取构造器Constructor?constructor?=?driverImplClass.getConstructor(Properties.class);//?创建实例ConfigService?vendorImpl?=?(ConfigService)?constructor.newInstance(properties);return?vendorImpl;}?catch?(Throwable?e)?{ throw?new?NacosException(NacosException.CLIENT_INVALID_PARAM,?e);}}监听器

       NacosContextRefresher 实现了 ApplicationListener ，一看这就是一个监听器了，我们看看这个在监听器里面做了什么操作

@Overridepublic?void?onApplicationEvent(ApplicationReadyEvent?event)?{ //?这是一个?CAS?操作，只设置一次if?(this.ready.compareAndSet(false,?true))?{ //?注册?Nacos?监听器对于应用this.registerNacosListenersForApplications();}}注册 Nacos 监听/**

       register Nacos Listeners. 注册Nacos监听器 */ private void registerNacosListenersForApplications() { // 默认是 true if (isRefreshEnabled()) { // 遍历Nacos属性资源中心 for (NacosPropertySource propertySource : NacosPropertySourceRepository .getAll()) { if (!propertySource.isRefreshable()) { continue; } // 获取资源ID ?String dataId = propertySource.getDataId(); // 通过组和 dataId 注册 Nacos 监听器 registerNacosListener(propertySource.getGroup(), dataId); } } }

       private void registerNacosListener(final String groupKey, final String dataKey) { // 构建 Key 信息 String key = NacosPropertySourceRepository.getMapKey(dataKey, groupKey); // 在 listenerMap中放入了 key 对应 AbstractSharedListener 响应的方法 Listener listener = listenerMap.computeIfAbsent(key, lst -> new AbstractSharedListener() { @Override public void innerReceive(String dataId, String group, String configInfo) { // 刷新次数 refreshCountIncrement(); // 记录刷新历史，就是改变历史 nacosRefreshHistory.addRefreshRecord(dataId, group, configInfo); // 发布刷新事件 applicationContext.publishEvent( new RefreshEvent(this, null, "Refresh Nacos config")); } }); // 向配置服务中添加监听器 configService.addListener(dataKey, groupKey, listener);

       }

####?向配置服务中添加监听器&nbsp;&nbsp;&nbsp;&nbsp;此时调用的是?NacosConfigService?中的?addListener?方法，但是最终执行的还是?ClientWorker?中的?addTenantListeners?方法，后面会进行分析?ClientWorker?这个类的```java@Overridepublic?void?addListener(String?dataId,?String?group,?Listener?listener)?throws?NacosException?{ //?这个?ClientWorker?worker?也是一个核心类worker.addTenantListeners(dataId,?group,?Arrays.asList(listener));}构建 CacheData 信息

       此时向 ClientWorker 中的 CacheData 中添加数据，之后遍历监听器添加到 CacheData 中

创建 CacheData 对象public?CacheData(ConfigFilterChainManager?configFilterChainManager,?String?name,?String?dataId,?String?group,String?tenant)?{ //?dataId?不能为空if?(null?==?dataId?||?null?==?group)?{ throw?new?IllegalArgumentException("dataId="?+?dataId?+?",?group="?+?group);}this.name?=?name;this.configFilterChainManager?=?configFilterChainManager;this.dataId?=?dataId;?//?设置dataIdthis.group?=?group;?//?设置组信息this.tenant?=?tenant;?//?设置租户listeners?=?new?CopyOnWriteArrayList<ManagerListenerWrap>();?//?装饰器集合this.isInitializing?=?true;//?加载缓存数据从本地磁盘this.content?=?loadCacheContentFromDiskLocal(name,?dataId,?group,?tenant);//?计算本地缓存信息的MD5this.md5?=?getMd5String(content);}向 CacheData 中添加数据public?void?addTenantListeners(String?dataId,?String?group,List<?extends?Listener>?listeners)throws?NacosException?{ //?DefaultGroupgroup?=?null2defaultGroup(group);String?tenant?=?agent.getTenant();?//?是?""//?向缓存数据中添加监听器CacheData?cache?=?addCacheDataIfAbsent(dataId,?group,?tenant);for?(Listener?listener?:?listeners)?{ cache.addListener(listener);}}public?CacheData?addCacheDataIfAbsent(String?dataId,?String?group,?String?tenant)throws?NacosException?{ //?获取Key信息String?key?=?GroupKey.getKeyTenant(dataId,?group,?tenant);//?在缓存?Map?中获取缓存数据CacheData?cacheData?=?cacheMap.get(key);//?如果不为空的情况下那么就返回，如果为空那么就创建一个?CacheDataif?(cacheData?!=?null)?{ return?cacheData;}//?创建一个?CacheData?cacheData?=?new?CacheData(configFilterChainManager,?agent.getName(),dataId,?group,?tenant);//?将创建好的?cacheData?放入缓存?Map?中CacheData?lastCacheData?=?cacheMap.putIfAbsent(key,?cacheData);//?如果缓存数据为空的话那么从配置中心拉取，不过此时不为空if?(lastCacheData?==?null)?{ //fix?issue?#?if?(enableRemoteSyncConfig)?{ String[]?ct?=?getServerConfig(dataId,?group,?tenant,?L);cacheData.setContent(ct[0]);}//?计算任务IDint?taskId?=?cacheMap.size()?/?(int)?ParamUtil.getPerTaskConfigSize();//?设置任务IDcacheData.setTaskId(taskId);lastCacheData?=?cacheData;}//?缓存数据初始化完成//?reset?so?that?server?not?hang?this?checklastCacheData.setInitializing(true);LOGGER.info("[{ }]?[subscribe]?{ }",?agent.getName(),?key);MetricsMonitor.getListenConfigCountMonitor().set(cacheMap.size());//?返回最新的缓存数据return?lastCacheData;}

       到这里 CacheData 对象 和 cacheMap 集合已经构建完成了，后续会用到这个数据的

NacosConfigService 分析

       NacosConfigService这个类在创建的时候主要做了什么事情,这这里面创建了一个 ClientWorker对象，这个对象是一个核心的类，有关于配置的一些操作都是归功于 ClientWorker类

public?NacosConfigService(Properties?properties)?throws?NacosException?{ ......this.agent?=?new?MetricsHttpAgent(new?ServerHttpAgent(properties));this.agent.start();//?核心工作类this.worker?=?new?ClientWorker(this.agent,this.configFilterChainManager,?properties);}核心配置类 ClientWorker

       分析一下这个类都在做什么事情，都有哪些核心方法 其实能看到里面有一个构造函数、添加缓存数据、添加监听器、检查配置中心相关方法、获取服务配置、解析数据响应、移除缓存数据、删除监听器以及 shutdown方法

构造函数

       看到这里其实看到了定义了两个调度线程池，一个是用于配置检测的，一个是用于执行长轮询服务的

@SuppressWarnings("PMD.ThreadPoolCreationRule")public?ClientWorker(final?HttpAgent?agent,final?ConfigFilterChainManager?configFilterChainManager,?final?Properties?properties){ this.agent?=?agent;this.configFilterChainManager?=?configFilterChainManager;//?初始化操作init(properties);//?定义一个调度线程池，只有一个线程还是守护线程this.executor?=?Executors.newScheduledThreadPool(1,?new?ThreadFactory()?{ @Overridepublic?Thread?newThread(Runnable?r)?{ Thread?t?=?new?Thread(r);t.setName("com.alibaba.nacos.client.Worker."?+?agent.getName());t.setDaemon(true);return?t;}});//?定义一个多个线程的调度线程池，线程个数和CPU?核心数有关，也是守护线程，是一个长轮询this.executorService?=?Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors(),?new?ThreadFactory()?{ @Overridepublic?Thread?newThread(Runnable?r)?{ Thread?t?=?new?Thread(r);t.setName("com.alibaba.nacos.client.Worker.longPolling."?+agent.getName());t.setDaemon(true);return?t;}});//?定义一个定时的调度任务，第一次执行的时候延时1毫秒，后续毫秒调度一次this.executor.scheduleWithFixedDelay(new?Runnable()?{ @Overridepublic?void?run()?{ try?{ //?检查配置信息方法checkConfigInfo();}?catch?(Throwable?e)?{ LOGGER.error("["?+?agent.getName()?+?"]?"+?"[sub-check]?rotate?check?error",?e);}}},?1L,?L,?TimeUnit.MILLISECONDS);}检查配置服务方法

       这个 cacheMap 包含了一些任务信息，这里面的任务是怎么来的呢，他是在添加监听器的时候添加的，上面已经分析过了

public?NacosContextRefresher(NacosConfigManager?nacosConfigManager,NacosRefreshHistory?refreshHistory)?{ //?获取配置属性信息this.nacosConfigProperties?=?nacosConfigManager.getNacosConfigProperties();//?刷新历史this.nacosRefreshHistory?=?refreshHistory;//?获取配置服务this.configService?=?nacosConfigManager.getConfigService();//?是否开启刷新，是truethis.isRefreshEnabled?=?this.nacosConfigProperties.isRefreshEnabled();}0长轮询任务 LongPollingRunnable

Nacos服务端源码分析(四): 拉取服务信息

       本文深入解析Nacos服务端源码，特别关注服务信息的主动拉取机制。主动拉取服务信息的URL为：https://localhost:/nacos/v1/ns/instance/list。依据此URL，Nacos服务端会处理请求，具体操作如下：

       首先，获取并校验参数，随后调用`getInstanceOperator().listInstance()`函数。

       `getInstanceOperator().listInstance()`执行流程如下：

       通过`createIpPortClientIfAbsent()`确保client管理正常，若未存在则加入`clients`。

       调用`clientOperationService.subscribeService()`发布事件`ClientOperationEvent.ClientSubscribeServiceEvent`，进行服务订阅。

       调用`ServiceUtil.selectInstancesWithHealthyProtection()`获取serviceInfo，包括实例列表。

       分析各个方法的内部逻辑：

       `createIpPortClientIfAbsent()`：若`clientManager`中不存在指定`clientId`，则加入`clients`。

       `clientOperationService.subscribeService()`：发布事件`ClientOperationEvent.ClientSubscribeServiceEvent`，涉及订阅操作，将服务作为key，保存在`subscriberIndexes`中。首次添加时，会触发事件`ServiceEvent.ServiceSubscribedEvent`，将服务信息推送至订阅客户端。

       `ServiceUtil.selectInstancesWithHealthyProtection()`：整合相关信息，筛选健康的服务实例，最终返回。

       总结以上分析，Nacos服务端主动拉取服务信息的过程涉及参数验证、事件发布、实例筛选等关键步骤。这一机制确保了服务信息的及时更新与准确传递。

       下篇文章预告：探讨Nacos之Distro协议的理论基础。