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

www.ydisp.cn/oss/202207/14/979122c877ff4de792289561be60142173cf01.jpg" alt="图片" title="图片" style="width: 640px; visibility: visible; height: 217px;" data-type="block">

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 对象；

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() {

ListcacheDatas = 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】通过长轮询请求检查服务端对应的配置是否发生变更

ListchangedGroupKeys = 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() 方法，云南idc服务商其实现逻辑与源码如下：

通过MetricsHttpAgent.httpPost()​ 方法（上面 1.2.1 有提到）调用/v1/cs/configs/listener 接口实现长轮询请求；长轮询请求在实现层面只是设置了一个比较长的超时时间，默认是 30s；如果服务端的数据发生了变更，客户端会收到一个 HttpResult ，服务端返回的是存在数据变更的 Data ID、Group、Tenant；获得这些信息之后，在LongPollingRunnable.run() 方法中调用 getServerConfig() 去 Nacos 服务器上读取具体的配置内容；ListcheckUpdateConfigStr(String probeUpdateString, boolean isInitializingCacheList) throws IOException {

Listparams = Arrays.asList("Listening-Configs", probeUpdateString);

Listheaders = 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() 将变更的配置保存到本地；

2. 服务端的长轮询定时机制

2.1 服务器接收请求 ConfigController.listener()

Nacos客户端 通过 HTTP 协议与服务器通信，那么在服务器源码里必然有对应接口的实现；

在 nacos-config 模块下的 controller 包，提供了个 ConfigController 类来处理请求，其中有个 /listener 接口，是客户端发起数据监听的源码下载接口，其主要逻辑和源码如下：

获取客户端需要监听的可能发生变化的配置，并计算 MD5 值；ConfigServletInner.doPollingConfig() 开始执行长轮询请求；2.2 执行长轮询请求 ConfigSer

vletInner.doPollingConfig()

进入 ConfigServletInner.doPollingConfig() 方法，该方法封装了长轮询的实现逻辑，同时兼容短轮询逻辑；

进入 LongPollingService.addLongPollingClient() 方法，里面是长轮询的核心处理逻辑，主要作用是把客户端的长轮询请求封装成 ClientPolling 交给 scheduler 执行；

2.3 创建线程执行定时任务 ClientLongPolling.run()

我们找到 ClientLongPolling.run() 方法，这里可以体现长轮询定时机制的核心原理，通俗来说，就是：

服务端收到请求之后，不立即返回，没有变更则在延后 (30-0.5)s 把请求结果返回给客户端；

这就使得客户端和服务端之间在 30s 之内数据没有发生变化的情况下一直处于连接状态；

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() 方法；

而在 1.3.2 版本之后，通过构造订阅者实现

效果是一样的，实现了对 LocalDataChangeEvent 事件的监听，并通过通过线程池执行 DataChangeTask 任务；

2.4.2 监听事件后的处理逻辑 DataChangeTask.run()

我们找到 DataChangeTask.run() 方法，这个线程任务实现了

3. 源码结构图小结

3.1 客户端的长轮询定时机制NacosPropertySourceLocator.locate() ：初始化 ConfigService 对象，定位配置；NacosConfigService.NacosConfigService()：NacosConfigService 的构造方法；Executors.newScheduledThreadPool()：创建 executor 线程池；Executors.newScheduledThreadPool()：创建 executorService 线程池；ClientWorker.checkConfigInfo()：使用 executor 线程池检查配置是否发生变化；ClientWorker.checkLocalConfig()：检查本地配置；ClientWorker.checkUpdateDataIds()：检查服务端对应的配置是否发生变更；ClientWorker.getServerConfig()：读取变更配置MetricsHttpAgent.httpPost()：调用 /v1/cs/configs/listener 接口实现长轮询请求；ClientWorker.checkUpdateConfigStr()：检查服务端对应的配置是否发生变更；MetricsHttpAgent.httpGet()：调用 /v1/cs/configs 接口获取配置；LongPollingRunnable.run()：运行长轮询定时线程；MetricsHttpAgent.MetricsHttpAgent()：初始化 HttpAgent；ClientWorker.ClientWorker()：初始化 ClientWorker；NacosFactory.createConfigService()：创建配置服务器；ConfigFactory.createConfigService()：利用反射机制创建配置服务器；3.2 服务端的长轮询定时机制ConfigController.listener() ：服务器接收请求；LongPollingService.addLongPollingClient()：长轮询的核心处理逻辑，提前 500ms 返回响应；ClientLongPolling.run()：长轮询定时机制的实现逻辑；Map.put()：将 ClientLongPolling 实例本身添加到 allSubs 队列中；Queue.remove()：把 ClientLongPolling 实例本身从 allSubs 队列中移除；MD5Util.compareMd5()：比较数据的 MD5 值；LongPollingService.sendResponse()：将变更的结果通过 response 返回给客户端；ConfigExecutor.scheduleLongPolling()：启动定时任务，延时时间为 29.5s；HttpServletRequest.getHeader()：获取客户端设置的请求超时时间；MD5Util.compareMd5()：和服务端的数据进行 MD5 对比；ConfigExecutor.executeLongPolling()：创建 ClientLongPolling 线程执行定时任务；MD5Util.getClientMd5Map()：计算 MD5 值；ConfigServletInner.doPollingConfig()：执行长轮询请求；3.3 Nacos 服务器配置变更的事件监听

Nacos 服务器上的配置发生变更后，发布一个 LocalDataChangeEvent 事件；

Subscriber.onEvent() ：监听 LocalDataChangeEvent 事件（1.3.2 版本后）；

DataChangeTask.run()：根据 groupKey 返回配置；ConfigExecutor.executeLongPolling()：通过线程池执行 DataChangeTask 任务；