前言
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 動態監聽的長輪詢機制原理圖，本篇將圍繞這張圖剖析長輪詢定時機制的原理；

Nacos 動態監聽的長輪詢機制

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

1. 客戶端的長輪詢定時機制

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

1.1 利用反射機制實體化 NacosConfigService 物件

客戶端的長輪詢定時任務是在 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);
    }
}

1.2 NacosConfigService 的構造方法里啟動長輪詢定時任務

進入 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);
}

1.2.1 初始化 HttpAgent

MetricsHttpAgent 類的設計如下：
ServerHttpAgent 類的設計如下：

1.2.2 初始化 ClientWorker

進入 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;
    }
}

1.3 檢查配置變更，讀取變更配置 LongPollingRunnable.run()

因為我們沒有這么多配置項，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 = https://www.cnblogs.com/dlhjw/archive/2022/01/24/(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），進入后才好理解；

1.3.1 檢查配置變更 ClientWorker.checkUpdateDataIds()

我們點進 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();
    }
}

1.3.2 讀取變更配置 ClientWorker.getServerConfig()

進入 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);
    }
}

result.content

2. 服務端的長輪詢定時機制

2.1 服務器接收請求 ConfigController.listener()

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

2.2 執行長輪詢請求 ConfigServletInner.doPollingConfig()

進入 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));
}

2.3 創建執行緒執行定時任務 ClientLongPolling.run()

我們找到 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);
}

2.4 監聽配置變更事件

2.4.1 監聽 LocalDataChangeEvent 事件的實作

當我們在 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 任務；

2.4.2 監聽事件后的處理邏輯 DataChangeTask.run()

我們找到 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));
    }
}

3. 原始碼結構圖小結

3.1 客戶端的長輪詢定時機制

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 介面獲取配置；

3.2 服務端的長輪詢定時機制

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 回傳給客戶端；

3.3 Nacos 服務器配置變更的事件監聽

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

最后

新人制作，如有錯誤，歡迎指出，感激不盡！歡迎關注公眾號，會分享一些更日常的東西！如需轉載，請標注出處！

轉載請註明出處，本文鏈接：https://www.uj5u.com/ruanti/420432.html

標籤：其他

上一篇：一圖搞懂掃碼登錄的技術原理

下一篇：會計學包含的兩種程式設計思想

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

前言