Apollo 中的 长轮询 定时机制
今天这篇文章来介绍一下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);}
进入
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); }}
注意：这里的断点需要注意 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() 将变更的配置保存到本地；
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 任务；
Apollo 中的 长轮询 定时机制相关推荐
- java 长轮询_java – Spring中的长轮询
我们有一个独特的案例,我们需要与外部API接口,这需要我们长时间轮询他们的端点以获得他们所谓的实时事件. 问题是我们可能有多达80,000人/设备在任何给定时间点击此端点,监听事件,每个设备/人1个连 ...
- RocketMQ 的长轮询如何实现 ?
最近仍然畅游在RocketMQ的源码中,这几天刚好翻到了消费者的源码,发现RocketMQ的对于push消费方式的实现简直太聪明了,所以趁着我脑子里还有点印象的时候,赶紧来写一篇文章,来掰扯一下,防止 ...
- 如何实施基本的“长轮询”？
我可以找到很多关于Long Polling如何工作的信息(例如, 这个和这个 ),但没有关于如何在代码中实现它的简单示例. 我所能找到的只是依赖于Dojo JS框架的cometd ,以及一个相当复杂的 ...
- HTTP - 长连接 短连接 长轮询 短轮询 心跳机制
错觉与突然的察觉 大多数人都知道HTTP1.0不支持长连接,知道HTTP1.1支持长连接. 这是业界的一个常识. 然而这样的描述导致了一些不做网络底层开发的开发者都下意识的认为HTTP1.1是一个可以 ...
- 从Apollo看长轮询
前言 如果让我设计一个配置中心,最先想到的两个核心功能:一个是如何将配置存储下来,另一个是怎么能够实时的获取到最新的配置;最简单的方式我们可以直接利用现有的一些中间件:Zookeeper.Redis等 ...
- 【Soul源码阅读】12.soul-admin 与 soul-bootstrap 同步机制之 http 长轮询解析（上）
目录 1.前情回顾 2.配置 2.1 soul-admin 2.2 soul-bootstrap 3.启动 3.1 启动 soul-admin 3.2 启动 soul-bootstrap 3.2.1 ...
- Node中的事件轮询机制
文章目录 2 node中的事件循环模型 2-1 一些属性 2-2 循环模型 node事件循环总共有==六个阶段== process.nextTick()函数 __实例__ 2 node中的事件循环模型 ...
- javaweb认识在web应用中重要的轮询机制（ajax）
**轮询是用来解决服务器压力过大的问题的.**如果保持多个长连接,服务器压力会过大,因此.专门建立一个轮询请求的接口,里面只保留一个任务id,只需要发送任务id,就可以获取当前任务的情况.如果返回了结 ...
- 转---谈谈HTTP协议中的短轮询、长轮询、长连接和短连接
作者:伯乐在线专栏作者 - 左潇龙 http://web.jobbole.com/85541/ 如有好文章投稿,请点击 → 这里了解详情 引言 最近刚到公司不到一个月,正处于熟悉项目和源码的阶段,因此 ...
最新文章
- oracle 测试sql执行时间_通过错误的SQL来测试推理SQL的解析过程
- 今天决定写一篇LNMP的深入调优，
- python语言翻译-从Python到CIL(C中间语言)的翻译
- centos修改磁盘uuid_CentOS 6如何修改磁盘配额限制
- 详解Linux 五种IO模型
- js 动态绑定事件 on click 完美解决绑定不成功
- python messagebox弹窗退出_python 弹窗提示警告框MessageBox的实例
- 哪些人可以报考公务员 哪些人不能报考公务员
- 10.docker build
- python图像exif信息复制
- 简易python程序 解决linux连接steam社区错误代码：-101
- 网络蠕虫和僵尸网络等恶意代码防范技术原理
- 学生信息管理系统代码
- 第十六周项目3电子词典
- 东方卫视携微软小冰打造人工智能新闻节目引热议
- DeepSort论文学习
- 经常调试笔记本服务器显示器,瞎折腾！闲置损坏笔记本电脑改造的DIY液晶显示屏！蜗牛星际附件。...
- [贪心]leetcode55：跳跃游戏(medium)
- Mac使用Aria2下载百度网盘，突破下载限速的方法教程
- Linux下U盘自动识别和挂载