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

2022-07-14 本文已影响0人程序员阿远

今天这篇文章来介绍一下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中的长轮询定时机制，太好用了

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 服务器配置变更的事件监听

猜你喜欢

热点阅读

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 服务器配置变更的事件监听

猜你喜欢

热点阅读

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