Nacos配置中心原理
- 2019 年 12 月 30 日
- 筆記
使用示例
@Before public void init() throws NacosException { Properties properties = new Properties(); properties.put(PropertyKeyConst.SERVER_ADDR, "127.0.0.1:" + 8848); properties.put(PropertyKeyConst.CONFIG_LONG_POLL_TIMEOUT, "20000"); properties.put(PropertyKeyConst.CONFIG_RETRY_TIME, "3000"); properties.put(PropertyKeyConst.MAX_RETRY, "5"); configService = NacosFactory.createConfigService(properties); } @Test public void test() throws InterruptedException, NacosException { configService.addListener("test", "DEFAULT_GROUP", new Listener() { @Override public Executor getExecutor() { return null; } @Override public void receiveConfigInfo(String configInfo) { System.out.println(configInfo); } }); TimeUnit.SECONDS.sleep(100); }
NacosConfigService
这是Nacos给客户端提供的API，可以通过该API：增、删、盖、查配置信息，还可以通过该API给配置添加Listener
创建ConfigService
// NacosFactory#createConfigService public static ConfigService createConfigService(Properties properties) throws NacosException { return ConfigFactory.createConfigService(properties); } // ConfigFactory#createConfigService public static ConfigService createConfigService(Properties properties) throws NacosException { try { Class<?> driverImplClass = Class.forName("com.alibaba.nacos.client.config.NacosConfigService"); Constructor constructor = driverImplClass.getConstructor(Properties.class); // 反射创建对象 ConfigService vendorImpl = (ConfigService) constructor.newInstance(properties); return vendorImpl; } catch (Throwable e) { throw new NacosException(NacosException.CLIENT_INVALID_PARAM, e); } }
通过反射的机制创建了一个NacosConfigService实例，它的构造函数如下
public NacosConfigService(Properties properties) throws NacosException { String encodeTmp = properties.getProperty(PropertyKeyConst.ENCODE); if (StringUtils.isBlank(encodeTmp)) { encode = Constants.ENCODE; } else { encode = encodeTmp.trim(); } initNamespace(properties); agent = new MetricsHttpAgent(new ServerHttpAgent(properties)); agent.start(); worker = new ClientWorker(agent, configFilterChainManager, properties); }
-
MetricsHttpAgent主要是用来记录一些Metric信息，内部持有一个ServerHttpAgent对象，所以主要逻辑还是关注ServerHttpAgent； -
ClientWorker是一个非常核心的类，里面封装了客户端从服务端主动PULL配置信息的关键逻辑，这个在下面重点介绍
ServerHttpAgent
public ServerHttpAgent(Properties properties) throws NacosException { serverListMgr = new ServerListManager(properties); init(properties); }
ServerHttpAgent内部主要就是封装了一个HTTP交互的通用方法，比如GET，PUT，DELETE等方法，在ServerHttpAgent构造函数中，主要做了两件事情：
- 创建
ServerListManager对象，这个对象的主要作用就是根据Properties解析出服务端的地址，然后维护在一个List<String>中 -
init方法中主要做了三件事情：初始化编码格式，如果没有就默认UTF-8；初始化accessKey和secretKey，这个应该是用来验证客户端的身份的；初始化重试次数，如果没传默认为3
在ServerListManager对象中，还有一个比较重要的属性isFixed，这个属性的主要作用就是标识服务器的地址信息是不是固定的，比如，如果服务器的地址是通过Properties传入，那isFixed的值为true
为什么要介绍这个属性呢，因为在NacosConfigService的构造函数中，调用了MetricsHttpAgent#start方法，而这个方法的内部调用链如下
MetricsHttpAgent#start => ServerHttpAgent#start => ServerListManager#start
ServerListManager#start方法主要做了什么事情呢？如果isFixed的值为true，就直接返回。否则先执行initServerlistRetryTimes次GetServerListTask#run方法获取，该方法用于更新服务器地址信息，如果执行initServerlistRetryTimes次 之后还是没有获取到服务器地址列表信息，则直接抛出异常，否则开启一个定时任务，每30s更新一次服务器地址列表信息。
至此，在NacosConfigService构造函数中，只剩下ClientWorker相关的逻辑了
ClientWorker
前面已经提到过，ClientWorker是一个非常核心的类，里面封装了客户端从服务端主动pull配置信息的关键逻辑。注意这里很明确指出了ClientWorker是通过pull模式从服务端获取配置信息的，而我们在使用的时候通常会给它添加Listener，这 会让我们以为它是push模式，这是一点需要注意的地方。
而它实现的pull模式的关键点在于两个线程池，这两个线程池在ClientWorker的构造函数中初始化。其中有一个线程池executor里面只有一个线程，它的作用就是让另一个线程池启动。
public ClientWorker(final HttpAgent agent, final ConfigFilterChainManager configFilterChainManager, final Properties properties) { init(properties); executor = Executors.newScheduledThreadPool(1, new ThreadFactory() { @Override 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 = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors(), new ThreadFactory() { @Override 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.scheduleWithFixedDelay(new Runnable() { @Override public void run() { try { checkConfigInfo(); } catch (Throwable e) { LOGGER.error("[" + agent.getName() + "] [sub-check] rotate check error", e); } } }, 1L, 10L, TimeUnit.MILLISECONDS); }
-
init(properties)方法主要就是初始化一些timeout的参数 -
executor线程池里面只有一个线程，它的唯一作用就是让另一个线程池开始执行，每10s中执行一次 -
executorService线程池用于更新配置信息，核心任务LongPollingRunnable
ClientWorker#checkConfigInfo方法主要作用是更新配置信息，目前已经获取到的配置信息会缓存到一个Map<String, CacheData>中，然后对map中的数据分批次，一个批次默认是3000条数据，每个批次的数据对应一个线程负责更新，如下：
public void checkConfigInfo() { // 分任务 int listenerSize = cacheMap.get().size(); // 向上取整为批数 int longingTaskCount = (int) Math.ceil(listenerSize / ParamUtil.getPerTaskConfigSize()); if (longingTaskCount > currentLongingTaskCount) { for (int i = (int) currentLongingTaskCount; i < longingTaskCount; i++) { // 要判断任务是否在执行 这块需要好好想想。 任务列表现在是无序的。变化过程可能有问题 executorService.execute(new LongPollingRunnable(i)); } currentLongingTaskCount = longingTaskCount; } }
从LongPollingRunnable的名字可以看出，客户端主要是通过长轮询的方式去更新配置信息
LongPollingRunnable
public void run() { // 本批次号的数据 List<CacheData> cacheDatas = new ArrayList<CacheData>(); // 本批次号中 isInitializing=true 的数据，CacheData首次出现在Map中并且是首次check更新时，isInitializing的值才为true List<String> inInitializingCacheList = new ArrayList<String>(); try { // check failover config for (CacheData cacheData : cacheMap.get().values()) { if (cacheData.getTaskId() == taskId) { cacheDatas.add(cacheData); try { // 涉及到FailoverFile checkLocalConfig(cacheData); // 如果有更新，需要更新Listener的MD5值，并执行Listener逻辑 if (cacheData.isUseLocalConfigInfo()) { cacheData.checkListenerMd5(); } } catch (Exception e) { LOGGER.error("get local config info error", e); } } } // check server config // 找到isUseLocalConfig=false的数据，然后将每个符合条件的CacheData的 `group + dataId + tenant` 拼成一个字符串传给服务端校验，然后服务端会返回一个需要更新的`List<String>`，该列表里面的每个元素代表一个CacheData的key List<String> changedGroupKeys = checkUpdateDataIds(cacheDatas, inInitializingCacheList); for (String groupKey : changedGroupKeys) { String[] key = GroupKey.parseKey(groupKey); String dataId = key[0]; String group = key[1]; String tenant = null; if (key.length == 3) { tenant = key[2]; } try { // 根据需要更新的key列表，从服务端获取配置信息 String content = getServerConfig(dataId, group, tenant, 3000L); CacheData cache = cacheMap.get().get(GroupKey.getKeyTenant(dataId, group, tenant)); cache.setContent(content); } catch (NacosException ioe) { } } // 这部分代码是在没看懂 for (CacheData cacheData : cacheDatas) { if (!cacheData.isInitializing() || inInitializingCacheList.contains(GroupKey.getKeyTenant(cacheData.dataId, cacheData.group, cacheData.tenant))) { cacheData.checkListenerMd5(); cacheData.setInitializing(false); } } inInitializingCacheList.clear(); // 重新提交自己 executorService.execute(this); } catch (Throwable e) { // If the rotation training task is abnormal, the next execution time of the task will be punished LOGGER.error("longPolling error : ", e); executorService.schedule(this, taskPenaltyTime, TimeUnit.MILLISECONDS); } }
LongPollingRunnable#run方法比较长：
- 从map中取出批次号符与该线程对应批次号相同的数据，因为每条配置信息
CacheData都维护了一个批次号信息，然后每个LongPollingRunnable也对应一个批次号信息，只需要负责更新批次号相同的数据，不同的由别的LongPollingRunnable去更新，这部分据放在变量cacheDatas中 - 如果客户端和服务端部署在同一个节点，此时客户端可以直接从本地文件中获取配置信息，避免远程交互，设置这部分数据的
isUseLocalConfig=true，并更新CacheData的值，同时更新CacheData的MD5值 - 如果步骤2可以直接根据本地文件更新值
isUseLocalConfig==true，此时执行cacheData#checkListenerMd5方法，该方法会拿CacheData的MD5值和Listener的MD5值对比，如果不一样就更新Listener的MD5值并回调Listener的相关方法 - 排除
isUseLocalConfig=true的配置信息，然后将每个符合条件的配置信息的group + dataId +tenant拼成一个字符串传给服务端校验，然后服务端会返回一个需要更新的List<String>，该列表里面的每个元素代表一个CacheData的key - 根据上一步骤返回的key列表，从服务端拉取配置信息，然后更新
CacheData的值 -
cacheData.isInitializing()代表该条配置信息首次出现在map中并且是首次更新，inInitializingCacheList代表本批次中isUseLocalConfig==false && isInitializing == true - 最后那个遍历不太清楚啥意思，
isInitializing == false || isInitializing == true？
配置监听器
上面提到ClientWorker中有一个Map，那里面的数据从哪里来呢？为什么要存这部分数据呢？答案是在为CacheData添加Listener的时候，会初始化一条CacheData数据，并添加到Map中
添加监听
ConfigService configService = NacosFactory.createConfigService(properties); configService.addListener("test", "DEFAULT_GROUP", new Listener() { @Override public Executor getExecutor() { return null; } @Override public void receiveConfigInfo(String configInfo) { System.out.println(configInfo); } });
源码
// NacosConfigService#addListener public void addListener(String dataId, String group, Listener listener) throws NacosException { worker.addTenantListeners(dataId, group, Arrays.asList(listener)); } // ClientWorker#addTenantListeners public void addTenantListeners(String dataId, String group, List<? extends Listener> listeners) throws NacosException { group = null2defaultGroup(group); String tenant = agent.getTenant(); CacheData cache = addCacheDataIfAbsent(dataId, group, tenant); for (Listener listener : listeners) { cache.addListener(listener); } }
内部还是委托给ClientWorker来实现，添加监时主要做了以下几件事情
- 将
dataId group tenant封装成key，从ClientWorker的Map缓存中获取CacheData1.1. 如果从缓存中拿到了数据，将CacheData的isInitializing属性设置为true 1.2. 如果从缓存中没拿到数据，则创建一个CacheData，同时判断是否开启远程同步,即enableRemoteSyncConfig属性的值是否为true。如果开启了，则从服务端获取该配置的值，并设置到CacheData中 - 将最新的
CacheData设置到缓存中 - 将
Listener添加到CacheData的CopyOnWriteArrayList<ManagerListenerWrap> listeners列表中
触发监听
其实在上面已经提到过，当配置有更新时，会触发Listener的回调逻辑，这部分逻辑在CacheData#checkListenerMd5方法中
void checkListenerMd5() { // `ManagerListenerWrap`只是一个包装类，里面维护了`Listener`和对应`CacheData`的MD5值 for (ManagerListenerWrap wrap : listeners) { // 如果不一样，说明配置有变更，此时更新ManagerListenerWrap的MD5值，并执行Listener的回调 if (!md5.equals(wrap.lastCallMd5)) { safeNotifyListener(dataId, group, content, md5, wrap); } } } private void safeNotifyListener(final String dataId, final String group, final String content, final String md5, final ManagerListenerWrap listenerWrap) { final Listener listener = listenerWrap.listener; Runnable job = new Runnable() { @Override public void run() { try { ConfigResponse cr = new ConfigResponse(); cr.setDataId(dataId); cr.setGroup(group); cr.setContent(content); configFilterChainManager.doFilter(null, cr); String contentTmp = cr.getContent(); listener.receiveConfigInfo(contentTmp); listenerWrap.lastCallMd5 = md5; } finally { Thread.currentThread().setContextClassLoader(myClassLoader); } } }; final long startNotify = System.currentTimeMillis(); try { if (null != listener.getExecutor()) { // 如果配置了线程池，交给线程池执行 listener.getExecutor().execute(job); } else { job.run(); } } catch (Throwable t) { } }
-
ManagerListenerWrap只是一个包装类，里面维护了Listener和对应CacheData的MD5值 - 判断
ManagerListenerWrap和当前CacheData的MD5值是否相同，如果不一样，说明配置有变更，此时需要更新ManagerListenerWrap的MD5值，并执行Listener的回调 - 更新
ManagerListenerWrap的MD5值和执行Listener的回调的逻辑都在safeNotifyListener方法中，同时会判断是否为Listener配置了线程池如，没有就直接执行，有就交给线程池执行
创建配置
ConfigService configService = NacosFactory.createConfigService(properties); configService.publishConfig("one", "LSZ", "大美女");
源码
private boolean publishConfigInner(String tenant, String dataId, String group, String tag, String appName, String betaIps, String content) throws NacosException { ...... 一大坨构造参数的代码 ...... HttpResult result = null; try { // 推送到服务端 result = agent.httpPost(url, headers, params, encode, POST_TIMEOUT); } catch (IOException ioe) { return false; } ...... }
获取配置
ConfigService configService = NacosFactory.createConfigService(properties); configService.getConfig("testlsz", "lszgroup", 1000000);
源码
private String getConfigInner(String tenant, String dataId, String group, long timeoutMs) throws NacosException { ...... // 优先使用本地配置 String content = LocalConfigInfoProcessor.getFailover(agent.getName(), dataId, group, tenant); if (content != null) { return content; } try { // 本地没有从服务器获取 content = worker.getServerConfig(dataId, group, tenant, timeoutMs); cr.setContent(content); configFilterChainManager.doFilter(null, cr); content = cr.getContent(); return content; } catch (NacosException ioe) { } ...... }
删除配置
ConfigService configService = NacosFactory.createConfigService(properties); configService.removeConfig("testlsz", "lszgroup");
源码
private boolean removeConfigInner(String tenant, String dataId, String group, String tag) throws NacosException { ...... 一大坨构造参数的代码 ...... HttpResult result = null; try { // 直接发送http请求 result = agent.httpDelete(url, null, params, encode, POST_TIMEOUT); } catch (IOException ioe) { LOGGER.warn("[remove] error, " + dataId + ", " + group + ", " + tenant + ", msg: " + ioe.toString()); return false; } ...... }
本地文件
在Naco中涉及到两个文件，FailoverFile和SnapshotFile
-
FailoverFile为容灾文件，当本地和数据库里面数据不一致的时候会去使用，一般不会用； -
SnapshotFile为配置的快照，当获取不到服务器上的配置的时候，会读取本地快照；FailoverFile在客户端不会自动生成，它是在服务端生成的，当更新了一条配置之后，就会反应到这个文件中。所以如果想在客户端使用到这个功能，需要手工将文件添加到客户端，然后客户端就不会去读取服务端的配置了，也许某些场景下可以用到
SnapshotFile
SnapshotFile文件位置：userhomenacosconfigfixed-127.0.0.1_8848_nacossnapshot{group}{dataId}
当客户端从服务端获取配置之后，会将该信息写入快照文件中，核心代码就在ClientWorker#getServerConfig中
public String getServerConfig(String dataId, String group, String tenant, long readTimeout) throws NacosException { HttpResult result = null; try { // 从服务端获取配置信息 result = agent.httpGet(Constants.CONFIG_CONTROLLER_PATH, null, params, agent.getEncode(), readTimeout); } catch (IOException e) { throw new NacosException(NacosException.SERVER_ERROR, e); } switch (result.code) { case HttpURLConnection.HTTP_OK: // 将配置更新到本地文件 LocalConfigInfoProcessor.saveSnapshot(agent.getName(), dataId, group, tenant, result.content); return result.content; case HttpURLConnection.HTTP_NOT_FOUND: // 根据 dataId、group、tenant 将本地的文件删除 LocalConfigInfoProcessor.saveSnapshot(agent.getName(), dataId, group, tenant, null); return null; case HttpURLConnection.HTTP_CONFLICT: { throw new NacosException(NacosException.CONFLICT,"data being modified, dataId=" + dataId + ", group=" + group + ",tenant=" + tenant); } case HttpURLConnection.HTTP_FORBIDDEN: { LOGGER.error("[{}] [sub-server-error] no right, dataId={}, group={}, tenant={}", agent.getName(), dataId, group, tenant); throw new NacosException(result.code, result.content); } default: { LOGGER.error("[{}] [sub-server-error] dataId={}, group={}, tenant={}, code={}", agent.getName(), dataId, group, tenant, result.code); throw new NacosException(result.code, "http error, code=" + result.code + ",dataId=" + dataId + ",group=" + group + ",tenant=" + tenant); } } }
FailoverFile
FailoverFile文件位置：userhomenacosconfigfixed-127.0.0.1_8848_nacosdataconfig-data{group}{dataId}
对FailoverFile文件的判断主要是在ClientWorker#checkLocalConfig方法，看了几篇文章，都说这个方法的校验是为了在服务端挂了的时候，可以直接从客户端获取文件，这明显是不对的。刚刚在上面也提到过，FailoverFile的主要作用是容灾，而且这个文件在客户端不会自动生成，想要使用这个功能必须手动添加
private void checkLocalConfig(CacheData cacheData) { final String dataId = cacheData.dataId; final String group = cacheData.group; final String tenant = cacheData.tenant; // 注意这个，容灾文件，这个文件在客户端是不会自动生成的 File path = LocalConfigInfoProcessor.getFailoverFile(agent.getName(), dataId, group, tenant); // 没有 -> 有 if (!cacheData.isUseLocalConfigInfo() && path.exists()) { String content = LocalConfigInfoProcessor.getFailover(agent.getName(), dataId, group, tenant); String md5 = MD5.getInstance().getMD5String(content); cacheData.setUseLocalConfigInfo(true); cacheData.setLocalConfigInfoVersion(path.lastModified()); cacheData.setContent(content); LOGGER.warn("[{}] [failover-change] failover file created. dataId={}, group={}, tenant={}, md5={}, content={}", agent.getName(), dataId, group, tenant, md5, ContentUtils.truncateContent(content)); return; } // 有 -> 没有。不通知业务监听器，从server拿到配置后通知。 if (cacheData.isUseLocalConfigInfo() && !path.exists()) { cacheData.setUseLocalConfigInfo(false); LOGGER.warn("[{}] [failover-change] failover file deleted. dataId={}, group={}, tenant={}", agent.getName(), dataId, group, tenant); return; } // 有变更 /** * isUseLocalConfig=true && && path.exists() && cacheData.getLocalConfigInfoVersion() != path.lastModified() * 说明配置有更新，所以此时需要更新 CacheData */ if (cacheData.isUseLocalConfigInfo() && path.exists() && cacheData.getLocalConfigInfoVersion() != path.lastModified()) { String content = LocalConfigInfoProcessor.getFailover(agent.getName(), dataId, group, tenant); String md5 = MD5.getInstance().getMD5String(content); cacheData.setUseLocalConfigInfo(true); cacheData.setLocalConfigInfoVersion(path.lastModified()); cacheData.setContent(content); LOGGER.warn("[{}] [failover-change] failover file changed. dataId={}, group={}, tenant={}, md5={}, content={}", agent.getName(), dataId, group, tenant, md5, ContentUtils.truncateContent(content)); } }
有关于获取配置的优化
从上面已经知道，有一个线程池会不停的执行LongPollingRunnable#run方法，这个方法主要作用就是从服务端拉取最新的配置信息，如果直接按照正常的做法来做，直接根据dataId group tenant去拉取就好了，如果每次都直接去服务器来配置信息，但这样会有一些性能问题：
- 配置信息变动的可能性很小，如果每次都需要全量去拉取，拉取的信息基本都是一样的，这很浪费资源；
- 如果从服务端拉取数据的频率太高，会太耗性能；如果拉取的频率太低，数据发生变更之后客户端响应不及时；
针对上面几个问题，Nacos做了以下几个优化
- 只拉取改动过的配置信息：客户端先通过一个HTTP请求发送一个
key列表给服务端，服务端返回发生了变更的Key列表，大部分时候，这可以过滤掉绝大部分没有配置项； - 通过HTTP长轮询减少少客户端和服务端的交互频率，但这必然要面对一个数据响应不实时问问题，怎么解决？
配置实时更新
先推荐一篇文章：Nacos配置实时更新原理分析 这篇文章已经写的非常详细了，不过那篇文章有点长，这里总结一下，为了自己以后看的时候方便。
通过HTTP长轮询较少客户端和服务端的交互频率，但这必然要面对一个数据响应不实时问问题，怎么解决？
在客户端向服务端拉取配置信息之前，需要先向服务端发送一个配置Key列表，然后服务端返回一个发生了变更的配置Key列表
ClientWorker#checkUpdateConfigStr
// timeout默认30s超时 HttpResult result = agent.httpPost(Constants.CONFIG_CONTROLLER_PATH + "/listener", headers, params, agent.getEncode(), timeout);
请求的服务端地址: /v1/cs/configs/listener
ConfigController#listener
@RequestMapping(value = "/listener", method = RequestMethod.POST) public void listener(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException { .... // do long-polling inner.doPollingConfig(request, response, clientMd5Map, probeModify.length()); }
// ConfigServletInner#doPollingConfig ，暂且只关注长轮询
// 长轮询 if (LongPollingService.isSupportLongPolling(request)) { longPollingService.addLongPollingClient(request, response, clientMd5Map, probeRequestSize); return HttpServletResponse.SC_OK + ""; }
LongPollingService#addLongPollingClient
public void addLongPollingClient(HttpServletRequest req, HttpServletResponse rsp, Map<String, String> clientMd5Map, int probeRequestSize) { int delayTime = SwitchService.getSwitchInteger(SwitchService.FIXED_DELAY_TIME, 500); /** * 提前500ms返回响应，为避免客户端超时 add delay time for LoadBalance */ long timeout = Math.max(10000, Long.parseLong(str) - delayTime); String ip = RequestUtil.getRemoteIp(req); // 一定要由HTTP线程调用，否则离开后容器会立即发送响应， 开启Servlet异步支持 final AsyncContext asyncContext = req.startAsync(); // AsyncContext.setTimeout()的超时时间不准，所以只能自己控制 asyncContext.setTimeout(0L); // 向线程池提交了一个任务，所以核心逻辑在 ClientLongPolling 中， scheduler.execute(new ClientLongPolling(asyncContext, clientMd5Map, ip, probeRequestSize, timeout, appName, tag)); }
ClientLongPolling#run
public void run() { asyncTimeoutFuture = scheduler.schedule(new Runnable() { @Override public void run() { try { getRetainIps().put(ClientLongPolling.this.ip, System.currentTimeMillis()); /** * 删除订阅关系 */ allSubs.remove(ClientLongPolling.this); if (isFixedPolling()) { // 根据客户端传过来的key列表，找到该发生了变更的配置的key列表 List<String> changedGroups = MD5Util.compareMd5((HttpServletRequest)asyncContext.getRequest(), (HttpServletResponse)asyncContext.getResponse(), clientMd5Map); if (changedGroups.size() > 0) { // 返回给客户端 sendResponse(changedGroups); } else { // 没有就返回空 sendResponse(null); } } else { sendResponse(null); } } catch (Throwable t) { LogUtil.defaultLog.error("long polling error:" + t.getMessage(), t.getCause()); } } }, timeoutTime, TimeUnit.MILLISECONDS); allSubs.add(this); }
- 在run方法里面又创建了一个任务，延迟 29.5s 后执行
- 将
ClientLongPolling添加到allSubs中 - 延迟时间到了之后，执行任务，先将
ClientLongPolling从allSubs中移除，然后通过AsyncContext将结果写回客户端
allSubs数据结构如下，可以把allSubs当作是一个订阅者列表，当配置发生变成的时候，会发布一个事件，然后这些订阅者会得到相应，然后再执行相应的功能
Queue<ClientLongPolling> allSubs = new ConcurrentLinkedQueue<ClientLongPolling>();
修改配置
延迟 29.5s 执行，要是在这段时间内，数据发生了变更怎么办，难道要客户端 29.5s 之后才知道吗？肯定不是的。可以看看当数据发生变更时，会涉及到什么接口
ConfigController#publishConfig
请求的服务端地址: /v1/cs/configs
// 持久化 persistService.insertOrUpdate(srcIp, srcUser, configInfo, time, configAdvanceInfo, false); // 触发事件 EventDispatcher.fireEvent(new ConfigDataChangeEvent(false, dataId, group, tenant, time.getTime()));
EventDispatcher#fireEvent
static public void fireEvent(Event event) { for (AbstractEventListener listener : getEntry(event.getClass()).listeners) { try { listener.onEvent(event); } catch (Exception e) { log.error(e.toString(), e); } } } static public abstract class AbstractEventListener { public AbstractEventListener() { EventDispatcher.addEventListener(this); } } // EventDispatcher#addEventListener static public void addEventListener(AbstractEventListener listener) { for (Class<? extends Event> type : listener.interest()) { getEntry(type).listeners.addIfAbsent(listener); } }
注意观察AbstractEventListener的构造函数，在其构造函数中涉及到Listener的注册过程，而AbstractEventListener有以下几个子类：
AsyncNotifyService -> ConfigDataChangeEvent LongPollingService -> LocalDataChangeEvent MockListener 计数用的
刚刚看上面发布的是一个ConfigDataChangeEvent事件，所以会先执行AsyncNotifyService#onEvent方法。该方法中会先获取服务端所有IP列表，依次通过Http通知对象/v1/cs/communication/dataChange?dataId=xx&group=xx，接收到请求后， 会dump出来所有config信息，同时回调LocalDataChangeEvent事件，然后执行LongPollingService#onEvent方法。
LongPollingService#onEvent
public void onEvent(Event event) { if (isFixedPolling()) { } else { if (event instanceof LocalDataChangeEvent) { LocalDataChangeEvent evt = (LocalDataChangeEvent)event; scheduler.execute(new DataChangeTask(evt.groupKey, evt.isBeta, evt.betaIps)); } } }
提交了一个任务，关键逻辑在DataChangeTask#run方法中
public void run() { try { ConfigService.getContentBetaMd5(groupKey); for (Iterator<ClientLongPolling> iter = allSubs.iterator(); iter.hasNext(); ) { // ClientLongPolling 作为订阅者，配置类似于Topic，更新配置就是发布了一个事件 ClientLongPolling clientSub = iter.next(); // 找到订阅了 当前发生了变更配置项 的ClientLongPolling if (clientSub.clientMd5Map.containsKey(groupKey)) { getRetainIps().put(clientSub.ip, System.currentTimeMillis()); // 删除订阅关系 iter.remove(); // 向客户端回写数据 clientSub.sendResponse(Arrays.asList(groupKey)); } } } catch (Throwable t) { LogUtil.defaultLog.error("data change error:" + t.getMessage(), t.getCause()); } }
- 找到订阅了 当前发生了变更配置项 的ClientLongPolling
- 向客户端回写数据
这里会有一个问题，如果DataChangeTask任务完成了向客户端写数据，此时ClientLongPolling中的调度任务又开始执行了怎么办呢？这时任务都被移除了，肯定会报错啊
很简单，只要在进行"推送"操作之前，先将原来等待执行的调度任务取消掉就可以了，如下：
void sendResponse(List<String> changedGroups) { /** * 取消超时任务 */ if (null != asyncTimeoutFuture) { asyncTimeoutFuture.cancel(false); } generateResponse(changedGroups); }
这样，就达到了一个类似于数据"推送"的效果，如果一直没有更新，客户端等待时间接近 30s，如果在等待期间有数据发生变更，几乎可以实时的返回给客户端
