Nacos 和 Apollo中的长轮询定时机制，太好用了

2023-02-28

this 服务器长轮询

今天这篇文章来介绍一下Nacos配置中心的原理之一：长轮询机制的应用为方便理解与表达，这里把Nacos控制台和Nacos注册中心称为Nacos服务器（就是web界面那个），我们编写的业务服务称为Nacso客户端；Nacos动态监听的长轮询机制原理图，本篇将围绕这张图剖析长轮询定时机制的原理：Conf

今天这篇文章来介绍一下Nacos配置中心的原理之一：长轮询机制的应用

为方便理解与表达，这里把 Nacos 控制台和 Nacos 注册中心称为 Nacos 服务器（就是 web 界面那个），我们编写的业务服务称为 Nacso 客户端；

Nacos 动态监听的长轮询机制原理图，本篇将围绕这张图剖析长轮询定时机制的原理：

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);
}1.
2.
3.
4.

进入 ConfigFactory.createConfigService()，发现其使用反射机制实例化 NacosConfigService 对象；

1.2 NacosConfigService 的构造方法里启动长轮询定时任务

进入 NacosConfigService.NacosConfigService() 构造方法，里面设置了一些更远程任务相关的属性；

1.2.1 初始化 HttpAgent

MetricsHttpAgent 类的设计如下：

ServerHttpAgent 类的设计如下：

1.2.2 初始化 ClientWorker

进入 ClientWorker.ClientWorker() 构造方法，主要是创建了两个定时调度的线程池，并启动一个定时任务；

进入 ClientWorker.checkConfigInfo()，每隔 10s 检查一次配置是否发生变化；

cacheMap：是一个 AtomicReference<Map<String, CacheData>> 对象，用来存储监听变更的缓存集合，key 是根据 datalD/group/tenant（租户）拼接的值。Value 是对应的存储在 Nacos 服务器上的配置文件的内容；
长轮询任务拆分：默认情况下，每个长轮询 LongPollingRunnable 任务处理3000个监听配置集。如果超过3000个，则需要启动多个 LongPollingRunnable 去执行；

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 = (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);
    }
}1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
32.
33.
34.
35.
36.
37.
38.
39.
40.
41.
42.
43.
44.
45.
46.
47.
48.
49.
50.
51.
52.
53.
54.
55.
56.
57.
58.
59.
60.
61.
62.
63.
64.
65.
66.

注意：这里的断点需要在 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.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.

1.3.2 读取变更配置 ClientWorker.getServerConfig()

进入 ClientWorker.getServerConfig() 方法；读取服务器上的变更配置；最终调用的是 MetricsHttpAgent.httpGet() 方法（上面 1.2.1 有提到），调用 /v1/cs/configs 接口获取配置；然后通过调用 LocalConfigInfoProcessor.saveSnapshot() 将变更的配置保存到本地；