Nacos 长轮询定时机制的源码分析

vlambda
2022-02-22

Nacos 长轮询定时机制的源码分析

前言

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

由于篇幅有限，这里将源码分析分为上下两篇，其中上篇讲获取配置与事件订阅机制，下篇讲长轮询定时机制；在《微服务架构 | 2.2 Alibaba Nacos 的统一配置管理》中提到一张 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 对象；

public static ConfigService createConfigService(Properties properties) throws NacosException {
    try {
        //通过 Class.forName 来加载 NacosConfigService
        Class<?> driverImplClass = Class.forName("com.alibaba.nacos.client.config.NacosConfigService");
        //根据对象的属性创建对象构造器（反射）
        Constructor constructor = driverImplClass.getConstructor(Properties.class);
        //【断点步入 1.2】通过构造器实例化对象（反射）
        ConfigService vendorImpl = (ConfigService)constructor.newInstance(properties);
        return vendorImpl;
    } catch (Throwable var4) {
        throw new NacosException(-400, var4);
    }
}

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

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

public NacosConfigService(Properties properties) throws NacosException {
    String encodeTmp = properties.getProperty("encode");
    if (StringUtils.isBlank(encodeTmp)) {
        this.encode = "UTF-8";
    } else {
        this.encode = encodeTmp.trim();
    }
    //初始化命名空间
    this.initNamespace(properties);
    //【断点步入 1.2.1】初始化 HttpAgent，用到了装饰器模式，实际工作的类是 ServerHttpAgent
    this.agent = new MetricsHttpAgent(new ServerHttpAgent(properties));
    this.agent.start();
    //【断点步入 1.2.2】ClientWorker 是客户端的一个工作类，agent 作为参数传入 ClientWorker
    this.worker = new ClientWorker(this.agent, this.configFilterChainManager, properties);
}

1.2.1 初始化 HttpAgent

MetricsHttpAgent 类的设计如下：
ServerHttpAgent 类的设计如下：

1.2.2 初始化 ClientWorker

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

public ClientWorker(final HttpAgent agent, ConfigFilterChainManager configFilterChainManager, Properties properties) {
    this.agent = agent;
    this.configFilterChainManager = configFilterChainManager;
    this.init(properties);
    //创建 executor 线程池，只拥有一个核心线程，每隔 10ms 就会执行一次 checkConfiglnfo() 方法，检查配置信息
    this.executor = Executors.newScheduledThreadPool(1, new ThreadFactory() {
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r);
            t.setName("com.alibaba.nacos.client.Worker." + agent.getName());
            t.setDaemon(true);
            return t;
        }
    });
    //创建 executorService 线程池，只完成了初始化，后续会用到，主要用于实现客户端的定时长轮询功能
    this.executorService = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors(), new ThreadFactory() {
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r);
            t.setName("com.alibaba.nacos.client.Worker.longPolling." + agent.getName());
            t.setDaemon(true);
            return t;
        }
    });
    //使用 executor 启动一个每隔 10s 执行一次的定时任务
    this.executor.scheduleWithFixedDelay(new Runnable() {
        public void run() {
            try {
                //【断点步入】检查配置是否发生变化
                ClientWorker.this.checkConfigInfo();
            } catch (Throwable var2) {
                ClientWorker.LOGGER.error("[" + agent.getName() + "] [sub-check] rotate check error", var2);
            }
        }
    }, 1L, 10L, TimeUnit.MILLISECONDS);
}

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

cacheMap：是一个 AtomicReference
<map

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

public void checkConfigInfo() {
    //分任务
    int listenerSize = ((Map)this.cacheMap.get()).size();
    //向上取整为批数
    int longingTaskCount = (int)Math.ceil((double)listenerSize / ParamUtil.getPerTaskConfigSize());
    //如果监听配置集超过 3000，就创建多个 LongPollingRunnable 线程
    if ((double)longingTaskCount > this.currentLongingTaskCount) {
        for(int i = (int)this.currentLongingTaskCount; i < longingTaskCount; ++i) {
            //【点进去】LongPollingRunnable 实际上是一个线程
            this.executorService.execute(new ClientWorker.LongPollingRunnable(i));
        }
        this.currentLongingTaskCount = (double)longingTaskCount;
    }
}

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

public String getServerConfig(String dataId, String group, String tenant, long readTimeout) throws NacosException {
    if (StringUtils.isBlank(group)) {
        group = "DEFAULT_GROUP";
    }
    HttpResult result = null;
    try {
        List<String> params = null;
        if (StringUtils.isBlank(tenant)) {
            params = Arrays.asList("dataId", dataId, "group", group);
        } else {
            params = Arrays.asList("dataId", dataId, "group", group, "tenant", tenant);
        }
        //获取变更配置的接口调用
        result = this.agent.httpGet("/v1/cs/configs", (List)null, params, this.agent.getEncode(), readTimeout);
    } catch (IOException var9) {
        String message = String.format("[%s] [sub-server] get server config exception, dataId=%s, group=%s, tenant=%s", this.agent.getName(), dataId, group, tenant);
        LOGGER.error(message, var9);
        throw new NacosException(500, var9);
    }
    switch(result.code) {
    //获取变更的配置成功，添加进缓存里
    case 200:
        LocalConfigInfoProcessor.saveSnapshot(this.agent.getName(), dataId, group, tenant, result.content);
        //result.content 就是我们变更后的配置信息
        return result.content;
    case 403:
        LOGGER.error("[{}] [sub-server-error] no right, dataId={}, group={}, tenant={}", new Object[]{this.agent.getName(), dataId, group, tenant});
        throw new NacosException(result.code, result.content);
    case 404:
        LocalConfigInfoProcessor.saveSnapshot(this.agent.getName(), dataId, group, tenant, (String)null);
        return null;
    case 409:
        LOGGER.error("[{}] [sub-server-error] get server config being modified concurrently, dataId={}, group={}, tenant={}", new Object[]{this.agent.getName(), dataId, group, tenant});
        throw new NacosException(409, "data being modified, dataId=" + dataId + ",group=" + group + ",tenant=" + tenant);
    default:
        LOGGER.error("[{}] [sub-server-error]  dataId={}, group={}, tenant={}, code={}", new Object[]{this.agent.getName(), dataId, group, tenant, result.code});
        throw new NacosException(result.code, "http error, code=" + result.code + ",dataId=" + dataId + ",group=" + group + ",tenant=" + tenant);
    }
}

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

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

Nacos客户端通过 HTTP 协议与服务器通信，那么在服务器源码里必然有对应接口的实现；
在 nacos-config 模块下的 controller 包，提供了个 ConfigController 类来处理请求，其中有个 /listener 接口，是客户端发起数据监听的接口，其主要逻辑和源码如下：

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

@PostMapping("/listener")
@Secured(action = ActionTypes.READ, parser = ConfigResourceParser.class)
public void listener(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
    request.setAttribute("org.apache.catalina.ASYNC_SUPPORTED", true);
    String probeModify = request.getParameter("Listening-Configs");
    if (StringUtils.isBlank(probeModify)) {
        throw new IllegalArgumentException("invalid probeModify");
    }

    probeModify = URLDecoder.decode(probeModify, Constants.ENCODE);

    Map<String, String> clientMd5Map;
    try {
        //计算 MD5 值
        clientMd5Map = MD5Util.getClientMd5Map(probeModify);
    } catch (Throwable e) {
        throw new IllegalArgumentException("invalid probeModify");
    }

    //【断点步入 2.2】执行长轮询请求
    inner.doPollingConfig(request, response, clientMd5Map, probeModify.length());
}

2.2 执行长轮询请求 ConfigServletInner.doPollingConfig()

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

public String doPollingConfig(HttpServletRequest request, HttpServletResponse response, Map<String, String> clientMd5Map, int probeRequestSize) throws IOException {
    //长轮询
    if (LongPollingService.isSupportLongPolling(request)) {
        //【断点步入】长轮询逻辑
        longPollingService.addLongPollingClient(request, response, clientMd5Map, probeRequestSize);
        return HttpServletResponse.SC_OK + "";
    }
    //兼容短轮询逻辑
    List<String> changedGroups = MD5Util.compareMd5(request, response, clientMd5Map);
    //兼容短轮询 result
    String oldResult = MD5Util.compareMd5OldResult(changedGroups);
    String newResult = MD5Util.compareMd5ResultString(changedGroups);
    String version = request.getHeader(Constants.CLIENT_VERSION_HEADER);
    if (version == null) {
        version = "2.0.0";
    }
    int versionNum = Protocol.getVersionNumber(version);

    //在 2.0.4 版本之前，返回值放入表头
    if (versionNum < START_LONG_POLLING_VERSION_NUM) {
        response.addHeader(Constants.PROBE_MODIFY_RESPONSE, oldResult);
        response.addHeader(Constants.PROBE_MODIFY_RESPONSE_NEW, newResult);
    } else {
        request.setAttribute("content", newResult);
    }
    Loggers.AUTH.info("new content:" + newResult);
    //禁用缓存
    response.setHeader("Pragma", "no-cache");
    response.setDateHeader("Expires", 0);
    response.setHeader("Cache-Control", "no-cache,no-store");
    response.setStatus(HttpServletResponse.SC_OK);
    return HttpServletResponse.SC_OK + "";
}

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

public void addLongPollingClient(HttpServletRequest req, HttpServletResponse rsp, Map<String, String> clientMd5Map, int probeRequestSize) {
    //获取客户端设置的请求超时时间
    String str = req.getHeader(LongPollingService.LONG_POLLING_HEADER);
    String noHangUpFlag = req.getHeader(LongPollingService.LONG_POLLING_NO_HANG_UP_HEADER);
    String appName = req.getHeader(RequestUtil.CLIENT_APPNAME_HEADER);
    String tag = req.getHeader("Vipserver-Tag");
    int delayTime = SwitchService.getSwitchInteger(SwitchService.FIXED_DELAY_TIME, 500);

    //为 LoadBalance 添加延迟时间，并提前 500ms 返回响应，避免客户端超时（即超时时间减 500ms 后赋值给 timeout 变量）
    long timeout = Math.max(10000, Long.parseLong(str) - delayTime);
    //判断是否为固定轮询，是则 30s 后执行；否则 29.5s 后执行
    if (isFixedPolling()) {
        timeout = Math.max(10000, getFixedPollingInterval());
        // Do nothing but set fix polling timeout.
    } else {
        long start = System.currentTimeMillis();
        //和服务端的数据进行 MD5 对比，没有变化则直接返回
        List<String> changedGroups = MD5Util.compareMd5(req, rsp, clientMd5Map);
        if (changedGroups.size() > 0) {
            generateResponse(req, rsp, changedGroups);
            LogUtil.CLIENT_LOG.info("{}|{}|{}|{}|{}|{}|{}", System.currentTimeMillis() - start, "instant", RequestUtil.getRemoteIp(req), "polling", clientMd5Map.size(), probeRequestSize, changedGroups.size());
            return;
        } else if (noHangUpFlag != null && noHangUpFlag.equalsIgnoreCase(TRUE_STR)) {
            LogUtil.CLIENT_LOG.info("{}|{}|{}|{}|{}|{}|{}", System.currentTimeMillis() - start, "nohangup", RequestUtil.getRemoteIp(req), "polling", clientMd5Map.size(), probeRequestSize, changedGroups.size());
            return;
        }
    }
    String ip = RequestUtil.getRemoteIp(req);

    //一定要由 HTTP 线程调用，否则离开容器会立即发送响应
    final AsyncContext asyncContext = req.startAsync();

    //AsyncContext.setTimeout()的超时时间不准，所以自己控制
    asyncContext.setTimeout(0L);

    //【点进去】调用 scheduler.execute 执行 ClientLongPolling 线程
    ConfigExecutor.executeLongPolling(new ClientLongPolling(asyncContext, clientMd5Map, ip, probeRequestSize, timeout, appName, tag));
}

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

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

服务端收到请求之后，不立即返回，没有变更则在延后 (30-0.5)s 把请求结果返回给客户端；
这就使得客户端和服务端之间在 30s 之内数据没有发生变化的情况下一直处于连接状态；

@Override
public void run() {
    //启动定时任务，延时时间为 29.5s
    asyncTimeoutFuture = ConfigExecutor.scheduleLongPolling(new Runnable() {
        @Override
        public void run() {
            try {
                //将 ClientLongPolling 实例本身添加到 allSubs 队列中，它主要维护一个长轮询的订阅关系
                getRetainIps().put(ClientLongPolling.this.ip, System.currentTimeMillis());
                //定时任务执行后，先把 ClientLongPolling 实例本身从 allSubs 队列中移除
                allSubs.remove(ClientLongPolling.this);
                //判断是否为固定轮询
                if (isFixedPolling()) {
                    LogUtil.CLIENT_LOG.info("{}|{}|{}|{}|{}|{}", (System.currentTimeMillis() - createTime), "fix", RequestUtil.getRemoteIp((HttpServletRequest) asyncContext.getRequest()),"polling", clientMd5Map.size(), probeRequestSize);
                    //比较数据的 MD5 值，判断是否发生变更
                    List<String> changedGroups = MD5Util.compareMd5((HttpServletRequest) asyncContext.getRequest(), (HttpServletResponse) asyncContext.getResponse(), clientMd5Map);
                    if (changedGroups.size() > 0) {
                        //并将变更的结果通过response返回给客户端
                        sendResponse(changedGroups);
                    } else {
                        sendResponse(null);
                    }
                } else {
                    LogUtil.CLIENT_LOG.info("{}|{}|{}|{}|{}|{}", (System.currentTimeMillis() - createTime), "timeout", RequestUtil.getRemoteIp((HttpServletRequest) asyncContext.getRequest()),"polling", clientMd5Map.size(), probeRequestSize);
                    sendResponse(null);
                }
            } catch (Throwable t) {
                LogUtil.DEFAULT_LOG.error("long polling error:" + t.getMessage(), t.getCause());
            }
        }
    }, timeoutTime, TimeUnit.MILLISECONDS);
    allSubs.add(this);
}

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() 方法；
点击查看 github 上的 1.3.1 版本源码；

@Service
public class LongPollingService extends AbstractEventListener {

   //省略其他代码

    @Override
    public void onEvent(Event event) {
        if (isFixedPolling()) {
            // Ignore.
        } else {
            if (event instanceof LocalDataChangeEvent) {
                LocalDataChangeEvent evt = (LocalDataChangeEvent) event;
                //【点进去 2.4.2】通过线程池执行 DataChangeTask 任务
                scheduler.execute(new DataChangeTask(evt.groupKey, evt.isBeta, evt.betaIps));
            }
        }
    }
}

而在 1.3.2 版本之后，通过构造订阅者实现，点击查看 github 上的 1.3.2 版本源码；

NotifyCenter.registerSubscriber(new Subscriber() {

    @Override
    public void onEvent(Event event) {
        if (isFixedPolling()) {
            // Ignore.
        } else {
            if (event instanceof LocalDataChangeEvent) {
                LocalDataChangeEvent evt = (LocalDataChangeEvent) event;
                //【点进去 2.4.2】通过线程池执行 DataChangeTask 任务
                ConfigExecutor.executeLongPolling(new DataChangeTask(evt.groupKey, evt.isBeta, evt.betaIps));
            }
        }
    }
    @Override
    public Class<? extends Event> subscribeType() {
        return LocalDataChangeEvent.class;
    }
});

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

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

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

@Override
public void run() {
    try {
        ConfigCacheService.getContentBetaMd5(groupKey);
        //遍历 allSubs 中的客户端长轮询请求
        for (Iterator<ClientLongPolling> iter = allSubs.iterator(); iter.hasNext(); ) {
            ClientLongPolling clientSub = iter.next();
            //比较每一个客户端长轮询请求携带的groupKey，如果服务端变更的配置和客户端请求关注的配置一致，则直接返回
            if (clientSub.clientMd5Map.containsKey(groupKey)) {
                //如果 beta 发布且不在 beta 列表，则直接跳过
                if (isBeta && !CollectionUtils.contains(betaIps, clientSub.ip)) {
                    continue;
                }
                //如果 tag 发布且不在 tag 列表，则直接跳过
                if (StringUtils.isNotBlank(tag) && !tag.equals(clientSub.tag)) {
                    continue;
                }
                getRetainIps().put(clientSub.ip, System.currentTimeMillis());
                iter.remove(); //删除订阅关系
                LogUtil.CLIENT_LOG.info("{}|{}|{}|{}|{}|{}|{}", (System.currentTimeMillis() - changeTime), "in-advance", RequestUtil.getRemoteIp((HttpServletRequest) clientSub.asyncContext.getRequest()), "polling", clientSub.clientMd5Map.size(), clientSub.probeRequestSize, groupKey);
                //发送响应
                clientSub.sendResponse(Arrays.asList(groupKey));
            }
        }
    } catch (Throwable t) {
        LogUtil.DEFAULT_LOG.error("data change error: {}", ExceptionUtil.getStackTrace(t));
    }
}

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

vlambda博客
学习文章列表

Nacos 长轮询定时机制的源码分析

前言

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 执行长轮询请求 ConfigServletInner.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服务器报错：nacos failed to req API after all servers tried

详解nacos注册中心服务注册流程

docker发布微服务读取nacos配置时出错

vlambda博客 学习文章列表

Nacos 长轮询定时机制的源码分析

前言

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 执行长轮询请求 ConfigServletInner.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服务器报错：nacos failed to req API after all servers tried

详解nacos注册中心服务注册流程

docker发布微服务读取nacos配置时出错

vlambda博客
学习文章列表