微服務架構 | *2.5 Nacos 長輪詢定時機制的源碼分析

前言
1. 用戶端的長輪詢定時機制
- 1.1 利用反射機制執行個體化 NacosConfigService 對象
- 1.2 NacosConfigService 的構造方法裡啟動長輪詢定時任務
  - 1.2.1 初始化 HttpAgent
  - 1.2.2 初始化 ClientWorker
- 1.3 檢查配置變更，讀取變更配置 LongPollingRunnable.run()
  - 1.3.1 檢查配置變更 ClientWorker.checkUpdateDataIds()
  - 1.3.2 讀取變更配置 ClientWorker.getServerConfig()
2. 服務端的長輪詢定時機制
- 2.1 伺服器接收請求 ConfigController.listener()
- 2.2 執行長輪詢請求 ConfigServletInner.doPollingConfig()
- 2.3 建立線程執行定時任務 ClientLongPolling.run()
- 2.4 監聽配置變更事件
  - 2.4.1 監聽 LocalDataChangeEvent 事件的實作
  - 2.4.2 監聽事件後的處理邏輯 DataChangeTask.run()
3. 源碼結構圖小結
- 3.1 用戶端的長輪詢定時機制
- 3.2 服務端的長輪詢定時機制
- 3.3 Nacos 伺服器配置變更的事件監聽
最後

參考資料：

《Spring Microservices in Action》

《Spring Cloud Alibaba 微服務原理與實戰》

《B站尚矽谷 SpringCloud 架構開發教程周陽》

為友善了解與表達，這裡把 Nacos 控制台和 Nacos 注冊中心稱為 Nacos 伺服器（就是 web 界面那個），我們編寫的業務服務稱為 Nacso 用戶端；

由于篇幅有限，這裡将源碼分析分為上下兩篇，其中上篇講擷取配置與事件訂閱機制，下篇講長輪詢定時機制；在《微服務架構 | 2.2 Alibaba Nacos 的統一配置管理》中提到一張 Nacos 動态監聽的長輪詢機制原理圖，本篇将圍繞這張圖剖析長輪詢定時機制的原理；

上篇《微服務架構 | *2.4 Nacos 配置中心的源碼分析（擷取配置與事件訂閱機制）》中的 1.1 提到，ConfigService 是 Nacos 用戶端提供的用于通路實作配置中心基本操作的類，我們将從 ConfigService 的執行個體化開始長輪詢定時機制的源碼之旅；

我們從上一篇文章的這裡開始【斷點步入】；
NacosPropertySourceLocator.locate() ；

微服務架構 | *2.5 Nacos 長輪詢定時機制的源碼分析

用戶端的長輪詢定時任務是在 NacosFactory.createConfigService() 方法中，建構 ConfigService 對象執行個體時啟動的，我們接着 1.1 處的源碼；
進入 NacosFactory.createConfigService() ：

public static ConfigService createConfigService(Properties properties) throws NacosException {
    //【斷點步入】建立 ConfigService
    return ConfigFactory.createConfigService(properties);
}

ConfigFactory.createConfigService() ，發現其使用反射機制執行個體化 NacosConfigService 對象；

public static ConfigService createConfigService(Properties properties) throws NacosException {
    try {
        //通過 Class.forName 來加載 NacosConfigService
        Class<?> driverImplClass = Class.forName("com.alibaba.nacos.client.config.NacosConfigService");
        //根據對象的屬性建立對象構造器（反射）
        Constructor constructor = driverImplClass.getConstructor(Properties.class);
        //【斷點步入 1.2】通過構造器執行個體化對象（反射）
        ConfigService vendorImpl = (ConfigService)constructor.newInstance(properties);
        return vendorImpl;
    } catch (Throwable var4) {
        throw new NacosException(-400, var4);
    }
}

NacosConfigService.NacosConfigService() 構造方法，裡面設定了一些更遠端任務相關的屬性；

public NacosConfigService(Properties properties) throws NacosException {
    String encodeTmp = properties.getProperty("encode");
    if (StringUtils.isBlank(encodeTmp)) {
        this.encode = "UTF-8";
    } else {
        this.encode = encodeTmp.trim();
    }
    //初始化命名空間
    this.initNamespace(properties);
    //【斷點步入 1.2.1】初始化 HttpAgent，用到了裝飾器模式，實際工作的類是 ServerHttpAgent
    this.agent = new MetricsHttpAgent(new ServerHttpAgent(properties));
    this.agent.start();
    //【斷點步入 1.2.2】ClientWorker 是用戶端的一個工作類，agent 作為參數傳入 ClientWorker
    this.worker = new ClientWorker(this.agent, this.configFilterChainManager, properties);
}

MetricsHttpAgent 類的設計如下：

微服務架構 | *2.5 Nacos 長輪詢定時機制的源碼分析
ServerHttpAgent 類的設計如下：

微服務架構 | *2.5 Nacos 長輪詢定時機制的源碼分析

ClientWorker.ClientWorker() 構造方法，主要是建立了兩個定時排程的線程池，并啟動一個定時任務；

public ClientWorker(final HttpAgent agent, ConfigFilterChainManager configFilterChainManager, Properties properties) {
    this.agent = agent;
    this.configFilterChainManager = configFilterChainManager;
    this.init(properties);
    //建立 executor 線程池，隻擁有一個核心線程，每隔 10ms 就會執行一次 checkConfiglnfo() 方法，檢查配置資訊
    this.executor = Executors.newScheduledThreadPool(1, new ThreadFactory() {
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r);
            t.setName("com.alibaba.nacos.client.Worker." + agent.getName());
            t.setDaemon(true);
            return t;
        }
    });
    //建立 executorService 線程池，隻完成了初始化，後續會用到，主要用于實作用戶端的定時長輪詢功能
    this.executorService = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors(), new ThreadFactory() {
        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;
        }
    });
    //使用 executor 啟動一個每隔 10s 執行一次的定時任務
    this.executor.scheduleWithFixedDelay(new Runnable() {
        public void run() {
            try {
                //【斷點步入】檢查配置是否發生變化
                ClientWorker.this.checkConfigInfo();
            } catch (Throwable var2) {
                ClientWorker.LOGGER.error("[" + agent.getName() + "] [sub-check] rotate check error", var2);
            }
        }
    }, 1L, 10L, TimeUnit.MILLISECONDS);
}

ClientWorker.checkConfigInfo() ，每隔 10s 檢查一次配置是否發生變化；
- cacheMap：是一個 AtomicReference<Map<String, CacheData>> 對象，用來存儲監聽變更的緩存集合，key 是根據 datalD/group/tenant（租戶）拼接的值。Value 是對應的存儲在 Nacos 伺服器上的配置檔案的内容；
- 長輪詢任務拆分：預設情況下，每個長輪詢 LongPollingRunnable 任務處理3000個監聽配置集。如果超過3000個，則需要啟動多個 LongPollingRunnable 去執行；

public void checkConfigInfo() {
    //分任務
    int listenerSize = ((Map)this.cacheMap.get()).size();
    //向上取整為批數
    int longingTaskCount = (int)Math.ceil((double)listenerSize / ParamUtil.getPerTaskConfigSize());
    //如果監聽配置集超過 3000，就建立多個 LongPollingRunnable 線程
    if ((double)longingTaskCount > this.currentLongingTaskCount) {
        for(int i = (int)this.currentLongingTaskCount; i < longingTaskCount; ++i) {
            //【點進去】LongPollingRunnable 實際上是一個線程
            this.executorService.execute(new ClientWorker.LongPollingRunnable(i));
        }
        this.currentLongingTaskCount = (double)longingTaskCount;
    }
}

因為我們沒有這麼多配置項，debug 不進去，是以直接找到 LongPollingRunnable.run() 方法，該方法的主要邏輯是：
- 根據 taskld 對 cacheMap 進行資料分割；
- 再通過 checkLocalConfig() 方法比較本地配置檔案（在 ${user}\nacos\config\ 裡）的資料是否存在變更，如果有變更則直接觸發通知；

public void run() {
    List<CacheData> cacheDatas = new ArrayList();
    ArrayList inInitializingCacheList = new ArrayList();
    try {
        //周遊 CacheData，檢查本地配置
        Iterator var3 = ((Map)ClientWorker.this.cacheMap.get()).values().iterator();
        while(var3.hasNext()) {
            CacheData cacheData = (CacheData)var3.next();
            if (cacheData.getTaskId() == this.taskId) {
                cacheDatas.add(cacheData);
                try {
                    //檢查本地配置
                    ClientWorker.this.checkLocalConfig(cacheData);
                    if (cacheData.isUseLocalConfigInfo()) {
                        cacheData.checkListenerMd5();
                    }
                } catch (Exception var13) {
                    ClientWorker.LOGGER.error("get local config info error", var13);
                }
            }
        }
        //【斷點步入 1.3.1】通過長輪詢請求檢查服務端對應的配置是否發生變更
        List<String> changedGroupKeys = ClientWorker.this.checkUpdateDataIds(cacheDatas, inInitializingCacheList);
        //周遊存在變更的 groupKey，重新加載最新資料
        Iterator var16 = changedGroupKeys.iterator();
        while(var16.hasNext()) {
            String groupKey = (String)var16.next();
            String[] key = GroupKey.parseKey(groupKey);
            String dataId = key[0];
            String group = key[1];
            String tenant = null;
            if (key.length == 3) {
                tenant = key[2];
            }
            try {
                //【斷點步入 1.3.2】讀取變更配置，這裡的 dataId、group 和 tenant 是【1.3.1】裡擷取的
                String content = ClientWorker.this.getServerConfig(dataId, group, tenant, 3000L);
                CacheData cache = (CacheData)((Map)ClientWorker.this.cacheMap.get()).get(GroupKey.getKeyTenant(dataId, group, tenant));
                cache.setContent(content);
                ClientWorker.LOGGER.info("[{}] [data-received] dataId={}, group={}, tenant={}, md5={}, content={}", new Object[]{ClientWorker.this.agent.getName(), dataId, group, tenant, cache.getMd5(), ContentUtils.truncateContent(content)});
            } catch (NacosException var12) {
                String message = String.format("[%s] [get-update] get changed config exception. dataId=%s, group=%s, tenant=%s", ClientWorker.this.agent.getName(), dataId, group, tenant);
                ClientWorker.LOGGER.error(message, var12);
            }
        }
        //觸發事件通知
        var16 = cacheDatas.iterator();
        while(true) {
            CacheData cacheDatax;
            do {
                if (!var16.hasNext()) {
                    inInitializingCacheList.clear();
                    //繼續定時執行目前線程
                    ClientWorker.this.executorService.execute(this);
                    return;
                }
                cacheDatax = (CacheData)var16.next();
            } while(cacheDatax.isInitializing() && !inInitializingCacheList.contains(GroupKey.getKeyTenant(cacheDatax.dataId, cacheDatax.group, cacheDatax.tenant)));
            cacheDatax.checkListenerMd5();
            cacheDatax.setInitializing(false);
        }
    } catch (Throwable var14) {
        ClientWorker.LOGGER.error("longPolling error : ", var14);
        ClientWorker.this.executorService.schedule(this, (long)ClientWorker.this.taskPenaltyTime, TimeUnit.MILLISECONDS);
    }
}

注意：這裡的斷點需要在 Nacos 伺服器上修改配置（間隔大于 30s），進入後才好了解；

我們點進 ClientWorker.checkUpdateDataIds() 方法，發現其最終調用的是 ClientWorker.checkUpdateConfigStr() 方法，其實作邏輯與源碼如下：
- 通過 MetricsHttpAgent.httpPost() 方法（上面 1.2.1 有提到）調用 /v1/cs/configs/listener 接口實作長輪詢請求；
- 長輪詢請求在實作層面隻是設定了一個比較長的逾時時間，預設是 30s；
- 如果服務端的資料發生了變更，用戶端會收到一個 HttpResult ，服務端傳回的是存在資料變更的 Data ID、Group、Tenant；
- 獲得這些資訊之後，在 LongPollingRunnable.run() 方法中調用 getServerConfig() 去 Nacos 伺服器上讀取具體的配置内容；

List<String> checkUpdateConfigStr(String probeUpdateString, boolean isInitializingCacheList) throws IOException {
    List<String> params = Arrays.asList("Listening-Configs", probeUpdateString);
    List<String> headers = new ArrayList(2);
    headers.add("Long-Pulling-Timeout");
    headers.add("" + this.timeout);
    if (isInitializingCacheList) {
        headers.add("Long-Pulling-Timeout-No-Hangup");
        headers.add("true");
    }
    if (StringUtils.isBlank(probeUpdateString)) {
        return Collections.emptyList();
    } else {
        try {
            //調用 /v1/cs/configs/listener 接口實作長輪詢請求，傳回的 HttpResult 裡包含存在資料變更的 Data ID、Group、Tenant
            HttpResult result = this.agent.httpPost("/v1/cs/configs/listener", headers, params, this.agent.getEncode(), this.timeout);
            if (200 == result.code) {
                this.setHealthServer(true);
                //
                return this.parseUpdateDataIdResponse(result.content);
            }
            this.setHealthServer(false);
            LOGGER.error("[{}] [check-update] get changed dataId error, code: {}", this.agent.getName(), result.code);
        } catch (IOException var6) {
            this.setHealthServer(false);
            LOGGER.error("[" + this.agent.getName() + "] [check-update] get changed dataId exception", var6);
            throw var6;
        }
        return Collections.emptyList();
    }
}

ClientWorker.getServerConfig() 方法；
讀取伺服器上的變更配置；
最終調用的是 MetricsHttpAgent.httpGet() 方法（上面 1.2.1 有提到），調用 /v1/cs/configs 接口擷取配置；
然後通過調用 LocalConfigInfoProcessor.saveSnapshot() 将變更的配置儲存到本地；

public String getServerConfig(String dataId, String group, String tenant, long readTimeout) throws NacosException {
    if (StringUtils.isBlank(group)) {
        group = "DEFAULT_GROUP";
    }
    HttpResult result = null;
    try {
        List<String> params = null;
        if (StringUtils.isBlank(tenant)) {
            params = Arrays.asList("dataId", dataId, "group", group);
        } else {
            params = Arrays.asList("dataId", dataId, "group", group, "tenant", tenant);
        }
        //擷取變更配置的接口調用
        result = this.agent.httpGet("/v1/cs/configs", (List)null, params, this.agent.getEncode(), readTimeout);
    } catch (IOException var9) {
        String message = String.format("[%s] [sub-server] get server config exception, dataId=%s, group=%s, tenant=%s", this.agent.getName(), dataId, group, tenant);
        LOGGER.error(message, var9);
        throw new NacosException(500, var9);
    }
    switch(result.code) {
    //擷取變更的配置成功，添加進緩存裡
    case 200:
        LocalConfigInfoProcessor.saveSnapshot(this.agent.getName(), dataId, group, tenant, result.content);
        //result.content 就是我們變更後的配置資訊
        return result.content;
    case 403:
        LOGGER.error("[{}] [sub-server-error] no right, dataId={}, group={}, tenant={}", new Object[]{this.agent.getName(), dataId, group, tenant});
        throw new NacosException(result.code, result.content);
    case 404:
        LocalConfigInfoProcessor.saveSnapshot(this.agent.getName(), dataId, group, tenant, (String)null);
        return null;
    case 409:
        LOGGER.error("[{}] [sub-server-error] get server config being modified concurrently, dataId={}, group={}, tenant={}", new Object[]{this.agent.getName(), dataId, group, tenant});
        throw new NacosException(409, "data being modified, dataId=" + dataId + ",group=" + group + ",tenant=" + tenant);
    default:
        LOGGER.error("[{}] [sub-server-error]  dataId={}, group={}, tenant={}, code={}", new Object[]{this.agent.getName(), dataId, group, tenant, result.code});
        throw new NacosException(result.code, "http error, code=" + result.code + ",dataId=" + dataId + ",group=" + group + ",tenant=" + tenant);
    }
}

Nacos用戶端通過 HTTP 協定與伺服器通信，那麼在伺服器源碼裡必然有對應接口的實作；
在 nacos-config 子產品下的 controller 包，提供了個 ConfigController 類來處理請求，其中有個 /listener 接口，是用戶端發起資料監聽的接口，其主要邏輯和源碼如下：
- 擷取用戶端需要監聽的可能發生變化的配置，并計算 MD5 值；
- ConfigServletInner.doPollingConfig() 開始執行長輪詢請求；

@PostMapping("/listener")
@Secured(action = ActionTypes.READ, parser = ConfigResourceParser.class)
public void listener(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
    request.setAttribute("org.apache.catalina.ASYNC_SUPPORTED", true);
    String probeModify = request.getParameter("Listening-Configs");
    if (StringUtils.isBlank(probeModify)) {
        throw new IllegalArgumentException("invalid probeModify");
    }
    
    probeModify = URLDecoder.decode(probeModify, Constants.ENCODE);
    
    Map<String, String> clientMd5Map;
    try {
        //計算 MD5 值
        clientMd5Map = MD5Util.getClientMd5Map(probeModify);
    } catch (Throwable e) {
        throw new IllegalArgumentException("invalid probeModify");
    }
    
    //【斷點步入 2.2】執行長輪詢請求
    inner.doPollingConfig(request, response, clientMd5Map, probeModify.length());
}

ConfigServletInner.doPollingConfig() 方法，該方法封裝了長輪詢的實作邏輯，同時相容短輪詢邏輯；

public String doPollingConfig(HttpServletRequest request, HttpServletResponse response, Map<String, String> clientMd5Map, int probeRequestSize) throws IOException {
    //長輪詢
    if (LongPollingService.isSupportLongPolling(request)) {
        //【斷點步入】長輪詢邏輯
        longPollingService.addLongPollingClient(request, response, clientMd5Map, probeRequestSize);
        return HttpServletResponse.SC_OK + "";
    }
    //相容短輪詢邏輯
    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 版本之前，傳回值放入表頭
    if (versionNum < START_LONG_POLLING_VERSION_NUM) {
        response.addHeader(Constants.PROBE_MODIFY_RESPONSE, oldResult);
        response.addHeader(Constants.PROBE_MODIFY_RESPONSE_NEW, newResult);
    } else {
        request.setAttribute("content", newResult);
    }
    Loggers.AUTH.info("new 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 + "";
}

LongPollingService.addLongPollingClient() 方法，裡面是長輪詢的核心處理邏輯，主要作用是把用戶端的長輪詢請求封裝成 ClientPolling 交給 scheduler 執行；

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);
    
    //為 LoadBalance 添加延遲時間，并提前 500ms 傳回響應，避免用戶端逾時（即逾時時間減 500ms 後指派給 timeout 變量）
    long timeout = Math.max(10000, Long.parseLong(str) - delayTime);
    //判斷是否為固定輪詢，是則 30s 後執行；否則 29.5s 後執行
    if (isFixedPolling()) {
        timeout = Math.max(10000, getFixedPollingInterval());
        // Do nothing but set fix polling timeout.
    } else {
        long start = System.currentTimeMillis();
        //和服務端的資料進行 MD5 對比，沒有變化則直接傳回
        List<String> changedGroups = MD5Util.compareMd5(req, rsp, clientMd5Map);
        if (changedGroups.size() > 0) {
            generateResponse(req, rsp, changedGroups);
            LogUtil.CLIENT_LOG.info("{}|{}|{}|{}|{}|{}|{}", System.currentTimeMillis() - start, "instant", RequestUtil.getRemoteIp(req), "polling", clientMd5Map.size(), probeRequestSize, changedGroups.size());
            return;
        } else if (noHangUpFlag != null && noHangUpFlag.equalsIgnoreCase(TRUE_STR)) {
            LogUtil.CLIENT_LOG.info("{}|{}|{}|{}|{}|{}|{}", System.currentTimeMillis() - start, "nohangup", RequestUtil.getRemoteIp(req), "polling", clientMd5Map.size(), probeRequestSize, changedGroups.size());
            return;
        }
    }
    String ip = RequestUtil.getRemoteIp(req);
    
    //一定要由 HTTP 線程調用，否則離開容器會立即發送響應
    final AsyncContext asyncContext = req.startAsync();
    
    //AsyncContext.setTimeout()的逾時時間不準，是以自己控制
    asyncContext.setTimeout(0L);
    
    //【點進去】調用 scheduler.execute 執行 ClientLongPolling 線程
    ConfigExecutor.executeLongPolling(new ClientLongPolling(asyncContext, clientMd5Map, ip, probeRequestSize, timeout, appName, tag));
}

我們找到 ClientLongPolling.run() 方法，這裡可以展現長輪詢定時機制的核心原理，通俗來說，就是：
- 服務端收到請求之後，不立即傳回，沒有變更則在延後 (30-0.5)s 把請求結果傳回給用戶端；
- 這就使得用戶端和服務端之間在 30s 之内資料沒有發生變化的情況下一直處于連接配接狀态；

@Override
public void run() {
    //啟動定時任務，延時時間為 29.5s
    asyncTimeoutFuture = ConfigExecutor.scheduleLongPolling(new Runnable() {
        @Override
        public void run() {
            try {
                //将 ClientLongPolling 執行個體本身添加到 allSubs 隊列中，它主要維護一個長輪詢的訂閱關系
                getRetainIps().put(ClientLongPolling.this.ip, System.currentTimeMillis());
                //定時任務執行後，先把 ClientLongPolling 執行個體本身從 allSubs 隊列中移除
                allSubs.remove(ClientLongPolling.this);
                //判斷是否為固定輪詢
                if (isFixedPolling()) {
                    LogUtil.CLIENT_LOG.info("{}|{}|{}|{}|{}|{}", (System.currentTimeMillis() - createTime), "fix", RequestUtil.getRemoteIp((HttpServletRequest) asyncContext.getRequest()),"polling", clientMd5Map.size(), probeRequestSize);
                    //比較資料的 MD5 值，判斷是否發生變更
                    List<String> changedGroups = MD5Util.compareMd5((HttpServletRequest) asyncContext.getRequest(), (HttpServletResponse) asyncContext.getResponse(), clientMd5Map);
                    if (changedGroups.size() > 0) {
                        //并将變更的結果通過response傳回給用戶端
                        sendResponse(changedGroups);
                    } else {
                        sendResponse(null);
                    }
                } else {
                    LogUtil.CLIENT_LOG.info("{}|{}|{}|{}|{}|{}", (System.currentTimeMillis() - createTime), "timeout", RequestUtil.getRemoteIp((HttpServletRequest) asyncContext.getRequest()),"polling", clientMd5Map.size(), probeRequestSize);
                    sendResponse(null);
                }
            } catch (Throwable t) {
                LogUtil.DEFAULT_LOG.error("long polling error:" + t.getMessage(), t.getCause());
            }
        }
    }, timeoutTime, TimeUnit.MILLISECONDS);
    allSubs.add(this);
}

當我們在 Nacos 伺服器或通過 API 方式變更配置後，會釋出一個 LocalDataChangeEvent 事件，該事件會被 LongPollingService 監聽；
這裡 LongPollingService 為什麼具有監聽功能在 1.3.1 版本後有些變化：
- 1.3.1 前： LongPollingService.onEvent()
- 1.3.1 後： Subscriber.onEvent()
在 Nacos 1.3.1 版本之前，通過 LongPollingService 繼承 AbstractEventListener 實作監聽，覆寫 onEvent() 方法；
點選檢視 github 上的 1.3.1 版本源碼；

@Service
public class LongPollingService extends AbstractEventListener {

   //省略其他代碼

    @Override
    public void onEvent(Event event) {
        if (isFixedPolling()) {
            // Ignore.
        } else {
            if (event instanceof LocalDataChangeEvent) {
                LocalDataChangeEvent evt = (LocalDataChangeEvent) event;
                //【點進去 2.4.2】通過線程池執行 DataChangeTask 任務
                scheduler.execute(new DataChangeTask(evt.groupKey, evt.isBeta, evt.betaIps));
            }
        }
    }
}

而在 1.3.2 版本之後，通過構造訂閱者實作，點選檢視 github 上的 1.3.2 版本源碼；

NotifyCenter.registerSubscriber(new Subscriber() {
    
    @Override
    public void onEvent(Event event) {
        if (isFixedPolling()) {
            // Ignore.
        } else {
            if (event instanceof LocalDataChangeEvent) {
                LocalDataChangeEvent evt = (LocalDataChangeEvent) event;
                //【點進去 2.4.2】通過線程池執行 DataChangeTask 任務
                ConfigExecutor.executeLongPolling(new DataChangeTask(evt.groupKey, evt.isBeta, evt.betaIps));
            }
        }
    }
    @Override
    public Class<? extends Event> subscribeType() {
        return LocalDataChangeEvent.class;
    }
});

效果是一樣的，實作了對 LocalDataChangeEvent 事件的監聽，并通過通過線程池執行 DataChangeTask 任務；

DataChangeTask.run() 方法，這個線程任務實作了

@Override
public void run() {
    try {
        ConfigCacheService.getContentBetaMd5(groupKey);
        //周遊 allSubs 中的用戶端長輪詢請求
        for (Iterator<ClientLongPolling> iter = allSubs.iterator(); iter.hasNext(); ) {
            ClientLongPolling clientSub = iter.next();
            //比較每一個用戶端長輪詢請求攜帶的groupKey，如果服務端變更的配置和用戶端請求關注的配置一緻，則直接傳回
            if (clientSub.clientMd5Map.containsKey(groupKey)) {
                //如果 beta 釋出且不在 beta 清單，則直接跳過
                if (isBeta && !CollectionUtils.contains(betaIps, clientSub.ip)) {
                    continue;
                }
                //如果 tag 釋出且不在 tag 清單，則直接跳過
                if (StringUtils.isNotBlank(tag) && !tag.equals(clientSub.tag)) {
                    continue;
                }
                getRetainIps().put(clientSub.ip, System.currentTimeMillis());
                iter.remove(); //删除訂閱關系
                LogUtil.CLIENT_LOG.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.DEFAULT_LOG.error("data change error: {}", ExceptionUtil.getStackTrace(t));
    }
}

NacosPropertySourceLocator.locate()：初始化 ConfigService 對象，定位配置；
- NacosFactory.createConfigService()：建立配置伺服器；
- ConfigFactory.createConfigService()：利用反射機制建立配置伺服器；
  - NacosConfigService.NacosConfigService()：NacosConfigService 的構造方法；
    - MetricsHttpAgent.MetricsHttpAgent()：初始化 HttpAgent；
    - ClientWorker.ClientWorker()：初始化 ClientWorker；
      - Executors.newScheduledThreadPool()：建立 executor 線程池；
      - Executors.newScheduledThreadPool()：建立 executorService 線程池；
      - ClientWorker.checkConfigInfo()：使用 executor 線程池檢查配置是否發生變化；
        
        LongPollingRunnable.run()：運作長輪詢定時線程；
        
        ClientWorker.checkLocalConfig()：檢查本地配置；
        
        ClientWorker.checkUpdateDataIds()：檢查服務端對應的配置是否發生變更；
        
        ClientWorker.checkUpdateConfigStr()：檢查服務端對應的配置是否發生變更；
        
        MetricsHttpAgent.httpPost()：調用 /v1/cs/configs/listener 接口實作長輪詢請求；
        
        ClientWorker.getServerConfig()：讀取變更配置
        
        MetricsHttpAgent.httpGet()：調用 /v1/cs/configs 接口擷取配置；

ConfigController.listener()：伺服器接收請求；
- MD5Util.getClientMd5Map()：計算 MD5 值；
- ConfigServletInner.doPollingConfig()：執行長輪詢請求；
  - LongPollingService.addLongPollingClient()：長輪詢的核心處理邏輯，提前 500ms 傳回響應；
    - HttpServletRequest.getHeader()：擷取用戶端設定的請求逾時時間；
    - MD5Util.compareMd5()：和服務端的資料進行 MD5 對比；
    - ConfigExecutor.executeLongPolling()：建立 ClientLongPolling 線程執行定時任務；
      - ClientLongPolling.run()：長輪詢定時機制的實作邏輯；
        
        ConfigExecutor.scheduleLongPolling()：啟動定時任務，延時時間為 29.5s；
        
        Map.put()：将 ClientLongPolling 執行個體本身添加到 allSubs 隊列中；
        
        Queue.remove()：把 ClientLongPolling 執行個體本身從 allSubs 隊列中移除；
        
        MD5Util.compareMd5()：比較資料的 MD5 值；
        
        LongPollingService.sendResponse()：将變更的結果通過 response 傳回給用戶端；

Nacos 伺服器上的配置發生變更後，釋出一個 LocalDataChangeEvent 事件；
Subscriber.onEvent()：監聽 LocalDataChangeEvent 事件（1.3.2 版本後）；
- ConfigExecutor.executeLongPolling()：通過線程池執行 DataChangeTask 任務；
  - DataChangeTask.run()：根據 groupKey 傳回配置；

新人制作，如有錯誤，歡迎指出，感激不盡！

歡迎關注公衆号，會分享一些更日常的東西！

如需轉載，請标注出處！

微服務架構 | *2.5 Nacos 長輪詢定時機制的源碼分析

繼續閱讀

Kafka：Topic概念與API介紹

ZooKeeper ： Curator架構之分布式屏障DistributedDoubleBarrier

RabbitMQ：交換機（fanout exchange）

Doris SQL 原了解析

ZooKeeper ： Curator架構之分布式鎖InterProcessMutex

阿裡巴巴分布式服務架構 Dubbo 團隊成員梁飛專訪

資料遷移方法資料遷移原則資料遷移之雙寫方案資料遷移之級聯同步方案

微服務-性能壓測\緩存redis和分布式鎖redisson和SpringCache

Nacos 2.0 更新前後性能對比壓測

Spring資料和Redis

redis叢集資料一緻性_RedisRaft為Redis叢集帶來強大的資料一緻性

Centos7 下 Hadoop 2.6.4 分布式叢集環境搭建摘要叢集準備安裝JDK 安裝 Hadoop 2.6.4 部署 slaver1-slaver4 啟動 hadoop 叢集成功了

niginx的location的location詳細規則和優先級關系以及測試案例2.1比對文法2.2優先級判斷2.3練習3.1處理邏輯如何執行3.2功能測試3.3流程解析

celery使用入門

MapReduce的幾個企業級經典面試案例MapReduce的幾個企業級經典面試案例