锐客网

  1. 首页 > 生活百科 > it技术 > >

微服务架构 | *2.5 Nacos 长轮询定时机制的源码分析

IT知识微服务架构服务架构

一箫一剑平生意，负尽狂名十五年。这篇文章主要讲述微服务架构 | *2.5 Nacos 长轮询定时机制的源码分析相关的知识，希望能为你提供帮助。
@[TOC](微服务架构 | *2.5 Nacos 长轮询定时机制的源码分析)
前言参考资料：
《Spring Microservices in Action》
《Spring Cloud Alibaba 微服务原理与实战》
《B站 尚硅谷 SpringCloud 框架开发教程 周阳》
为方便理解与表达，这里把 Nacos 控制台和 Nacos 注册中心称为 Nacos 服务器（就是 web 界面那个），我们编写的业务服务称为 Nacso 客户端；
由于篇幅有限，这里将源码分析分为上下两篇，其中上篇讲获取配置与事件订阅机制，下篇讲长轮询定时机制；在《微服务架构 | 2.2 Alibaba Nacos 的统一配置管理》中提到一张 Nacos 动态监听的长轮询机制原理图，本篇将围绕这张图剖析长轮询定时机制的原理；

微服务架构 | *2.5 Nacos 长轮询定时机制的源码分析

文章图片

上篇《微服务架构 | *2.4 Nacos 配置中心的源码分析（获取配置与事件订阅机制）》中的 1.1 提到，ConfigService 是 Nacos 客户端提供的用于访问实现配置中心基本操作的类，我们将从 ConfigService 的实例化开始长轮询定时机制的源码之旅；
1. 客户端的长轮询定时机制
  • 我们从上一篇文章的这里开始【断点步入】；
  • NacosPropertySourceLocator.locate()；
    微服务架构 | *2.5 Nacos 长轮询定时机制的源码分析

    文章图片
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 类的设计如下：
    微服务架构 | *2.5 Nacos 长轮询定时机制的源码分析

    文章图片
  • ServerHttpAgent 类的设计如下：
    微服务架构 | *2.5 Nacos 长轮询定时机制的源码分析

    文章图片
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< String, CacheData> > 对象，用来存储监听变更的缓存集合，key 是根据 datalD/group/tenant（租户）拼接的值。Value 是对应的存储在 Nacos 服务器上的配置文件的内容；
    • 长轮询任务拆分：默认情况下，每个长轮询 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 = https://www.songbingjia.com/android/(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.5 Nacos 长轮询定时机制的源码分析

文章图片

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 对象，定位配置；
    • NacosFactory.createConfigService()：创建配置服务器；
    • ConfigFactory.createConfigService()：利用反射机制创建配置服务器；
    • NacosConfigService.NacosConfigService()：NacosConfigService 的构造方法；
      • MetricsHttpAgent.MetricsHttpAgent()：初始化 HttpAgent；
      • ClientWorker.ClientWorker()：初始化 ClientWorker；
      • Executors.newScheduledThreadPool()：创建 executor 线程池；
      • Executors.newScheduledThreadPool()：创建 executorService 线程池；
      • ClientWorker.checkConfigInfo()：使用 executor 线程池检查配置是否发生变化；
        • LongPollingRunnable.run()：运行长轮询定时线程；
        • ClientWorker.checkLocalConfig()：检查本地配置；
        • ClientWorker.checkUpdateDataIds()：检查服务端对应的配置是否发生变更；
          • ClientWorker.checkUpdateConfigStr()：检查服务端对应的配置是否发生变更；
          • MetricsHttpAgent.httpPost()：调用 /v1/cs/configs/listener 接口实现长轮询请求；
        • ClientWorker.getServerConfig()：读取变更配置
          • MetricsHttpAgent.httpGet()：调用 /v1/cs/configs 接口获取配置；
3.2 服务端的长轮询定时机制
  • ConfigController.listener()：服务器接收请求；
    • MD5Util.getClientMd5Map()：计算 MD5 值；
    • ConfigServletInner.doPollingConfig()：执行长轮询请求；
    • LongPollingService.addLongPollingClient()：长轮询的核心处理逻辑，提前 500ms 返回响应；
      • HttpServletRequest.getHeader()：获取客户端设置的请求超时时间；
      • MD5Util.compareMd5()：和服务端的数据进行 MD5 对比；
      • ConfigExecutor.executeLongPolling()：创建 ClientLongPolling 线程执行定时任务；
      • ClientLongPolling.run()：长轮询定时机制的实现逻辑；
        • ConfigExecutor.scheduleLongPolling()：启动定时任务，延时时间为 29.5s；
        • Map.put()：将 ClientLongPolling 实例本身添加到 allSubs 队列中；
        • Queue.remove()：把 ClientLongPolling 实例本身从 allSubs 队列中移除；
        • MD5Util.compareMd5()：比较数据的 MD5 值；
          • LongPollingService.sendResponse()：将变更的结果通过 response 返回给客户端；
3.3 Nacos 服务器配置变更的事件监听
  • Nacos 服务器上的配置发生变更后，发布一个 LocalDataChangeEvent 事件；
  • Subscriber.onEvent()：监听LocalDataChangeEvent 事件（1.3.2 版本后）；
    • ConfigExecutor.executeLongPolling()：通过线程池执行 DataChangeTask 任务；
    • DataChangeTask.run()：根据 groupKey 返回配置；
最后::: hljs-center
新人制作，如有错误，欢迎指出，感激不尽！
:::
::: hljs-center
欢迎关注公众号，会分享一些更日常的东西！
:::
::: hljs-center
如需转载，请标注出处！
:::
::: hljs-center
微服务架构 | *2.5 Nacos 长轮询定时机制的源码分析

文章图片

【微服务架构 | *2.5 Nacos 长轮询定时机制的源码分析】:::

    推荐阅读


    上一篇：普通开发者，如何成为Apache顶级项目的commiter （）

    下一篇：Centos7下设置MySql自动启动