Spring?Cloud集成Nacos?Config動態(tài)刷新源碼剖析

更新時間：2022年08月12日 16:55:33 作者：用戶28128825

這篇文章主要為大家介紹了Spring?Cloud集成Nacos?Config動態(tài)刷新源碼剖析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

正文

從遠(yuǎn)端服務(wù)器獲取變更數(shù)據(jù)的主要模式有兩種：推（push）和拉（pull）。Push 模式簡單來說就是服務(wù)端主動將數(shù)據(jù)變更信息推送給客戶端，這種模式優(yōu)點是時效性好，服務(wù)端數(shù)據(jù)發(fā)生變更可以立馬通知到客戶端，但這種模式需要服務(wù)端維持與客戶端的心跳連接，會增加服務(wù)端實現(xiàn)的復(fù)雜度，服務(wù)端也需要占用更多的資源來維持與客戶端的連接。

而 Pull 模式則是客戶端主動去服務(wù)器請求數(shù)據(jù)，例如，每間隔10ms就向服務(wù)端發(fā)起請求獲取數(shù)據(jù)。顯而易見pull模式存在時效性問題。

請求的間隔也不太好設(shè)置，間隔太短，對服務(wù)器請求壓力過大。間隔時間過長，那么必然會造成時效性很差。而且如果配置長時間不更新，并且存在大量的客戶端就會產(chǎn)生大量無效的pull請求。

Nacos Config動態(tài)刷新機制

Nacos 沒有采用上述的兩種模式，而是采用了長輪詢方式結(jié)合了推和拉的優(yōu)點：

長輪詢也是輪詢，因此 Nacos 客戶端會默認(rèn)每10ms向服務(wù)端發(fā)起請求，當(dāng)客戶端請求服務(wù)端時會在請求頭上攜帶長輪詢的超時時間，默認(rèn)是30s。而服務(wù)端接收到該請求時會hang住請求，為了防止客戶端超時會在請求頭攜帶的超時時間上減去500ms，因此默認(rèn)會hang住請求29.5s。
在這期間如果服務(wù)端發(fā)生了配置變更會產(chǎn)生相應(yīng)的事件，監(jiān)聽到該事件后，會響應(yīng)對應(yīng)的客戶端。這樣一來客戶端不會頻繁發(fā)起輪詢請求，而服務(wù)端也不需要維持與客戶端的心跳，兼?zhèn)淞藭r效性和復(fù)雜度。

如果你覺得源碼枯燥的話，可以選擇不看后半部分的源碼，先通過這張流程圖去了解Nacos動態(tài)刷新機制的流程：

Nacos Config 長輪詢源碼剖析

首先，打開 com.alibaba.cloud.nacos.NacosConfigBootstrapConfiguration 這個類，從類名也可以看出該類是Nacos Config的啟動配置類，是Nacos Config自動裝配的入口。在該類中的 nacosConfigManager 方法實例化了一個 NacosConfigManager 對象，并注冊到容器中：

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

在 NacosConfigManager 的構(gòu)造器中調(diào)用了 createConfigService 方法，這是一個靜態(tài)方法用來創(chuàng)建 ConfigService 對象的單例。

/**
 * Compatible with old design,It will be perfected in the future.
 */
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;
}

ConfigService 的具體實現(xiàn)是 NacosConfigService，在該類的構(gòu)造器中主要初始化了 HttpAgent 和 ClientWorker 對象。

ClientWorker構(gòu)造器初始化線程池

ClientWorker 的構(gòu)造器中則初始化了幾個線程池：

public ClientWorker(final HttpAgent agent, final ConfigFilterChainManager configFilterChainManager,
        final Properties properties) {
    this.agent = agent;
    this.configFilterChainManager = configFilterChainManager;
    // Initialize the timeout parameter
    init(properties);
    // 創(chuàng)建具有定時執(zhí)行功能的單線程池，用于定時執(zhí)行 checkConfigInfo 方法
    this.executor = Executors.newScheduledThreadPool(1, new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r);
            t.setName("com.alibaba.nacos.client.Worker." + agent.getName());
            t.setDaemon(true);
            return t;
        }
    });
    // 創(chuàng)建具有定時執(zhí)行功能的且線程數(shù)與cpu核數(shù)相對應(yīng)的線程池，用于根據(jù)需要動態(tài)刷新的配置文件執(zhí)行 LongPollingRunnable，因此長輪詢?nèi)蝿?wù)是可以有多個并行的
    this.executorService = Executors
            .newScheduledThreadPool(Runtime.getRuntime().availableProcessors(), new ThreadFactory() {
                @Override
                public Thread newThread(Runnable r) {
                    Thread t = new Thread(r);
                    t.setName("com.alibaba.nacos.client.Worker.longPolling." + agent.getName());
                    t.setDaemon(true);
                    return t;
                }
            });
    // 每10ms執(zhí)行一次 checkConfigInfo 方法
    this.executor.scheduleWithFixedDelay(new Runnable() {
        @Override
        public void run() {
            try {
                checkConfigInfo();
            } catch (Throwable e) {
                LOGGER.error("[" + agent.getName() + "] [sub-check] rotate check error", e);
            }
        }
    }, 1L, 10L, TimeUnit.MILLISECONDS);
}
private void init(Properties properties) {
    // 長輪詢的超時時間，默認(rèn)為30秒，此參數(shù)會被放到請求頭中帶到服務(wù)端，服務(wù)端會根據(jù)該參數(shù)去做長輪詢的hold
    timeout = Math.max(ConvertUtils.toInt(properties.getProperty(PropertyKeyConst.CONFIG_LONG_POLL_TIMEOUT),
            Constants.CONFIG_LONG_POLL_TIMEOUT), Constants.MIN_CONFIG_LONG_POLL_TIMEOUT);
    taskPenaltyTime = ConvertUtils
            .toInt(properties.getProperty(PropertyKeyConst.CONFIG_RETRY_TIME), Constants.CONFIG_RETRY_TIME);
    this.enableRemoteSyncConfig = Boolean
            .parseBoolean(properties.getProperty(PropertyKeyConst.ENABLE_REMOTE_SYNC_CONFIG));
}
/**
 * Check config info.
 */
public void checkConfigInfo() {
    // Dispatch taskes.
    // 獲取需要監(jiān)聽的文件數(shù)量
    int listenerSize = cacheMap.size();
    // Round up the longingTaskCount.
    // 默認(rèn)一個 LongPollingRunnable 可以處理監(jiān)聽3k個配置文件的變化，超過3k個才會創(chuàng)建新的 LongPollingRunnable
    int longingTaskCount = (int) Math.ceil(listenerSize / ParamUtil.getPerTaskConfigSize());
    if (longingTaskCount > currentLongingTaskCount) {
        for (int i = (int) currentLongingTaskCount; i < longingTaskCount; i++) {
            // The task list is no order.So it maybe has issues when changing.
            executorService.execute(new LongPollingRunnable(i));
        }
        currentLongingTaskCount = longingTaskCount;
    }
}

LongPollingRunnable 類主要用于檢查本地配置，以及長輪詢地去服務(wù)端獲取變更配置的 dataid 和 group，其代碼位于 com.alibaba.nacos.client.config.impl.ClientWorker 類，代碼如下：

class LongPollingRunnable implements Runnable {
    private final int taskId;
    public LongPollingRunnable(int taskId) {
        this.taskId = taskId;
    }
    @Override
    public void run() {
        List<CacheData> cacheDatas = new ArrayList<CacheData>();
        List<String> inInitializingCacheList = new ArrayList<String>();
        try {
            // check failover config
            // 遍歷本地緩存的配置
            for (CacheData cacheData : cacheMap.values()) {
                if (cacheData.getTaskId() == taskId) {
                    cacheDatas.add(cacheData);
                    try {
                        // 檢查本地配置
                        checkLocalConfig(cacheData);
                        if (cacheData.isUseLocalConfigInfo()) {
                            cacheData.checkListenerMd5();
                        }
                    } catch (Exception e) {
                        LOGGER.error("get local config info error", e);
                    }
                }
            }
            // check server config
            // 通過長輪詢檢查服務(wù)端配置
            List<String> changedGroupKeys = checkUpdateDataIds(cacheDatas, inInitializingCacheList);
            if (!CollectionUtils.isEmpty(changedGroupKeys)) {
                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(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);
                }
            }
            for (CacheData cacheData : cacheDatas) {
                if (!cacheData.isInitializing() || inInitializingCacheList
                        .contains(GroupKey.getKeyTenant(cacheData.dataId, cacheData.group, cacheData.tenant))) {
                    cacheData.checkListenerMd5();
                    cacheData.setInitializing(false);
                }
            }
            inInitializingCacheList.clear();
            executorService.execute(this);
        } catch (Throwable e) {
            // If the rotation training task is abnormal, the next execution time of the task will be punished
            LOGGER.error("longPolling error : ", e);
            executorService.schedule(this, taskPenaltyTime, TimeUnit.MILLISECONDS);
        }
    }
}

上面有個 checkUpdateDataIds 方法，用于獲取發(fā)生變更了的配置文件的dataId列表，它同樣位于 ClientWorker 內(nèi)。

如下：

/**
 * Fetch the dataId list from server.
 *
 * @param cacheDatas              CacheDatas for config infomations.
 * @param inInitializingCacheList initial cache lists.
 * @return String include dataId and group (ps: it maybe null).
 * @throws Exception Exception.
 */
List<String> checkUpdateDataIds(List<CacheData> cacheDatas, List<String> inInitializingCacheList) throws Exception {
    // 拼接出配置文件的唯一標(biāo)識
    StringBuilder sb = new StringBuilder();
    for (CacheData cacheData : cacheDatas) {
        if (!cacheData.isUseLocalConfigInfo()) {
            sb.append(cacheData.dataId).append(WORD_SEPARATOR);
            sb.append(cacheData.group).append(WORD_SEPARATOR);
            if (StringUtils.isBlank(cacheData.tenant)) {
                sb.append(cacheData.getMd5()).append(LINE_SEPARATOR);
            } else {
                sb.append(cacheData.getMd5()).append(WORD_SEPARATOR);
                sb.append(cacheData.getTenant()).append(LINE_SEPARATOR);
            }
            if (cacheData.isInitializing()) {
                // It updates when cacheData occours in cacheMap by first time.
                inInitializingCacheList
                        .add(GroupKey.getKeyTenant(cacheData.dataId, cacheData.group, cacheData.tenant));
            }
        }
    }
    boolean isInitializingCacheList = !inInitializingCacheList.isEmpty();
    return checkUpdateConfigStr(sb.toString(), isInitializingCacheList);
}
/**
 * Fetch the updated dataId list from server.
 *
 * @param probeUpdateString       updated attribute string value.
 * @param isInitializingCacheList initial cache lists.
 * @return The updated dataId list(ps: it maybe null).
 * @throws IOException Exception.
 */
List<String> checkUpdateConfigStr(String probeUpdateString, boolean isInitializingCacheList) throws Exception {
    Map<String, String> params = new HashMap<String, String>(2);
    params.put(Constants.PROBE_MODIFY_REQUEST, probeUpdateString);
    Map<String, String> headers = new HashMap<String, String>(2);
    // 長輪詢的超時時間
    headers.put("Long-Pulling-Timeout", "" + timeout);
    // told server do not hang me up if new initializing cacheData added in
    if (isInitializingCacheList) {
        headers.put("Long-Pulling-Timeout-No-Hangup", "true");
    }
    if (StringUtils.isBlank(probeUpdateString)) {
        return Collections.emptyList();
    }
    try {
        // In order to prevent the server from handling the delay of the client's long task,
        // increase the client's read timeout to avoid this problem.
        long readTimeoutMs = timeout + (long) Math.round(timeout >> 1);
        // 向服務(wù)端發(fā)起一個http請求，該請求在服務(wù)端配置沒有變更的情況下默認(rèn)會hang住30s
        HttpRestResult<String> result = agent
                .httpPost(Constants.CONFIG_CONTROLLER_PATH + "/listener", headers, params, agent.getEncode(),
                        readTimeoutMs);
        if (result.ok()) {
            setHealthServer(true);
            // 響應(yīng)狀態(tài)是成功則解析響應(yīng)體得到 dataId、group、tenant 等信息并返回
            return parseUpdateDataIdResponse(result.getData());
        } else {
            setHealthServer(false);
            LOGGER.error("[{}] [check-update] get changed dataId error, code: {}", agent.getName(),
                    result.getCode());
        }
    } catch (Exception e) {
        setHealthServer(false);
        LOGGER.error("[" + agent.getName() + "] [check-update] get changed dataId exception", e);
        throw e;
    }
    return Collections.emptyList();
}

客戶端對 listener 接口的請求會進(jìn)入到服務(wù)端的com.alibaba.nacos.config.server.controller.ConfigController#listener 方法進(jìn)行處理，該方法主要是調(diào)用了 com.alibaba.nacos.config.server.controller.ConfigServletInner#doPollingConfig 方法。

代碼如下：

/**
 * 輪詢接口
 */
public String doPollingConfig(HttpServletRequest request, HttpServletResponse response,
                              Map<String, String> clientMd5Map, int probeRequestSize)
    throws IOException, ServletException {
    // 如果支持長輪詢則進(jìn)入長輪詢的流程
    if (LongPollingService.isSupportLongPolling(request)) {
        longPollingService.addLongPollingClient(request, response, clientMd5Map, probeRequestSize);
        return HttpServletResponse.SC_OK + "";
    }
    // else 兼容短輪詢邏輯
    List<String> changedGroups = MD5Util.compareMd5(request, response, clientMd5Map);
    // 兼容短輪詢result
    String oldResult = MD5Util.compareMd5OldResult(changedGroups);
    String newResult = MD5Util.compareMd5ResultString(changedGroups);
    String version = request.getHeader(Constants.CLIENT_VERSION_HEADER);
    if (version == null) {
        version = "2.0.0";
    }
    int versionNum = Protocol.getVersionNumber(version);
    /**
     * 2.0.4版本以前, 返回值放入header中
     */
    if (versionNum < START_LONGPOLLING_VERSION_NUM) {
        response.addHeader(Constants.PROBE_MODIFY_RESPONSE, oldResult);
        response.addHeader(Constants.PROBE_MODIFY_RESPONSE_NEW, newResult);
    } else {
        request.setAttribute("content", newResult);
    }
    // 禁用緩存
    response.setHeader("Pragma", "no-cache");
    response.setDateHeader("Expires", 0);
    response.setHeader("Cache-Control", "no-cache,no-store");
    response.setStatus(HttpServletResponse.SC_OK);
    return HttpServletResponse.SC_OK + "";
}

我們主要關(guān)注上面的 com.alibaba.nacos.config.server.service.LongPollingService#addLongPollingClient 長輪詢流程的方法。

長輪詢流程方法

代碼如下：

public void addLongPollingClient(HttpServletRequest req, HttpServletResponse rsp, Map<String, String> clientMd5Map,
                                 int probeRequestSize) {
    String str = req.getHeader(LongPollingService.LONG_POLLING_HEADER);
    String noHangUpFlag = req.getHeader(LongPollingService.LONG_POLLING_NO_HANG_UP_HEADER);
    String appName = req.getHeader(RequestUtil.CLIENT_APPNAME_HEADER);
    String tag = req.getHeader("Vipserver-Tag");
    int delayTime = SwitchService.getSwitchInteger(SwitchService.FIXED_DELAY_TIME, 500);
    /**
     * 提前500ms返回響應(yīng)，為避免客戶端超時 @qiaoyi.dingqy 2013.10.22改動  add delay time for LoadBalance
     */
    long timeout = Math.max(10000, Long.parseLong(str) - delayTime);
    if (isFixedPolling()) {
        timeout = Math.max(10000, getFixedPollingInterval());
        // do nothing but set fix polling timeout
    } else {
        long start = System.currentTimeMillis();
        List<String> changedGroups = MD5Util.compareMd5(req, rsp, clientMd5Map);
        if (changedGroups.size() > 0) {
            generateResponse(req, rsp, changedGroups);
            LogUtil.clientLog.info("{}|{}|{}|{}|{}|{}|{}",
                System.currentTimeMillis() - start, "instant", RequestUtil.getRemoteIp(req), "polling",
                clientMd5Map.size(), probeRequestSize, changedGroups.size());
            return;
        } else if (noHangUpFlag != null && noHangUpFlag.equalsIgnoreCase(TRUE_STR)) {
            LogUtil.clientLog.info("{}|{}|{}|{}|{}|{}|{}", System.currentTimeMillis() - start, "nohangup",
                RequestUtil.getRemoteIp(req), "polling", clientMd5Map.size(), probeRequestSize,
                changedGroups.size());
            return;
        }
    }
    String ip = RequestUtil.getRemoteIp(req);
    // 一定要由HTTP線程調(diào)用，否則離開后容器會立即發(fā)送響應(yīng)
    final AsyncContext asyncContext = req.startAsync();
    // AsyncContext.setTimeout()的超時時間不準(zhǔn)，所以只能自己控制
    asyncContext.setTimeout(0L);
	// 在 ClientLongPolling 的 run 方法會將 ClientLongPolling 實例（攜帶了本次請求的相關(guān)信息）放入 allSubs 中，然后會在29.5s后再執(zhí)行另一個 Runnable，該 Runnable 用于等待29.5s后依舊沒有相應(yīng)的配置變更時對客戶端進(jìn)行響應(yīng)，并將相應(yīng)的 ClientLongPolling 實例從 allSubs 中移出
    scheduler.execute(
        new ClientLongPolling(asyncContext, clientMd5Map, ip, probeRequestSize, timeout, appName, tag));
}

而 LongPollingService 實現(xiàn)了 AbstractEventListener，也就是說能接收事件通知，在其 com.alibaba.nacos.config.server.service.LongPollingService#onEvent 方法中可以看到，它關(guān)注的是 LocalDataChangeEvent 事件：

@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));
        }
    }
}

在nacos上修改配置后就會產(chǎn)生 LocalDataChangeEvent 事件，此時 LongPollingService 也就能監(jiān)聽到，當(dāng)收到該事件時就會遍歷 allSubs，找到匹配的請求并將 groupKey 返回給客戶端。

具體代碼在 DataChangeTask 中：

class DataChangeTask implements Runnable {
    @Override
    public void run() {
        try {
            ConfigService.getContentBetaMd5(groupKey);
            for (Iterator<ClientLongPolling> iter = allSubs.iterator(); iter.hasNext(); ) {
                ClientLongPolling clientSub = iter.next();
                if (clientSub.clientMd5Map.containsKey(groupKey)) {
                    // 如果beta發(fā)布且不在beta列表直接跳過
                    if (isBeta && !betaIps.contains(clientSub.ip)) {
                        continue;
                    }
                    // 如果tag發(fā)布且不在tag列表直接跳過
                    if (StringUtils.isNotBlank(tag) && !tag.equals(clientSub.tag)) {
                        continue;
                    }
                    getRetainIps().put(clientSub.ip, System.currentTimeMillis());
                    iter.remove(); // 刪除訂閱關(guān)系
                    LogUtil.clientLog.info("{}|{}|{}|{}|{}|{}|{}",
                        (System.currentTimeMillis() - changeTime),
                        "in-advance",
                        RequestUtil.getRemoteIp((HttpServletRequest)clientSub.asyncContext.getRequest()),
                        "polling",
                        clientSub.clientMd5Map.size(), clientSub.probeRequestSize, groupKey);
                    clientSub.sendResponse(Arrays.asList(groupKey));
                }
            }
        } catch (Throwable t) {
            LogUtil.defaultLog.error("data change error:" + t.getMessage(), t.getCause());
        }
    }
    DataChangeTask(String groupKey) {
        this(groupKey, false, null);
    }
    DataChangeTask(String groupKey, boolean isBeta, List<String> betaIps) {
        this(groupKey, isBeta, betaIps, null);
    }
    DataChangeTask(String groupKey, boolean isBeta, List<String> betaIps, String tag) {
        this.groupKey = groupKey;
        this.isBeta = isBeta;
        this.betaIps = betaIps;
        this.tag = tag;
    }
    final String groupKey;
    final long changeTime = System.currentTimeMillis();
    final boolean isBeta;
    final List<String> betaIps;
    final String tag;
}

當(dāng)客戶端收到變更的dataid+group后，就會去服務(wù)端獲取最新的配置數(shù)據(jù)，并更新本地數(shù)據(jù) cacheData，然后發(fā)送數(shù)據(jù)變更事件，整個流程結(jié)束。

獲取服務(wù)端最新配置數(shù)據(jù)的方法：com.alibaba.nacos.client.config.impl.ClientWorker#getServerConfig
發(fā)送數(shù)據(jù)變更事件的方法：com.alibaba.nacos.client.config.impl.CacheData#checkListenerMd5

最后附上一張流程與源碼的對應(yīng)圖：