Nacos源码

因为最近项目在做容器化处理，容器化后涉及到不同进程对同一个文件的读写，考虑到可能会存在同一文件的配置文件，可能会把彼此覆盖掉，所以这里学习一下Nacos源码。

整体结构图

这边主要看配置中心、服务注册中心的源码，其他也会说，但不会那么细致。

源码解析

配置服务源码

这一块源码，我们从nacos的测试的demo入手进行学习，先总概整体过程再拆分出来一个个剖析。

先找到ConfigExample，配置服务示例（建议起一个Nacos服务，对着学源码），重点看下以下代码
- 获取配置文件过程：使用nacos地址—>获取配置对象—>使用配置对象、文件名、组名获取配置信息
- 设置监听器：使用配置对象、文件名、组名配置—>必须重写两方法
- 更新配置信息：使用配置对象，把文件名、组名给上即可
- 删除配置：使用配置对象，把文件名、组名给上即可

public class ConfigExample {public static void main(String[] args) throws NacosException, InterruptedException {//*************获取配置信息****************String serverAddr = "localhost";  //设置nacos的地址String dataId = "test"; //文件名或者叫id也可以String group = "DEFAULT_GROUP"; //对应的组Properties properties = new Properties();//创造一个配置类properties.put("serverAddr", serverAddr);//将nacos服务地址设置进去//使用Nacos的ip地址，创建nacos配置服务对象（后面细说）ConfigService configService = NacosFactory.createConfigService(properties);//从配置服务对象获取配置内容（传入文件名、组名、超时时间）String content = configService.getConfig(dataId, group, 5000);System.out.println(content); //拿到的配置内容//*************增加服务配置的监听器****************configService.addListener(dataId, group, new Listener() {//输出配置信息@Overridepublic void receiveConfigInfo(String configInfo) {System.out.println("receive:" + configInfo);}//可以获取线程，当坚挺到配置的时候，执行某些任务@Overridepublic Executor getExecutor() {return null;}});//*************推送新配置信息****************//判断推送是否成功boolean isPublishOk = configService.publishConfig(dataId, group, "content");System.out.println(isPublishOk);//*************获取新的配置信息****************Thread.sleep(3000);content = configService.getConfig(dataId, group, 5000);System.out.println(content);//*************获取新的配置信息****************boolean isRemoveOk = configService.removeConfig(dataId, group);System.out.println(isRemoveOk);Thread.sleep(3000);//再次获取配置信息确保已经删除content = configService.getConfig(dataId, group, 5000);System.out.println(content);Thread.sleep(300000);}
}

下面把上面的部分拆开细讲。

获取配置文件

ctrl+鼠标左键点击getConfig进入源码查看

        Thread.sleep(3000);content = configService.getConfig(dataId, group, 5000);System.out.println(content);

点击查看实现类

进入getConfigInner进去继续查看

    @Overridepublic String getConfig(String dataId, String group, long timeoutMs) throws NacosException {return getConfigInner(namespace, dataId, group, timeoutMs);}

开始剖析getConfigInner，直接看注释即可
- 特别注意下：tenant和namespace是一个东西，从nacos的显示的日志可以看出。
- 整个过程如下：

先从本地磁盘中加载配置，因为应用在启动时，会加载远程配置缓存到本地，如果本地文件的内容不为空，直接返回。

如果本地文件的内容为空，则调用worker.getServerConfig加载远程配置

如果出现异常，则调用本地快照文件加载配置

    private String getConfigInner(String tenant, String dataId, String group, long timeoutMs) throws NacosException {//判断group是否为空，为空则设为默认group = blank2defaultGroup(group);//校验dataId、group保证不为空ParamUtils.checkKeyParam(dataId, group);//创建配置响应对象ConfigResponse cr = new ConfigResponse();//设置文件id、命名空间、组cr.setDataId(dataId);cr.setTenant(tenant);cr.setGroup(group);//优先加载本地配置String content = LocalConfigInfoProcessor.getFailover(worker.getAgentName(), dataId, group, tenant);if (content != null) {//如果本地内容不为空，则告知从本地加载成功（后面细说）LOGGER.warn("[{}] [get-config] get failover ok, dataId={}, group={}, tenant={}, config={}",worker.getAgentName(), dataId, group, tenant, ContentUtils.truncateContent(content));//将加载到的内容放入配置对象cr.setContent(content);//获取容灾配置的EncryptedDataKeyString encryptedDataKey = LocalEncryptedDataKeyProcessor.getEncryptDataKeyFailover(agent.getName(), dataId, group, tenant);//放入容灾配置的EncryptedDataKeycr.setEncryptedDataKey(encryptedDataKey);//过滤链configFilterChainManager.doFilter(null, cr);//从响应对象重新获取配置内容content = cr.getContent();//返回配置内容return content;}try {//加载远程配置，获取配置对象（后面细说）ConfigResponse response = worker.getServerConfig(dataId, group, tenant, timeoutMs, false);//从远程配置里获取配置内容，并设置cr.setContent(response.getContent());//从配置响应对象获取EncryptedDataKey，并设置EncryptedDataKeycr.setEncryptedDataKey(response.getEncryptedDataKey());//过滤链configFilterChainManager.doFilter(null, cr);//从响应对象重新获取配置内容content = cr.getContent();//返回配置内容return content;} catch (NacosException ioe) {//出现问题的情况：//请求失败、配置正在被删除、配置已不存在if (NacosException.NO_RIGHT == ioe.getErrCode()) {throw ioe;}LOGGER.warn("[{}] [get-config] get from server error, dataId={}, group={}, tenant={}, msg={}",worker.getAgentName(), dataId, group, tenant, ioe.toString());}LOGGER.warn("[{}] [get-config] get snapshot ok, dataId={}, group={}, tenant={}, config={}",worker.getAgentName(), dataId, group, tenant, ContentUtils.truncateContent(content));//从快照中获取（后面细说，在快照加载部分）content = LocalConfigInfoProcessor.getSnapshot(worker.getAgentName(), dataId, group, tenant);//放入获取到的配置内容cr.setContent(content);//获取encryptedDataKeyString encryptedDataKey = LocalEncryptedDataKeyProcessor.getEncryptDataKeyFailover(agent.getName(), dataId, group, tenant);//向配置响应对象里放入encryptedDataKeycr.setEncryptedDataKey(encryptedDataKey);//过滤链configFilterChainManager.doFilter(null, cr);//重新获取配置内容content = cr.getContent();//返回配置内容return content;}

本地配置加载

从getFailover进入查看本地配置源码

        // use local config firstString content = LocalConfigInfoProcessor.getFailover(worker.getAgentName(), dataId, group, tenant);

进入getFailover查看
- 获取本地配置地址，根据命名空间进行拼串
  - 没有命名空间：/${serverName}_nacos/data/config-data
  - 存在命名空间：/serverNamenacos/data/config−data−tenant/{serverName}_nacos/data/config-data-tenant/serverNamenacos/data/config−data−tenant/{tenant}
- 进入readFile，继续查看

    public static String getFailover(String serverName, String dataId, String group, String tenant) {File localPath = getFailoverFile(serverName, dataId, group, tenant);if (!localPath.exists() || !localPath.isFile()) {//如果文件不存在则返空return null;}        try {//读取文件return readFile(localPath);} catch (IOException ioe) {LOGGER.error("[" + serverName + "] get failover error, " + localPath, ioe);return null;}}

进入readFile方法继续查看

    protected static String readFile(File file) throws IOException {if (!file.exists() || !file.isFile()) {//判断文件路径对应的文件是否存在，不存在则返空return null;}//判断是否为多实例，多实例则采取文件锁获取if (JvmUtil.isMultiInstance()) {return ConcurrentDiskUtil.getFileContent(file, Constants.ENCODE);} else {//否则使用正常的文件打开方式读取配置即可try (InputStream is = new FileInputStream(file)) {return IoUtils.toString(is, Constants.ENCODE);}}}

进入getFileContent方法查看

    public static String getFileContent(File file, String charsetName) throws IOException {RandomAccessFile fis = null; //创建一个随机流FileLock rlock = null;    //创建一个文件锁try {//因为只读取配置，所以权限为“只读”fis = new RandomAccessFile(file, READ_ONLY);FileChannel fcin = fis.getChannel();int i = 0;do {try {//尝试获取该文件的文件锁，如果获取不到则返回nullrlock = fcin.tryLock(0L, Long.MAX_VALUE, true);} catch (Exception e) {//没有获取到则抛出异常++i;//如果尝试10次还获取不到锁则抛出异常if (i > RETRY_COUNT) {LOGGER.error("read {} fail;retryed time:{}", file.getName(), i);throw new IOException("read " + file.getAbsolutePath() + " conflict");}//休眠10ms*次数sleep(SLEEP_BASETIME * i);LOGGER.warn("read {} conflict;retry time:{}", file.getName(), i);}} while (null == rlock);//自旋锁，一直尝试获取锁//获取当前通道大小int fileSize = (int) fcin.size();//创建缓冲区ByteBuffer byteBuffer = ByteBuffer.allocate(fileSize);//把字节读入fcin.read(byteBuffer);//刷新缓冲区byteBuffer.flip();//将缓冲区字节转换为字符串return byteBufferToString(byteBuffer, charsetName);} finally {if (rlock != null) {//释放锁rlock.release();rlock = null;}if (fis != null) {//关闭流IoUtils.closeQuietly(fis);fis = null;}}}

远程配置加载

触发远程配置加载的情况是，本地配置文件为空，则会从远程中心去调用：

    public ConfigResponse getServerConfig(String dataId, String group, String tenant, long readTimeout, boolean notify)throws NacosException {if (StringUtils.isBlank(group)) {//判断组是否为空，为空则使用默认组名group = Constants.DEFAULT_GROUP;} //使用配置id、组名、命名空间、超时时间和是否使用缓存return this.agent.queryConfig(dataId, group, tenant, readTimeout, notify);}

进入queryConfig方法继续查看
- 如果notify为true，则会从缓存里找到对应的缓存数据，使用缓存数据去获取先前加载过这个配置的客户端。

        @Overridepublic ConfigResponse queryConfig(String dataId, String group, String tenant, long readTimeouts, boolean notify)throws NacosException {//使用文件id、组名、命名空间生成请求ConfigQueryRequest request = ConfigQueryRequest.build(dataId, group, tenant);//把notify放入请求的头部信息中request.putHeader(NOTIFY_HEADER, String.valueOf(notify));//以0位id，获取一个正在运行的客户端RpcClient rpcClient = getOneRunningClient();if (notify) {//查看notify是否为true，为true则从缓存里获取对应客户端CacheData cacheData = cacheMap.get().get(GroupKey.getKeyTenant(dataId, group, tenant));if (cacheData != null) {//如果缓存不为空，则获取数据对应存储的客户端使用rpcClient = ensureRpcClient(String.valueOf(cacheData.getTaskId()));}}//获取查询响应对象ConfigQueryResponse response = (ConfigQueryResponse) requestProxy(rpcClient, request, readTimeouts);//创建配置对象ConfigResponse configResponse = new ConfigResponse();if (response.isSuccess()) {//如果响应成功，则将其存储到快照中（执行逻辑：拿组名、命名空间、文件id去查询）（细说saveSnapshot）//1、首先获取快照文件（是否存在命名空间） // 不存在命名空间：/config_rpc_client_nacos/snapshot/${group}/${dataId}//    存在命名空间：/config_rpc_client_nacos/snapshot-tenant/${tenant}/${group}/${dataId}LocalConfigInfoProcessor.saveSnapshot(this.getName(), dataId, group, tenant, response.getContent());//将文件内容置入configResponse.setContent(response.getContent());//初始化配置类型String configType;//判断响应实体里的文本类型是否为空if (StringUtils.isNotBlank(response.getContentType())) {//不为空则从响应体获取文件类型configType = response.getContentType();} else {//如果响应体没有设置文件类型，则设置为TEXTconfigType = ConfigType.TEXT.getType();}//将配置文件类型给上configResponse.setConfigType(configType);//获取encryptedDataKey（理解成一个秘钥就可以）String encryptedDataKey = response.getEncryptedDataKey();//保存快照，内容为空则删除LocalEncryptedDataKeyProcessor.saveEncryptDataKeySnapshot(agent.getName(), dataId, group, tenant, encryptedDataKey);//把encryptedDataKey设置进去configResponse.setEncryptedDataKey(encryptedDataKey);//返回响应结果return configResponse;//错误码300，此时配置为null，删除同名的快照，如果快照不存在，则报错。（细说saveSnapshot）} else if (response.getErrorCode() == ConfigQueryResponse.CONFIG_NOT_FOUND) {LocalConfigInfoProcessor.saveSnapshot(this.getName(), dataId, group, tenant, null);LocalEncryptedDataKeyProcessor.saveEncryptDataKeySnapshot(agent.getName(), dataId, group, tenant, null);return configResponse;//错误码400，直接抛异常} else if (response.getErrorCode() == ConfigQueryResponse.CONFIG_QUERY_CONFLICT) {LOGGER.error("[{}] [sub-server-error] get server config being modified concurrently, dataId={}, group={}, "+ "tenant={}", this.getName(), dataId, group, tenant);throw new NacosException(NacosException.CONFLICT,"data being modified, dataId=" + dataId + ",group=" + group + ",tenant=" + tenant);} else {//其他的，也直接抛异常LOGGER.error("[{}] [sub-server-error]  dataId={}, group={}, tenant={}, code={}", this.getName(), dataId,group, tenant, response);throw new NacosException(response.getErrorCode(),"http error, code=" + response.getErrorCode() + ",msg=" + response.getMessage() + ",dataId=" + dataId + ",group=" + group+ ",tenant=" + tenant);}}

细说一下快照存储的过程
- 不存在命名空间：/config_rpc_client_nacos/snapshot/group/{group}/group/{dataId}
- 存在命名空间：/config_rpc_client_nacos/snapshot-tenant/tenant/{tenant}/tenant/{group}/${dataId}

    public static void saveSnapshot(String envName, String dataId, String group, String tenant, String config) {//查看是否开启快照功能，没有则直接返回if (!SnapShotSwitch.getIsSnapShot()) {return;}//获取路径//不存在命名空间：/config_rpc_client_nacos/snapshot/${group}/${dataId}//存在命名空间：/config_rpc_client_nacos/snapshot-tenant/${tenant}/${group}/${dataId}File file = getSnapshotFile(envName, dataId, group, tenant);if (null == config) { //如果传入的配置文件内容为空try {//删除这个配置文件IoUtils.delete(file);} catch (IOException ioe) {LOGGER.error("[" + envName + "] delete snapshot error, " + file, ioe);}} else {try {//获取父类目录File parentFile = file.getParentFile();//如果父类目录不存在if (!parentFile.exists()) {//新建目录boolean isMdOk = parentFile.mkdirs();//如果新建目录失败if (!isMdOk) {//新建目录出错LOGGER.error("[{}] save snapshot error", envName);}}if (JvmUtil.isMultiInstance()) {//如果是多实例，使用文件锁将配置文件写入文件（下面细说）ConcurrentDiskUtil.writeFileContent(file, config, Constants.ENCODE);} else {//如果是单实例，直接写入IoUtils.writeStringToFile(file, config, Constants.ENCODE);}} catch (IOException ioe) {LOGGER.error("[" + envName + "] save snapshot error, " + file, ioe);}}}

进入writeFileContent进行查看

    public static Boolean writeFileContent(File file, String content, String charsetName) throws IOException {//如果文件不存在if (!file.exists()) {//创建文件，返回失败结果boolean isCreateOk = file.createNewFile();if (!isCreateOk) {return false;}}//初始化信道，文件锁，随机流FileChannel channel = null;FileLock lock = null;RandomAccessFile raf = null;try {//随机流读取文件，给与读和写的权限raf = new RandomAccessFile(file, READ_WRITE);//获取信道channel = raf.getChannel();//初始化获取锁的次数int i = 0;do {try {//尝试获取锁lock = channel.tryLock();} catch (Exception e) {++i;//获取锁的次数+1//获取次数高于10次报错if (i > RETRY_COUNT) {LOGGER.error("write {} fail;retryed time:{}", file.getName(), i);throw new IOException("write " + file.getAbsolutePath() + " conflict");}//休眠，防止进程里的线程占用过高，导致cpu爆满sleep(SLEEP_BASETIME * i);LOGGER.warn("write {} conflict;retry time:{}", file.getName(), i);}} while (null == lock);//将配置文件内容压到缓冲区ByteBuffer sendBuffer = ByteBuffer.wrap(content.getBytes(charsetName));//将其通过信道写入while (sendBuffer.hasRemaining()) {channel.write(sendBuffer);}channel.truncate(content.length());} catch (FileNotFoundException e) {throw new IOException("file not exist");} finally {if (lock != null) {try {lock.release();lock = null;} catch (IOException e) {LOGGER.warn("close wrong", e);}}if (channel != null) {try {channel.close();channel = null;} catch (IOException e) {LOGGER.warn("close wrong", e);}}if (raf != null) {try {raf.close();raf = null;} catch (IOException e) {LOGGER.warn("close wrong", e);}}}return true;}

使用快照加载

从getSnapshot进入快照获取配置文件的源码中

    public static String getSnapshot(String name, String dataId, String group, String tenant) {//如果没有开启快照，则直接返空if (!SnapShotSwitch.getIsSnapShot()) {return null;}//获取路径//不存在命名空间：/config_rpc_client_nacos/snapshot/${group}/${dataId}//存在命名空间：/config_rpc_client_nacos/snapshot-tenant/${tenant}/${group}/${dataId}File file = getSnapshotFile(name, dataId, group, tenant);//如果文件不存在或者不是文件，则直接返空if (!file.exists() || !file.isFile()) {return null;}try {//读文件入口return readFile(file);} catch (IOException ioe) {LOGGER.error("[" + name + "]+get snapshot error, " + file, ioe);return null;}}

readFile这个方法讲过了，返回本地配置加载的readFile方法查看即可。

至此，Nacos的配置文件获取源码就讲完了。

删除配置文件

删除配置文件的入口

boolean isRemoveOk = configService.removeConfig(dataId, group);

一路追踪到removeConfig

查看配置移除的代码

        @Overridepublic boolean removeConfig(String dataId, String group, String tenant, String tag) throws NacosException {//新建一个删除配置请求的实体，传入文件Id，组名、命名空间ConfigRemoveRequest request = new ConfigRemoveRequest(dataId, group, tenant, tag);//使用网关发起请求ConfigRemoveResponse response = (ConfigRemoveResponse) requestProxy(getOneRunningClient(), request);  //返回响应结果return response.isSuccess();}

更新配置文件

从先前的demo一路追到publishConfig

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-4jn82D5y-1656468857039)(nacos源码学习/image-20220120162428117.png)]

进入publishConfig查看

    private boolean publishConfigInner(String tenant, String dataId, String group, String tag, String appName,String betaIps, String content, String type, String casMd5) throws NacosException {//查看组名是否为空，为空则使用DEFAULT_GROUP为默认组名group = blank2defaultGroup(group);//校验文件id，组名和配置文件内容ParamUtils.checkParam(dataId, group, content);//初始化配置文件实体ConfigRequest cr = new ConfigRequest();//放入文件id、组名、命名空间、类型等cr.setDataId(dataId);cr.setTenant(tenant);cr.setGroup(group);cr.setContent(content);cr.setType(type);//过滤链configFilterChainManager.doFilter(cr, null);//重新获取配置文件内容content = cr.getContent();//获取秘钥String encryptedDataKey = (String) cr.getParameter("encryptedDataKey");return worker.publishConfig(dataId, group, tenant, appName, tag, betaIps, content, encryptedDataKey, casMd5, type);}

从worker的publishConfig一路追到代理的实现

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-QZm87yD3-1656468857040)(nacos源码学习/image-20220120163230662.png)]

进入publishConfig查看

        @Overridepublic boolean publishConfig(String dataId, String group, String tenant, String appName, String tag,String betaIps, String content, String encryptedDataKey, String casMd5, String type)throws NacosException {try {//将要更新的内容，文件id，组名，命名空间放入请求中ConfigPublishRequest request = new ConfigPublishRequest(dataId, group, tenant, content);//设置请求需要的各种参数，例如文件类型等request.setCasMd5(casMd5);request.putAdditionalParam(TAG_PARAM, tag);request.putAdditionalParam(APP_NAME_PARAM, appName);request.putAdditionalParam(BETAIPS_PARAM, betaIps);request.putAdditionalParam(TYPE_PARAM, type);request.putAdditionalParam(ENCRYPTED_DATA_KEY_PARAM, encryptedDataKey);//使用网关发送请求ConfigPublishResponse response = (ConfigPublishResponse) requestProxy(getOneRunningClient(), request);//请求如果不成功返回falseif (!response.isSuccess()) {LOGGER.warn("[{}] [publish-single] fail, dataId={}, group={}, tenant={}, code={}, msg={}",this.getName(), dataId, group, tenant, response.getErrorCode(), response.getMessage());return false;} else {//成功返回trueLOGGER.info("[{}] [publish-single] ok, dataId={}, group={}, tenant={}, config={}", getName(),dataId, group, tenant, ContentUtils.truncateContent(content));return true;}} catch (Exception e) {LOGGER.warn("[{}] [publish-single] error, dataId={}, group={}, tenant={}, code={}, msg={}",this.getName(), dataId, group, tenant, "unkonw", e.getMessage());return false;}}

《一步一步看源码：Nacos》框架源码系列之一（其1，配置服务源码）相关推荐

JDK1.8源码下载及获取、导入IDEA阅读、配置JDK源码
proxmox学习使用系列--1.安装后配置软件源
Proxmox VE(PVE)+ceph+物理网络规划-超融合生产环境安装部署案例前面的安装参考别人的上面的文章吧,那里的老版本还是buster,后面的配置根据现用的bullseye做个记录,系统安 ...
Golang流媒体实战之六：lal拉流服务源码阅读
欢迎访问我的GitHub 这里分类和汇总了欣宸的全部原创(含配套源码):https://github.com/zq2599/blog_demos <Golang流媒体实战>系列的链接体验 ...
Java之SpringCloud Alibaba【一】【Nacos一篇文章精通系列】
Java之SpringCloud Alibaba[一][Nacos一篇文章精通系列] 一.微服务介绍 1.系统架构演变 1)单体应用架构 2)垂直应用架构 3)分布式 4)SOA架构 5)微服务框架 ...
openEuler虚拟机配置yum源
openEuler虚拟机配置yum源正文目录 1.首先查看系统内核情况 2.查看原yum源 3.下面开始配置 4.然后更新yum源 5.yum源配置完毕,下面做个测试(安装常用基本软件) 开始 1. ...
nacos集群的ap cp切换_配置中心Nacos
Nacos概述英文全称Dynamic Naming and Configuration Service,是指该注册/配置中心都是以服务为核心. Nacos是阿里云中间件团队开源的一个项目.项目地址: ...
电源管理-配置唤醒源
由上一次的分析可知,在suspend_ops->enter(state);中会进行唤醒源的配置.下面分析平台代码: //位于linux-3.18\arch\arm\plat-samsung\pm ...
centos配置yum源
本文主要赘述在centos系统配置yum源的两种方式. 参考文章: centos配置yum源 Yum工具详解配置外网yum源确认可以访问外网. curl www.baidu.com 查看yum源, ...
调试JDK源码-一步一步看HashMap怎么Hash和扩容
调试JDK源码-一步一步看HashMap怎么Hash和扩容调试JDK源码-ConcurrentHashMap实现原理调试JDK源码-HashSet实现原理调试JDK源码-调试JDK源码-Hash ...

《一步一步看源码：Nacos》框架源码系列之一（其1，配置服务源码）