太原做网站费用/百度竞价代运营

又是美好的一天呀~
个人博客地址： huanghong.top

服务端启动时的配置事件推送

服务端启动初始化DumpService

//本地存储，基于derby数据库(与jvm共用内存，内嵌式数据库)
//com.alibaba.nacos.config.server.service.dump.ExternalDumpService
@Conditional(ConditionOnEmbeddedStorage.class)
@Component
public class EmbeddedDumpService extends DumpService//外部存储(mysql)
//com.alibaba.nacos.config.server.service.dump.EmbeddedDumpService
@Conditional(ConditionOnExternalStorage.class)
@Component
@DependsOn({"rpcConfigChangeNotifier"})
public class ExternalDumpService extends DumpService

@Conditional(ConditionOnEmbeddedStorage.class)

//com.alibaba.nacos.config.server.configuration.ConditionOnExternalStorage
public class ConditionOnExternalStorage implements Condition {@Overridepublic boolean matches(ConditionContext context, AnnotatedTypeMetadata metadata) {//判断是否为外部存储return !PropertyUtil.isEmbeddedStorage();}
}//单机且配置spring.datasource.platform配置不开启
private static boolean embeddedStorage = EnvUtil.getStandaloneMode();

配置本地存储

EmbeddedDumpService后置处理

//com.alibaba.nacos.config.server.service.dump.EmbeddedDumpService#init
@PostConstruct
@Override
protected void init() throws Throwable {//单节点启动if (EnvUtil.getStandaloneMode()) {//执行配置文件dump操作dumpOperate(processor, dumpAllProcessor, dumpAllBetaProcessor, dumpAllTagProcessor);return;}//集群模式(使用本地存储)//获取CP协议CPProtocol protocol = protocolManager.getCpProtocol();AtomicReference<Throwable> errorReference = new AtomicReference<>(null);CountDownLatch waitDumpFinish = new CountDownLatch(1);// watch path => /nacos_config/leader/ has value ?//观察nacos_config中leader是否有值Observer observer = new Observer() {@Overridepublic void update(Observable o) {if (!(o instanceof ProtocolMetaData.ValueItem)) {return;}final Object arg = ((ProtocolMetaData.ValueItem) o).getData();GlobalExecutor.executeByCommon(() -> {// must make sure that there is a value here to perform the correct operation that followsif (Objects.isNull(arg)) {return;}// Identify without a timeout mechanismEmbeddedStorageContextUtils.putExtendInfo(Constants.EXTEND_NEED_READ_UNTIL_HAVE_DATA, "true");// Remove your own listening to avoid task accumulationboolean canEnd = false;//如果由于Raft状态机的内部问题而导致读取失败，则无法通过重试来补救//如果只是普通的读取失败，可以通过重试来解决。//循环重试导出数据for (; ; ) {try {dumpOperate(processor, dumpAllProcessor, dumpAllBetaProcessor, dumpAllTagProcessor);protocol.protocolMetaData().unSubscribe(Constants.CONFIG_MODEL_RAFT_GROUP, MetadataKey.LEADER_META_DATA, this);canEnd = true;} catch (Throwable ex) {if (!shouldRetry(ex)) {errorReference.set(ex);canEnd = true;}}if (canEnd) {ThreadUtils.countDown(waitDumpFinish);break;}ThreadUtils.sleep(500L);}EmbeddedStorageContextUtils.cleanAllContext();});}};protocol.protocolMetaData().subscribe(Constants.CONFIG_MODEL_RAFT_GROUP, MetadataKey.LEADER_META_DATA, observer);// We must wait for the dump task to complete the callback operation before// continuing with the initializationThreadUtils.latchAwait(waitDumpFinish);// If an exception occurs during the execution of the dump task, the exception// needs to be thrown, triggering the node to start the failed processfinal Throwable ex = errorReference.get();if (Objects.nonNull(ex)) {throw ex;}
}

dumpOperate

//com.alibaba.nacos.config.server.service.dump.DumpService#dumpOperate
protected void dumpOperate(DumpProcessor processor, DumpAllProcessor dumpAllProcessor,DumpAllBetaProcessor dumpAllBetaProcessor, DumpAllTagProcessor dumpAllTagProcessor) throws NacosException {String dumpFileContext = "CONFIG_DUMP_TO_FILE";TimerContext.start(dumpFileContext);try {LogUtil.DEFAULT_LOG.warn("DumpService start");//导出所有配置信息任务Runnable dumpAll = () -> dumpAllTaskMgr.addTask(DumpAllTask.TASK_ID, new DumpAllTask());//导出所有beta任务Runnable dumpAllBeta = () -> dumpAllTaskMgr.addTask(DumpAllBetaTask.TASK_ID, new DumpAllBetaTask());//导出所有配置信息任务Runnable dumpAllTag = () -> dumpAllTaskMgr.addTask(DumpAllTagTask.TASK_ID, new DumpAllTagTask());//清除历史配置信息Runnable clearConfigHistory = () -> {LOGGER.warn("clearConfigHistory start");if (canExecute()) {try {Timestamp startTime = getBeforeStamp(TimeUtils.getCurrentTime(), 24 * getRetentionDays());int pageSize = 1000;LOGGER.warn("clearConfigHistory, getBeforeStamp:{}, pageSize:{}", startTime, pageSize);//执行delete SQL删除his_config_info中历史数据persistService.removeConfigHistory(startTime, pageSize);} catch (Throwable e) {LOGGER.error("clearConfigHistory error : {}", e.toString());}}};try {//导出配置信息dumpConfigInfo(dumpAllProcessor);// update Beta cacheLogUtil.DEFAULT_LOG.info("start clear all config-info-beta.");//更新beta缓存DiskUtil.clearAllBeta();if (persistService.isExistTable(BETA_TABLE_NAME)) {dumpAllBetaProcessor.process(new DumpAllBetaTask());}// update Tag cacheLogUtil.DEFAULT_LOG.info("start clear all config-info-tag.");//更新标签缓存DiskUtil.clearAllTag();if (persistService.isExistTable(TAG_TABLE_NAME)) {dumpAllTagProcessor.process(new DumpAllTagTask());}// add to dump aggrList<ConfigInfoChanged> configList = persistService.findAllAggrGroup();if (configList != null && !configList.isEmpty()) {total = configList.size();List<List<ConfigInfoChanged>> splitList = splitList(configList, INIT_THREAD_COUNT);for (List<ConfigInfoChanged> list : splitList) {MergeAllDataWorker work = new MergeAllDataWorker(list);work.start();}LOGGER.info("server start, schedule merge end.");}} catch (Exception e) {LogUtil.FATAL_LOG.error("Nacos Server did not start because dumpservice bean construction failure :\n" + e);throw new NacosException(NacosException.SERVER_ERROR,"Nacos Server did not start because dumpservice bean construction failure :\n" + e.getMessage(),e);}//非单节点模式if (!EnvUtil.getStandaloneMode()) {//持久化心跳文件Runnable heartbeat = () -> {String heartBeatTime = TimeUtils.getCurrentTime().toString();// write disktry {// status/heartBeat.txtDiskUtil.saveHeartBeatToDisk(heartBeatTime);} catch (IOException e) {LogUtil.FATAL_LOG.error("save heartbeat fail" + e.getMessage());}};//每10s执行一次ConfigExecutor.scheduleConfigTask(heartbeat, 0, 10, TimeUnit.SECONDS);long initialDelay = new Random().nextInt(INITIAL_DELAY_IN_MINUTE) + 10;LogUtil.DEFAULT_LOG.warn("initialDelay:{}", initialDelay);//每6小时保存一次所有配置文件ConfigExecutor.scheduleConfigTask(dumpAll, initialDelay, DUMP_ALL_INTERVAL_IN_MINUTE, TimeUnit.MINUTES);//每6小时保存一次所有Beta缓存ConfigExecutor.scheduleConfigTask(dumpAllBeta, initialDelay, DUMP_ALL_INTERVAL_IN_MINUTE, TimeUnit.MINUTES);//每6小时保存一次配置标签缓存ConfigExecutor.scheduleConfigTask(dumpAllTag, initialDelay, DUMP_ALL_INTERVAL_IN_MINUTE, TimeUnit.MINUTES);}//每10分钟执行一次清除配置历史信息ConfigExecutor.scheduleConfigTask(clearConfigHistory, 10, 10, TimeUnit.MINUTES);} finally {TimerContext.end(dumpFileContext, LogUtil.DUMP_LOG);}}

拓展线程池中的scheduleWithFixedDelay

//java.util.concurrent.ScheduledExecutorService#scheduleWithFixedDelay
//利用ScheduledExecutorService实现异步定时任务public void scheduleThreadPool() throws ExecutionException, InterruptedException {log.info("Thread start:{}",Thread.currentThread().getName());ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(10);ScheduledFuture<?> schedule = scheduledExecutorService.schedule(() -> {log.info("ScheduledExecutorService schedule 10s starting...");log.info("ScheduledExecutorService schedule 10s ending...");return "ok";}, 10, TimeUnit.SECONDS);//延时10s后执行定时任务,每5秒执行一次ScheduledFuture<?> scheduledFuture = scheduledExecutorService.scheduleWithFixedDelay(() -> {log.info("ScheduledExecutorService scheduleWithFixedDelay starting...");log.info("ScheduledExecutorService scheduleWithFixedDelay ending...");}, 10, 5, TimeUnit.SECONDS);log.info("Thread end:{}",Thread.currentThread().getName());String scheduleResult = (String) schedule.get();log.info("ScheduledExecutorService schedule 10s result:{}",scheduleResult);scheduledFuture.get();}

dumpConfigInfo

//com.alibaba.nacos.config.server.service.dump.DumpService#dumpConfigInfo
private void dumpConfigInfo(DumpAllProcessor dumpAllProcessor) throws IOException {int timeStep = 6;boolean isAllDump = true;// initial dump allFileInputStream fis = null;Timestamp heartheatLastStamp = null;try {//是否快速启动 默认falseif (isQuickStart()) {File heartbeatFile = DiskUtil.heartBeatFile();if (heartbeatFile.exists()) {fis = new FileInputStream(heartbeatFile);String heartheatTempLast = IoUtils.toString(fis, Constants.ENCODE);heartheatLastStamp = Timestamp.valueOf(heartheatTempLast);//距上次全量导出时长小于6小时则不进行全量导出if (TimeUtils.getCurrentTime().getTime() - heartheatLastStamp.getTime()< timeStep * 60 * 60 * 1000) {isAllDump = false;}}}//是否全部导出 if (isAllDump) {LogUtil.DEFAULT_LOG.info("start clear all config-info.");//清除所有数据DiskUtil.clearAll();//执行导出dumpAllProcessor.process(new DumpAllTask());} else {//获取上次heartbeat持久化的时间戳Timestamp beforeTimeStamp = getBeforeStamp(heartheatLastStamp, timeStep);//构建DumpChangeProcessor实例DumpChangeProcessor dumpChangeProcessor = new DumpChangeProcessor(this, beforeTimeStamp,TimeUtils.getCurrentTime());//执行dump任务dumpChangeProcessor.process(new DumpChangeTask());//每12小时执行一次配置文件md5比较更新任务Runnable checkMd5Task = () -> {LogUtil.DEFAULT_LOG.error("start checkMd5Task");List<String> diffList = ConfigCacheService.checkMd5();for (String groupKey : diffList) {String[] dg = GroupKey.parseKey(groupKey);String dataId = dg[0];String group = dg[1];String tenant = dg[2];ConfigInfoWrapper configInfo = persistService.queryConfigInfo(dataId, group, tenant);ConfigCacheService.dumpChange(dataId, group, tenant, configInfo.getContent(),configInfo.getLastModified(), configInfo.getEncryptedDataKey());}LogUtil.DEFAULT_LOG.error("end checkMd5Task");};ConfigExecutor.scheduleConfigTask(checkMd5Task, 0, 12, TimeUnit.HOURS);}} catch (IOException e) {LogUtil.FATAL_LOG.error("dump config fail" + e.getMessage());throw e;} finally {if (null != fis) {try {fis.close();} catch (IOException e) {LogUtil.DEFAULT_LOG.warn("close file failed");}}}
}

process

//com.alibaba.nacos.config.server.service.dump.processor.DumpAllProcessor#process
public boolean process(NacosTask task) {//获取config_info表中最大的IDlong currentMaxId = persistService.findConfigMaxId();long lastMaxId = 0;while (lastMaxId < currentMaxId) {//循环查询config_info表中数据(根据ID升序)，每次查询1000条Page<ConfigInfoWrapper> page = persistService.findAllConfigInfoFragment(lastMaxId, PAGE_SIZE);if (page != null && page.getPageItems() != null && !page.getPageItems().isEmpty()) {//遍历配置信息for (ConfigInfoWrapper cf : page.getPageItems()) {long id = cf.getId();//比较配置ID并赋值lastMaxId = Math.max(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());}//保存配置文件并在缓存中更新md5值ConfigCacheService.dump(cf.getDataId(), cf.getGroup(), cf.getTenant(), cf.getContent(),cf.getLastModified(), cf.getType(), cf.getEncryptedDataKey());//获取配置内容final String content = cf.getContent();//根据配置内容生成md5final String md5 = MD5Utils.md5Hex(content, Constants.ENCODE);LogUtil.DUMP_LOG.info("[dump-all-ok] {}, {}, length={}, md5={}",GroupKey2.getKey(cf.getDataId(), cf.getGroup()), cf.getLastModified(), content.length(),md5);}DEFAULT_LOG.info("[all-dump] {} / {}", lastMaxId, currentMaxId);} else {//查询的配置文件列表为空，手动重新赋值lastMaxIdlastMaxId += PAGE_SIZE;}}return true;
}

dump

//com.alibaba.nacos.config.server.service.ConfigCacheService#dump
public static boolean dump(String dataId, String group, String tenant, String content, long lastModifiedTs,String type, String encryptedDataKey) {//生成groupKey，格式：dataId+group(+tenant)String groupKey = GroupKey2.getKey(dataId, group, tenant);//获取本地缓存值，若不存在则先手动放入本地缓存中CacheItem ci = makeSure(groupKey, encryptedDataKey, false);//设置文件类型ci.setType(type);//根据同一配置文件缓存对象获取写锁(存在同时操作的可能)final int lockResult = tryWriteLock(groupKey);assert (lockResult != 0);//获取写锁失败if (lockResult < 0) {DUMP_LOG.warn("[dump-error] write lock failed. {}", groupKey);return false;}try {//根据内容生成md5加密串final String md5 = MD5Utils.md5Hex(content, Constants.ENCODE);//比较当前文件的最后修改时间小于缓存中的值，那么说明当前文件已经被更改过，直接不用保存，返回trueif (lastModifiedTs < ConfigCacheService.getLastModifiedTs(groupKey)) {DUMP_LOG.warn("[dump-ignore] the content is old. groupKey={}, md5={}, lastModifiedOld={}, "+ "lastModifiedNew={}", groupKey, md5, ConfigCacheService.getLastModifiedTs(groupKey),lastModifiedTs);return true;}//和缓存中的md5一致 && 服务器缓存中有对应配置文件，则忽略保存缓存文件if (md5.equals(ConfigCacheService.getContentMd5(groupKey)) && DiskUtil.targetFile(dataId, group, tenant).exists()) {DUMP_LOG.warn("[dump-ignore] ignore to save cache file. groupKey={}, md5={}, lastModifiedOld={}, "+ "lastModifiedNew={}", groupKey, md5, ConfigCacheService.getLastModifiedTs(groupKey),lastModifiedTs);} else if (!PropertyUtil.isDirectRead()) {//如果使用mysql，决定是否直接读取数据，如果使用raft+derby，决定是否减少数据库读取压力，减少leader读取压力,来进行缓存文件//持久化配置信息DiskUtil.saveToDisk(dataId, group, tenant, content);}//更新md5并发布LocalDataChangeEvent事件updateMd5(groupKey, md5, lastModifiedTs, encryptedDataKey);return true;} catch (IOException ioe) {//持久化失败DUMP_LOG.error("[dump-exception] save disk error. " + groupKey + ", " + ioe);if (ioe.getMessage() != null) {String errMsg = ioe.getMessage();if (NO_SPACE_CN.equals(errMsg) || NO_SPACE_EN.equals(errMsg) || errMsg.contains(DISK_QUATA_CN) || errMsg.contains(DISK_QUATA_EN)) {// Protect from disk full.FATAL_LOG.error("磁盘满自杀退出", ioe);System.exit(0);}}return false;} finally {//释放写锁releaseWriteLock(groupKey);}
}

updateMd5

判断md5是否为空或者不同则更新本地缓存

//com.alibaba.nacos.config.server.service.ConfigCacheService#updateMd5
public static void updateMd5(String groupKey, String md5, long lastModifiedTs, String encryptedDataKey) {//获取本地缓存，md5为空或者不同则更新本地缓存CacheItem cache = makeSure(groupKey, encryptedDataKey, false);if (cache.md5 == null || !cache.md5.equals(md5)) {cache.md5 = md5;//更新修改时间cache.lastModifiedTs = lastModifiedTs;//发布LocalDataChangeEvent事件NotifyCenter.publishEvent(new LocalDataChangeEvent(groupKey));}
}

感谢阅读完本篇文章！！！
个人博客地址： huanghong.top