Nacos 和 Apollo 中的 长轮询 定时机制,太好用了
今天这篇文章来介绍一下 Nacos 配置中心的原理之一:长轮询机制的应用
为方便理解与表达,这里把 Nacos 控制台和 Nacos 注册中心称为 Nacos 服务器(就是 web 界面那个),我们编写的业务服务称为 Nacso 客户端;
Nacos 动态监听的长轮询机制原理图,本篇将围绕这张图剖析长轮询定时机制的原理:
ConfigService 是 Nacos 客户端提供的用于访问实现配置中心基本操作的类型,我们将从 ConfigService 的实例化开始长轮询定时机制的源码之旅;
1. 客户端的长轮询定时机制
我们从 NacosPropertySourceLocator.locate()开始【断点步入】:
1.1 利用反射机制实例化 NacosConfigService 对象
客户端的长轮询定时任务是在 NacosFactory.createConfigService() 方法中,构建 ConfigService 对象是实例时启动的,我们接着 1.1 处的源码;
进入 NacosFactory.createConfigService():
进入 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\ 里)的数据是否存在变更,如果有变更则直接触发通知;
注意:这里的断点需要注意 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 服务器上读取具体的配置内容;
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 任务;
评论