當前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

nacos配置刷新失败导致的cpu上升和频繁重启，nacos配置中心源码解析

發(fā)布時間：2023/12/10 编程问答 32 豆豆

生活随笔收集整理的這篇文章主要介紹了 nacos配置刷新失败导致的cpu上升和频繁重启，nacos配置中心源码解析小編覺得挺不錯的,現(xiàn)在分享給大家,幫大家做個參考.

大家好，我是烤鴨：
nacos 版本 1.3.2，先說下結(jié)論，頻繁重啟的原因確實沒有找到，跟nacos有關，日志沒有保留多少，只能從源碼找下頭緒(出問題的版本 server用的是 nacos 1.1，nacos-client 1.0)

nacos 拉取配置原理

有兩個核心的類 ClientWorker 和 ServerHttpAgent，先從頭捋一下。

NacosConfigBootstrapConfiguration 初始化 NacosConfigManager

@Bean @ConditionalOnMissingBean public NacosConfigManager nacosConfigManager(NacosConfigProperties nacosConfigProperties) {return new NacosConfigManager(nacosConfigProperties); }

NacosConfigManager

static ConfigService createConfigService(NacosConfigProperties nacosConfigProperties) {if (Objects.isNull(service)) {synchronized (NacosConfigManager.class) {try {if (Objects.isNull(service)) {service = NacosFactory.createConfigService(nacosConfigProperties.assembleConfigServiceProperties());}}catch (NacosException e) {log.error(e.getMessage());throw new NacosConnectionFailureException(nacosConfigProperties.getServerAddr(), e.getMessage(), e);}}}return service; }

NacosFactory -> ConfigFactory 初始化 NacosConfigService

public static ConfigService createConfigService(Properties properties) throws NacosException {try {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);} }

executor 每10ms 檢查配置信息, executorService 執(zhí)行 LongPollingRunnable 的線程池(單線程)

public ClientWorker(final HttpAgent agent, final ConfigFilterChainManager configFilterChainManager, final Properties properties) {this.agent = agent;this.configFilterChainManager = configFilterChainManager;// 設置超時時間(默認30s)init(properties);//...// 每10ms 檢查配置信息executor.scheduleWithFixedDelay(new Runnable() {@Overridepublic void run() {try {checkConfigInfo();} catch (Throwable e) {LOGGER.error("[" + agent.getName() + "] [sub-check] rotate check error", e);}}}, 1L, 10L, TimeUnit.MILLISECONDS); }

ClientWorker.checkConfigInfo
拉取任務的執(zhí)行條件，cacheMap 是維護在內(nèi)存里的任務數(shù)據(jù)(Map的key是groupid+dataid 分別 urlencode、value是CacheData)

public void checkConfigInfo() {// 獲取cache的數(shù)據(jù)數(shù)量int listenerSize = cacheMap.get().size();// 向上取整int longingTaskCount = (int) Math.ceil(listenerSize / ParamUtil.getPerTaskConfigSize());if (longingTaskCount > currentLongingTaskCount) {for (int i = (int) currentLongingTaskCount; i < longingTaskCount; i++) {// 要判斷任務是否在執(zhí)行這塊需要好好想想。任務列表現(xiàn)在是無序的。變化過程可能有問題executorService.execute(new LongPollingRunnable(i));}currentLongingTaskCount = longingTaskCount;}}

ClientWorker$LongPollingRunnable

class LongPollingRunnable implements Runnable {private int taskId;public LongPollingRunnable(int taskId) {this.taskId = taskId;}@Overridepublic void run() {List<CacheData> cacheDatas = new ArrayList<CacheData>();List<String> inInitializingCacheList = new ArrayList<String>();try {// 校驗本地數(shù)據(jù)for (CacheData cacheData : cacheMap.get().values()) {if (cacheData.getTaskId() == taskId) {cacheDatas.add(cacheData);try {// 看下面具體方法checkLocalConfig(cacheData);if (cacheData.isUseLocalConfigInfo()) {// 是否有屬性變更(變更會通過notify改變監(jiān)聽器的md5值),變更的話更新cacheDatacacheData.checkListenerMd5();}} catch (Exception e) {LOGGER.error("get local config info error", e);}}}// 獲取變更屬性,更多看下面List<String> changedGroupKeys = checkUpdateDataIds(cacheDatas, inInitializingCacheList);LOGGER.info("get changedGroupKeys:" + changedGroupKeys);// 解析變更屬性for (String groupKey : changedGroupKeys) {String[] key = GroupKey.parseKey(groupKey);String dataId = key[0];String group = key[1];String tenant = null;if (key.length == 3) {tenant = key[2];}try {String[] ct = getServerConfig(dataId, group, tenant, 3000L);CacheData cache = cacheMap.get().get(GroupKey.getKeyTenant(dataId, group, tenant));cache.setContent(ct[0]);if (null != ct[1]) {cache.setType(ct[1]);}LOGGER.info("[{}] [data-received] dataId={}, group={}, tenant={}, md5={}, content={}, type={}", agent.getName(), dataId, group, tenant, cache.getMd5(), ContentUtils.truncateContent(ct[0]), ct[1]);} catch (NacosException ioe) {String message = String.format( "[%s] [get-update] get changed config exception. dataId=%s, group=%s, tenant=%s", agent.getName(), dataId, group, tenant);LOGGER.error(message, ioe);}}// 更新緩存cacheDatafor (CacheData cacheData : cacheDatas) {if (!cacheData.isInitializing() || inInitializingCacheList.contains(GroupKey.getKeyTenant(cacheData.dataId, cacheData.group, cacheData.tenant))) {cacheData.checkListenerMd5();cacheData.setInitializing(false);}}inInitializingCacheList.clear();// 由于從server端拉取配置是 30s一次(同步操作), 這里可以理解為每30s執(zhí)行的定時線程executorService.execute(this);} catch (Throwable e) {// If the rotation training task is abnormal, the next execution time of the task will be punishedLOGGER.error("longPolling error : ", e);executorService.schedule(this, taskPenaltyTime, TimeUnit.MILLISECONDS);}}}

ClientWorker.checkLocalConfig

private void checkLocalConfig(CacheData cacheData) {final String dataId = cacheData.dataId;final String group = cacheData.group;final String tenant = cacheData.tenant;File path = LocalConfigInfoProcessor.getFailoverFile(agent.getName(), dataId, group, tenant);// 開啟本地配置并且本地文件存在,以本地文件為準,更新緩存if (!cacheData.isUseLocalConfigInfo() && path.exists()) {String content = LocalConfigInfoProcessor.getFailover(agent.getName(), dataId, group, tenant);final String md5 = MD5Utils.md5Hex(content, Constants.ENCODE);cacheData.setUseLocalConfigInfo(true);cacheData.setLocalConfigInfoVersion(path.lastModified());cacheData.setContent(content);LOGGER.warn("[{}] [failover-change] failover file created. dataId={}, group={}, tenant={}, md5={}, content={}", agent.getName(), dataId, group, tenant, md5, ContentUtils.truncateContent(content));return;}// 開啟本地配置并且本地文件不存在,以服務器拉取為準,同時關閉本地配置if (cacheData.isUseLocalConfigInfo() && !path.exists()) {cacheData.setUseLocalConfigInfo(false);LOGGER.warn("[{}] [failover-change] failover file deleted. dataId={}, group={}, tenant={}", agent.getName(), dataId, group, tenant);return;}// 開啟本地配置并且本地文件存在并且緩存中版本和本地文件版本不一樣,以本地文件為準,更新緩存if (cacheData.isUseLocalConfigInfo() && path.exists() && cacheData.getLocalConfigInfoVersion() != path.lastModified()) {String content = LocalConfigInfoProcessor.getFailover(agent.getName(), dataId, group, tenant);final String md5 = MD5Utils.md5Hex(content, Constants.ENCODE);cacheData.setUseLocalConfigInfo(true);cacheData.setLocalConfigInfoVersion(path.lastModified());cacheData.setContent(content);LOGGER.warn("[{}] [failover-change] failover file changed. dataId={}, group={}, tenant={}, md5={}, content={}", agent.getName(), dataId, group, tenant, md5, ContentUtils.truncateContent(content));}}

ClientWorker.checkUpdateDataIds —> ClientWorker.checkUpdateConfigStr —> (ServerHttpAgent)HttpAgent.httpPost
組裝請求頭和參數(shù)，重試，異常處理的代碼都刪掉了，這里只保留了這塊，這就是為什么每次請求server端會本地阻塞30s。

public HttpRestResult<String> httpPost(String path, Map<String, String> headers, Map<String, String> paramValues,String encode, long readTimeoutMs) throws Exception {// ...do {// ...HttpRestResult<String> result = NACOS_RESTTEMPLATE.postForm(getUrl(currentServerAddr, path), httpConfig, newHeaders,new HashMap<String, String>(0), paramValues, String.class);// ...} while (System.currentTimeMillis() <= endTime);}

CacheData.checkListenerMd5 ，判斷內(nèi)存中的和server端的md5 值

void checkListenerMd5() {for (ManagerListenerWrap wrap : listeners) {if (!md5.equals(wrap.lastCallMd5)) {safeNotifyListener(dataId, group, content, type, md5, wrap);}}}

CacheData.safeNotifyListener ，可以看到 listener.receiveConfigChange 是執(zhí)行具體的刷新操作(熱加載)，最后會更新內(nèi)存的 lastCallMd5 值

Runnable job = new Runnable() {@Overridepublic void run() {//...// 執(zhí)行回調(diào)之前先將線程classloader設置為具體webapp的classloader，以免回調(diào)方法中調(diào)用spi接口是出現(xiàn)異?；蝈e用（多應用部署才會有該問題）。Thread.currentThread().setContextClassLoader(appClassLoader);// ...// compare lastContent and contentif (listener instanceof AbstractConfigChangeListener) {Map data = ConfigChangeHandler.getInstance().parseChangeData(listenerWrap.lastContent, content, type);ConfigChangeEvent event = new ConfigChangeEvent(data);// 執(zhí)行屬性變更操作((AbstractConfigChangeListener) listener).receiveConfigChange(event);listenerWrap.lastContent = content;}listenerWrap.lastCallMd5 = md5;LOGGER.info("[{}] [notify-ok] dataId={}, group={}, md5={}, listener={} ", name, dataId, group, md5,listener);// ...}};

問題排查

先看下當時的日志：
能看出來服務在不停的刷新應用(熱重啟，間隔時間30s)，不停熱重啟引起cpu升高（又是另一個問題了）

再看一下機器的cpu，3臺不同的機器都幾乎是同時出現(xiàn)了這個問題，cpu在24h之內(nèi)緩慢上升

日志內(nèi)容很少，結(jié)合剛才看的源碼，獲取服務端配置確實是30s一次，也就是找到觸發(fā)點就知道了。

假設猜想

1.0 的代碼和 1.3.2 還是有點區(qū)別，這是 1.0 的 CacheData.safeNotifyListener

private void safeNotifyListener(final String dataId, final String group, final String content,final String md5, final ManagerListenerWrap listenerWrap) {final Listener listener = listenerWrap.listener;Runnable job = new Runnable() {public void run() {// ...Thread.currentThread().setContextClassLoader(appClassLoader);ConfigResponse cr = new ConfigResponse();cr.setDataId(dataId);cr.setGroup(group);cr.setContent(content);configFilterChainManager.doFilter(null, cr);String contentTmp = cr.getContent();listener.receiveConfigInfo(contentTmp);listenerWrap.lastCallMd5 = md5;LOGGER.info("[{}] [notify-ok] dataId={}, group={}, md5={}, listener={} ", name, dataId, group, md5,listener);}};

結(jié)合日志猜測是 listenerWrap.lastCallMd5 = md5; 沒有執(zhí)行，也就是執(zhí)行 listener.receiveConfigInfo(contentTmp); 報錯了。

listener.receiveConfigInfo 都干啥了

簡單來說就是調(diào)用了 applicationContext.publishEvent，傳入的是刷新事件 RefreshEvent

RefreshEventListener 會接收刷新事件執(zhí)行刷新操作，繼續(xù)發(fā)布事件 RefreshScopeRefreshedEvent

對 @RefreshScope 的 Bean 完成熱加載。

看一段nacos正常刷新屬性（相同的項目）的日志。

后續(xù)是對注冊中心再次注冊和發(fā)現(xiàn)服務的過程 (DiscoveryClient)，目前來看是這個地方報錯了，但是具體原因還是不知道。

總結(jié)

雖然是nacos 引起的，但是問題原因并沒有找到。

注冊中心使用的是eurka。

服務是新服務，暫時沒流量，重啟之后也沒再發(fā)現(xiàn)類似的問題，等下次出現(xiàn)了一定要保留日志…

總結(jié)

以上是生活随笔為你收集整理的nacos配置刷新失败导致的cpu上升和频繁重启，nacos配置中心源码解析的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。