本文分析一下,Nacos 在启动的时候是怎么将所有的配置文件信息 Dump 到磁盘上的; 读完本文,你将了解到以下知识
[x] 项目启动之初 Dump 配置数据的流程
[x] 如何快速启动(isQuickStart：快速启动不用全量 Dump 配置信息)
[x] DiskUtil.heartBeatFile 这个心跳文件的作用
[x] 全量 Dump 执行器 DumpAllProcessor
[x] AggrWhitelist 这个类是作用是什么
[x] ClientIpWhiteList 作用是什么
[x] SwitchService 作用是什么
[x] Dump 配置数据的事件通知
Dump 文件的类在 DumpService 中,所以我们主要分析这个类
DumpService 初始化
Spring 启动加载时,会执行带有@PostConstruct 注解的初始化方法;
@PostConstruct public void init() { DumpAllProcessor dumpAllProcessor = new DumpAllProcessor(this); dumpTaskMgr = new TaskManager("com.alibaba.nacos.server.DumpTaskManager",new DumpProcessor(this)); dumpAllTaskMgr = new TaskManager("com.alibaba.nacos.server.DumpAllTaskManager",dumpAllProcessor); try { //全量Dump配置信息 dumpConfigInfo(dumpAllProcessor); DiskUtil.clearAllBeta(); if (persistService.isExistTable(BETA_TABLE_NAME)) { new DumpAllBetaProcessor(this).process(DumpAllBetaTask.TASK_ID, new DumpAllBetaTask()); } // 更新Tag缓存 DiskUtil.clearAllTag(); if (persistService.isExistTable(TAG_TABLE_NAME)) { new DumpAllTagProcessor(this).process(DumpAllTagTask.TASK_ID, new DumpAllTagTask()); } // add to dump aggr List<ConfigInfoChanged> configList = persistService.findAllAggrGroup(); if (!CollectionUtils.isEmpty(configList)) { total = configList.size(); List<List<ConfigInfoChanged>> splitList = splitList(configList, INIT_THREAD_COUNT); for (List<ConfigInfoChanged> list : splitList) { MergeAllDataWorker work = new MergeAllDataWorker(list); work.start(); } log.info("server start, schedule merge end."); } } catch (Exception e) { } if (!STANDALONE_MODE) { //Write the current time to the status/heartBeat.txt file every 10 seconds TimerTaskService.scheduleWithFixedDelay(()-> heartbeat(), 0, 10, TimeUnit.SECONDS); long initialDelay = new Random().nextInt(INITIAL_DELAY_IN_MINUTE) + 10; //Full Dump every DUMP_ALL_INTERVAL_IN_MINUTE minutes TimerTaskService.scheduleWithFixedDelay(()-> dumpAllTaskMgr.addTask(DumpAllTask.TASK_ID, new DumpAllTask()) , initialDelay, DUMP_ALL_INTERVAL_IN_MINUTE, TimeUnit.MINUTES); //Full Dump every DUMP_ALL_INTERVAL_IN_MINUTE minutes TimerTaskService.scheduleWithFixedDelay(()-> dumpAllTaskMgr.addTask(DumpAllBetaTask.TASK_ID, new DumpAllBetaTask()), initialDelay, DUMP_ALL_INTERVAL_IN_MINUTE, TimeUnit.MINUTES); } //Try to Delete the expiration history configuration record every ten minutes TimerTaskService.scheduleWithFixedDelay(()-> clearConfigHistory(), 10, 10, TimeUnit.MINUTES);
}
复制代码
上面的代码就是 DumpService 初始化之后调用的初始化方法;我们现在来深度解析一下究竟做了哪些事情呢？
全量 Dump 配置信息
类中的方法 dumpConfigInfo(dumpAllProcessor); 里面主要是将数据库中的所有 ConfigInfo 查询出来写到服务器的磁盘中; 方法中传入了一个 dumpAllProcessor 对象; 这个是一个TaskProcessor 任务处理器; 在上一篇文章中我们介绍了【Nacos源码 三】TaskManager 任务管理的使用 ； 所有看这里就很容易理解了; dumpAllProcessor中有个process()方法; 最终执行任务的时候就是执行这个方法的; 那我们进入dumpConfigInfo看看他做了什么?
dumpConfigInfo 方法保存配置数据到磁盘
private void dumpConfigInfo(DumpAllProcessor dumpAllProcessor) throws IOException { int timeStep = 6; Boolean isAllDump = true; // initial dump all FileInputStream fis = null; Timestamp heartheatLastStamp = null; try { if (isQuickStart()) { File heartbeatFile = DiskUtil.heartBeatFile(); if (heartbeatFile.exists()) { fis = new FileInputStream(heartbeatFile); String heartheatTempLast = IOUtils.toString(fis, Constants.ENCODE); heartheatLastStamp = Timestamp.valueOf(heartheatTempLast); if (TimeUtils.getCurrentTime().getTime() - heartheatLastStamp.getTime() < timeStep * 60 * 60 * 1000) { isAllDump = false; } } } if (isAllDump) { DiskUtil.clearAll(); dumpAllProcessor.process(DumpAllTask.TASK_ID, new DumpAllTask()); } else { Timestamp beforeTimeStamp = getBeforeStamp(heartheatLastStamp,timeStep); DumpChangeProcessor dumpChangeProcessor = new DumpChangeProcessor( this, beforeTimeStamp, TimeUtils.getCurrentTime()); dumpChangeProcessor.process(DumpChangeTask.TASK_ID,new DumpChangeTask()); TimerTaskService.scheduleWithFixedDelay(()-> checkMd5AndDumpChange(), 0, 12, TimeUnit.HOURS); } } catch (IOException e) { } finally { } }
复制代码
这里的代码 ,是我改动过的,源码里面这个方法写的很乱,不利于阅读,我给重构了一下,并且提了一个 PR;不知道 Nacos 有没有给我合并进去;但是改动之后的左右是没有变化的;
isQuickStart() 判断是否快速启动
val = env.getProperty("isQuickStart");if (val != null && TRUE_STR.equals(val)) { isQuickStart = true;}
复制代码
这个方法是判断我们在启动的时候是否指定了 快速启动; 默认为 false; 如果想快速启动的话,可以在配置文件 application.properties 加上一行
或者还可以设置 Jvm 系统属性 ;在启动脚本中加上 参数
JAVA_OPT="${JAVA_OPT} -Dnacos.home=${BASE_DIR}"##加入isQuickStart=true JAVA_OPT="${JAVA_OPT} -DisQuickStart=true"
复制代码
注意,这个一定不要放在后面;就放在-Dnacos.home 后面就行,不然不会生效; 为什么在 Jvm 中设置的-D 属性能够在 Spring 中的 Environment 中获取到？ 在另外一篇文章中有讲解到;Spring中配置优先级
DiskUtil.heartBeatFile 获取心跳文件
这个方法是获取心跳文件的方法,心跳文件在 {NACOS_HOME}/status/heartBeat.txt ; 这个{NACOS_HOME} 是获取 Jvm 属性nacos.home
System.getProperty("nacos.home")
复制代码
因为在startup.sh 启动脚本中定义的是
export BASE_DIR=`cd $(dirname $0)/..; pwd`JAVA_OPT="${JAVA_OPT} -Dnacos.home=${BASE_DIR}"
复制代码
其中dirname $0 表示的是执行当前脚本的路径,-Dnacos.home 就是设置了当前路径; 如果你想改成其他的路径可以改启动脚本;比如改成
JAVA_OPT="${JAVA_OPT} -Dnacos.home=/Users/shirenchuang/mynacos"
复制代码
如果这里不设置这个 Jvm 属性nacos.home ;那么{NACOS_HOME} 就会获取系统属性并且加上 后缀File.separator + "nacos"
System.getProperty("user.home")
复制代码
这个是操作系统属性,比如我的 Mac 电脑的user.home是 /Users/shirenchuang ; 那么最终的{NACOS_HOME}=/Users/shirenchuang/nacos
{NACOS_HOME}/status/heartBeat.txt ;是一个心跳文件,每十秒就会把当前时间写入到这个文件中;
heartBeat.txt 的作用是什么？为什么要每十秒保存当前时间
因为为了能够快速启动应用,那么我们在启动的时候,可以选择不需要全部 Dump 所有的配置文件,因为上一次可能已经 Dump 了文件在磁盘中了,每次启动项目,都 DumpAll 配置,如果配置很大的话,走 IO 还是会花费一定的时间的;所以我们每十秒来持久化一次当前时间,用于记录上一次服务正常距离现在有多长时间;
假设服务宕机了,半个小时之后才启动成功,那么我们只需要将这半小时之内数据库中的配置变化重新 Dump 到磁盘中就行了,不需要 DumpAll; 可以看到代码
if (isQuickStart()) { //如果上一次服务正常的时间距离现在不超过6个小时; //那么设置 isAllDump = false;表示不需要全量Dump}
复制代码
如果上一次服务正常的时间距离现在不超过 6 个小时;那么设置 isAllDump = false;表示不需要全量 Dump;
DumpAllProcessor 执行器做全量 Dump
这个执行器就是将数据库中的所有 ConfigInfo 全量 Dump 到磁盘中去
@Override public boolean process(String taskType, AbstractTask task) { long currentMaxId = persistService.findConfigMaxId(); long lastMaxId = 0; while (lastMaxId < currentMaxId) { Page<PersistService.ConfigInfoWrapper> page = persistService.findAllConfigInfoFragment(lastMaxId, PAGE_SIZE); if (page != null && page.getPageItems() != null) { for (PersistService.ConfigInfoWrapper cf : page.getPageItems()) { long id = cf.getId(); lastMaxId = id > lastMaxId ? id : lastMaxId; if (cf.getDataId().equals(AggrWhitelist.AGGRIDS_METADATA)) { AggrWhitelist.load(cf.getContent()); } if (cf.getDataId().equals(ClientIpWhiteList.CLIENT_IP_WHITELIST_METADATA)) { ClientIpWhiteList.load(cf.getContent()); }
if (cf.getDataId().equals(SwitchService.SWITCH_META_DATAID)) { SwitchService.load(cf.getContent()); }
boolean result = ConfigService.dump(cf.getDataId(), cf.getGroup(), cf.getTenant(), cf.getContent(), cf.getLastModified());
final String content = cf.getContent(); final String md5 = MD5.getInstance().getMD5String(content); LogUtil.dumpLog.info("[dump-all-ok] {}, {}, length={}, md5={}", GroupKey2.getKey(cf.getDataId(), cf.getGroup()), cf.getLastModified(), content.length(), md5); } defaultLog.info("[all-dump] {} / {}", lastMaxId, currentMaxId); } else { lastMaxId += PAGE_SIZE; } } return true; }
复制代码
以上代码很容易看懂,但是有几个特别的地方我们需要详细讲一下;
AggrWhitelist 这个类是什么？作用是什么？
static public final String AGGRIDS_METADATA = "com.alibaba.nacos.metadata.aggrIDs";
复制代码
这个是 Nacos 立马自定义的一个 DataId; 如果 ConfigInfo 的 DataId 是这个值的话就会被单独解析,将 Content 内容解析到
AtomicReference<List<Pattern>> AGGR_DATAID_WHITELIST
复制代码
Content 设置的值可以是一个正则表达式;例如
这个 AggrWhitelist 作用是什么呢？真实意图我也不清楚;但是我只在一个地方看到有调用到这些数据；
/** * 增加或更新非聚合数据。 * * @throws NacosException */ @RequestMapping(method = RequestMethod.POST) @ResponseBody public Boolean publishConfig{ //省略.... if (AggrWhitelist.isAggrDataId(dataId)) { log.warn("[aggr-conflict] {} attemp to publish single data, {}, {}", RequestUtil.getRemoteIp(request), dataId, group); throw new NacosException(NacosException.NO_RIGHT, "dataId:" + dataId + " is aggr"); } //省略.... }
复制代码
在增加非聚合数据的时候,如果这些 DataId 刚好匹配到了,就不让添加; 使用场景例如: 我不希望开发者 添加 DataId= com.shirc 开头 的配置;那么我可以怎么做呢? 我们可以在 DataId = com.alibaba.nacos.metadata.aggrIDs 中添加一行数据
如下在这里插入图片描述
ClientIpWhiteList 是什么？作用是什么其实看了 AggrWhitelist 之后,这个 ClientIpWhiteList 是跟它是差不多意思,也是 Nacos 自己定义的一个预留配置 DataId
static public final String CLIENT_IP_WHITELIST_METADATA = "com.alibaba.nacos.metadata.clientIpWhitelist";
复制代码
值是一个 Json
{ "isOpen": true, "ips": ["127.0.0.1", "127.0.0.2"]}
复制代码
它的作用,项目中暂时没有使用到,这算是一个预留配置; Ip 白名单;
SwitchService 是什么？作用是什么？
同上,也是 Nacos 内部预留的一个配置;DataId 是 com.alibaba.nacos.meta.switch ; 开发者可以配置这个里面的属性,来进行一些设置内部属性的操作; 例如已经 Nacos 中的开关属性有:
## 是否开启固定长轮询isFixedPolling=false/true##固定长轮询的间隔时间 只有isFixedPolling=true才生效fixedPollingInertval=1000##延迟时间fixedDelayTime=500
复制代码
在这里插入图片描述
ConfigService.dump 真正的磁盘写入操作
这个方法首先将配置保存到磁盘文件中,并且缓存配置信息的 MD5 到内存中;如果配置信息不一致（MD5 不一致）,则将会发送一个通知事件 LocalDataChangeEvent 告知本地数据有更改; 这里的源码就不贴了,我概述一下操作流程 CacheItem 是配置信息的对象;保存着配置信息的一些信息,但是没有保存 Content,只保存了 content 的 MD5;
1.如果内存中没有当前配置的缓存 CacheItem,则组装对象保存进去;这个时候的 md5 是空字符串;
2.计算 content 的 MD5;跟内存 CacheItem 中的 md5 做比较(第一次肯定不相等),如果不相等则将文件保存到磁盘中;
DiskUtil.saveToDisk(dataId, group, tenant, content);
复制代码
3.如果 MD5 不相同,则更新 CacheItem 中的 MD5 属性和 lastModifiedTs 属性；lastModifiedTs 是表示最后更新时间
4.如果 MD5 不相同,还要发送通知告知数据有变更;
EventDispatcher.fireEvent(new LocalDataChangeEvent(groupKey));
复制代码
EventDispatcher 是一个时间分发类,LocalDataChangeEvent 是本地数据变更事件；他们的使用我在 文章 【Nacos源码 二】Nacos中的事件发布与订阅--观察者模式 中有比较详细的介绍; 现在我们主要来看下是哪个监听器收听了 LocalDataChangeEvent 事件; 通过反查代码我们找到 LongPollingService 这个监听类;是监听了这个事件的;
@Override public void onEvent(Event event) { if (isFixedPolling()) { // ignore } else { if (event instanceof LocalDataChangeEvent) { LocalDataChangeEvent evt = (LocalDataChangeEvent)event; scheduler.execute(new DataChangeTask(evt.groupKey, evt.isBeta, evt.betaIps)); } } }
复制代码
isFixedPolling()就是上面介绍过的 SwitchService 配置中的一个属性 isFixedPolling; 是否固定长轮询
看代码最终执行的任务是 DataChangeTask ,这个任务做了数据更改之后的操作; 源码可以自己查看,这里就不列出来了,因为这里可以单独写一篇文章; 这里打上一个 TODO... 后面回来补上; 简要概述这里的操作;
遍历所有的长轮询订阅者者(怎么订阅上的？后面会有文章介绍)
如果是 beta 发布且不在 beta 列表直接跳过
如果 tag 发布且不在 tag 列表直接跳过
发送 Http 请求通知所有未被上面 2、3 过滤掉的的订阅者最新的配置数据 ConfigInfo
Nacos 写入本地的 Dump 文件地址
在这里插入图片描述
ConfigService.updateMd5() 更新 Md5 并通知
这个方法就是上面 ConfigService.dump 的 3、4 两个步骤
如果 MD5 不相同,则更新 CacheItem 中的 MD5 属性和 lastModifiedTs 属性；lastModifiedTs 是表示最后更新时间
如果 MD5 不相同,还要发送通知告知数据有变更;
如果是快速启动,并且 isAllDump = false 是如何 Dump 数据的
上面介绍的是全量 Dump 信息,那么如果我们启动的时候是isQuickStart=true;并且上一次心跳(也就是上一次服务正常的时候)距离现在不超过 6 小时;那么就不是全量 Dump 了；isAllDump = false；那么部分 Dump 是如何操作的呢？ 有个执行器是 DumpChangeProcessor ;这个执行器看名字就知道是 Dump 有变化的数据的执行器,代码不看了;下面概述流程;
DumpChangeProcessor
查询所有的配置文件执行 ConfigService.updateMd5()，这个方法做了什么看上面；其实就是将所有的配置文件缓存到内存中，并通知所有订阅的客户端
从 his_config_info 历史表中找到从上一次心跳时间(heartBeat.txt)到现在的所有被删除记录,his_config_info 记录的就是历史的配置文件;
遍历 2 中的拿到的历史配置数据的 dataId,group,Tenant;然后去 config_info 表中查找能不能查到数据如果能查到,说明配置不是被删除了,只是修改了 content；如果不能查到,说明整个配置文件都被删除了;如果文件被删除了;则调用方法ConfigService.remove 将磁盘对应的配置文件删除;并且通知订阅的客户端数据变更;
config_info 表总查找 从上一次心跳时间(heartBeat.txt)到现在的所有有被修改过的配置数据,然后执行ConfigService.dumpChange 将这个改过的配置 Dump 的磁盘中,并通知;
然后 load Nacos 内置的一些 DataId 配置; 上面提及的 ClientIpWhiteList 、AggrWhitelist、SwitchService ;
Dump 所有的 Beat 配置 DumpAllBetaProcessor
Dump 所有的 Tag 配置 DumpAllTagProcessor
上面两个基本是跟上面讲的都一样, Beat 是灰度发布的配置; 将这些特殊的数据 Dump 到磁盘中,并且更新内存中的缓存;然后通知到那些 被配置了灰度发布的 Ip 白名单的订阅者;
Aggr 合并聚合数据
// add to dump aggr List<ConfigInfoChanged> configList = persistService.findAllAggrGroup(); if (!CollectionUtils.isEmpty(configList)) { total = configList.size(); List<List<ConfigInfoChanged>> splitList = splitList(configList, INIT_THREAD_COUNT); for (List<ConfigInfoChanged> list : splitList) { MergeAllDataWorker work = new MergeAllDataWorker(list); work.start(); } log.info("server start, schedule merge end."); }
复制代码
这里回头再详细说明
每隔十秒钟将当前时间保持到心跳文件中
TimerTaskService.scheduleWithFixedDelay(()-> heartbeat(), 0, 10, TimeUnit.SECONDS);
复制代码
每隔 6 个小时全量 Dump 一次数据
//Full Dump every DUMP_ALL_INTERVAL_IN_MINUTE minutes TimerTaskService.scheduleWithFixedDelay(()-> dumpAllTaskMgr.addTask(DumpAllTask.TASK_ID, new DumpAllTask()) , initialDelay, DUMP_ALL_INTERVAL_IN_MINUTE, TimeUnit.MINUTES);
复制代码
每十分钟执行一次清空历史记录操作
his_config_info 存放的是历史记录,可以用于回滚操作; 每十分钟执行一次尝试删除操作; 删除的是超过 N 天的历史记录; 这个 N 是多少？默认是 30 天; 但是可以通过配置修改
nacos.config.retention.days=100
复制代码
删除历史记录的方法
/**Delete the expiration history configuration record**/ private void clearConfigHistory() { log.warn("clearConfigHistory start"); if (ServerListService.isFirstIp()) { try { Timestamp startTime = getBeforeStamp(TimeUtils.getCurrentTime(), 24 * getRetentionDays()); int totalCount = persistService.findConfigHistoryCountByTime(startTime); if (totalCount > 0) { int pageSize = 1000; int removeTime = (totalCount + pageSize - 1) / pageSize; log.warn("clearConfigHistory, getBeforeStamp:{}, totalCount:{}, pageSize:{}, removeTime:{}", new Object[] {startTime, totalCount, pageSize, removeTime}); while (removeTime > 0) { // 分页删除，以免批量太大报错 persistService.removeConfigHistory(startTime, pageSize); removeTime--; } } } catch (Throwable e) { log.error("clearConfigHistory error", e); } } }
复制代码
dumpConfigInfo()方法总结
光是 dumpConfigInfo 这一个方法就写了这么长的篇幅,我们给这个方法来做一个小结;
[ ] 全量 Dump
数据表 config_info 中查询所有的配置数据;
加载 Nacos 内置的特殊配置数据ClientIpWhiteList 、AggrWhitelist、SwitchService
分别对应的 dataId= com.alibaba.nacos.metadata.aggrIDs、com.alibaba.nacos.metadata.clientIpWhitelist、com.alibaba.nacos.meta.switch ;
将所有配置文件保存到磁盘中;
将所有配置文件缓存在内存中
将配置文件通过 Http 通知到客户端中
满足下面这个的条件是
配置文件需要配置 isQuickStart=true 2. 上一次服务心跳的时间;距离现在不超过 6 个小时
[ ] 快速启动,非全量 Dump
查询表 config_info 所有数据,将他们都缓存到内存中,并且通知到订阅的客户端，注意这个时候是没有 Dump 文件到磁盘的
查询所有已经被删除的配置文件; 然后到本地磁盘中也把这些配置文件删除
查询所有这段时间(上一次服务心跳的时间;距离现在的时间)有过更改的配置文件;然后把本地磁盘中的文件更新一下;
加载 Nacos 内置的特殊配置数据ClientIpWhiteList 、AggrWhitelist、SwitchService
分别对应的 dataId= com.alibaba.nacos.metadata.aggrIDs、com.alibaba.nacos.metadata.clientIpWhitelist、com.alibaba.nacos.meta.switch ;
问题
既然我们把数据 Dump 磁盘中,那么这么做的意义是什么呢? 它在哪里被用到了呢？
【Nacos源码之配置管理 五】为什么把配置文件Dump到磁盘中
评论