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

作者: 程序员阿远 | 来源:发表于2022-07-14 16:56 被阅读0次

Nacos 和 Apollo中的长轮询定时机制，太好用了
Apollo 8 — ConfigService 异步轮询接口的
微服务架构 | *2.5 Nacos 长轮询定时机制的源码分析
Apollo 3 — 定时/长轮询拉取配置的设计
Apollo 长轮询超时错误
三招通过Apollo和nacos的能力进行国际化热更新
简述长轮询实现在线聊天实现
http长短连接和长短轮询、webSocket
MySQL 5.6 5.7 如何自定义行锁超时等待（wait
Tornado-长轮询_webSocket

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

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

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

image.png

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

image.png

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

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

image.png

1.2.1 初始化 HttpAgent

MetricsHttpAgent 类的设计如下：

image.png

ServerHttpAgent 类的设计如下：

image.png

1.2.2 初始化 ClientWorker

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

image.png

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

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

image.png

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

image.png

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

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

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

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

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

2.2 执行长轮询请求 ConfigSer

image.png

vletInner.doPollingConfig()

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

image.png

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

image.png

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

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

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

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

image.png

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

image.png

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

image.png

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

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

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

image.png

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 任务；

原文链接：
https://developer.51cto.com/article/713995.html

Nacos 和 Apollo中的长轮询定时机制，太好用了
今天这篇文章来介绍一下Nacos配置中心的原理之一：长轮询机制的应用为方便理解与表达，这里把 Nacos 控制台...
Apollo 8 — ConfigService 异步轮询接口的
源码 Apollo 长轮询的实现，是通过客户端轮询 /notifications/v2 接口实现的。具体代码在 c...
微服务架构 | *2.5 Nacos 长轮询定时机制的源码分析
前言参考资料：《Spring Microservices in Action》《Spring Cloud Ali...
Apollo 3 — 定时/长轮询拉取配置的设计
前言如上图所示，Apollo portal 更新配置后，进行轮询的客户端获取更新通知，然后再调用接口获取最新配置...
Apollo 长轮询超时错误
最近线上碰到一个线上的go程序Apollo报错，请求apollo的时候超时，记录一下错误的原因。错误信息 Apo...
三招通过Apollo和nacos的能力进行国际化热更新
通过Apollo和nacos的能力进行国际化热更新 1、apollo的自动刷新 Apollo（阿波罗）是一款可靠的...
简述长轮询实现在线聊天实现
最近在做web端客服和客户即时聊天的功能，采用了长轮询来进行聊天细细的即时推送。长轮询Long Polling ...
http长短连接和长短轮询、webSocket
本文主要介绍长连接、短连接、长轮询、短轮询和 webSocket。长连接、短连接、长轮询、短轮询是基于http的...
MySQL 5.6 5.7 如何自定义行锁超时等待（wait
1. 问题最近由于业务的需要，写了个基于数据库锁机制的分布式调度简易框架，用于处理业务中的补偿任务和定时轮询任务...
Tornado-长轮询_webSocket
长轮询轮询(Polling)：是指不管服务器端有没有更新，客户端(通常是指浏览器)都定时的发送请求进行查询，轮询...

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

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 执行长轮询请求 ConfigSer

vletInner.doPollingConfig()

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

2.4 监听配置变更事件

2.4.1 监听 LocalDataChangeEvent 事件的实现

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

3. 源码结构图小结

3.1 客户端的长轮询定时机制

3.2 服务端的长轮询定时机制

3.3 Nacos 服务器配置变更的事件监听

相关文章