2023-09-15  阅读(194)
原文作者:王伟王胖胖 原文地址: https://blog.csdn.net/wangwei19871103/article/details/105762802

ClientWorker的checkConfigInfo

先得再说下这个,这个方法是每10毫秒执行一次,但是并不是会一直去执行新的LongPollingRunnable任务,是根据监听器的数量决定要不要再启动一个LongPollingRunnable,每个LongPollingRunnable默认可以负责3000个监听器的轮询。所以一般就只是开启了一个,因为是Math.ceil向上取整,最开始就会开启。

        public void checkConfigInfo() {
            // 分任务 根据监听器数量开启任务,默认一个任务3000个监听器
            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的run

这个就是整个轮询获取配置信息的过程,首先会遍历所有的CacheData,找出是当前任务的加入到一个集合里,这里如果多的话会不会效率比较低啊,可能想办法直接放在任务中比较好吧。然后如果是本次任务的就先检查本地配置,如果有改变的话就要通知监听器。然后去请求服务器获取配置信息,如果是有在初始化的CacheData,那服务器就会立即返回,否则会被挂起,这个原因就是为了不进行频繁的空轮询,又能实现动态配置,只要在挂起的时间段内有改变,就可以理解响应给客户端。获取完之后再检查有没改变,有的话也要通知,然后继续调度当前任务。

     @Override
            public void run() {
    
                List<CacheData> cacheDatas = new ArrayList<CacheData>();
                //是否是在初始化的CacheData,会影响服务器是否挂起或者立即返回
                List<String> inInitializingCacheList = new ArrayList<String>();
                try {
                    // check failover config
                    for (CacheData cacheData : cacheMap.get().values()) {
                        if (cacheData.getTaskId() == taskId) {//属于当前长轮询任务的
                            cacheDatas.add(cacheData);
                            try {
                                checkLocalConfig(cacheData);
                                if (cacheData.isUseLocalConfigInfo()) {//用本地配置
                                    cacheData.checkListenerMd5();//有改变的话会通知
                                }
                            } catch (Exception e) {
                                LOGGER.error("get local config info error", e);
                            }
                        }
                    }
                    //获取有变化的配置列表dataid+group,访问的url是/listener
                    // check server config
                    List<String> changedGroupKeys = checkUpdateDataIds(cacheDatas, inInitializingCacheList);
                    LOGGER.info("get changedGroupKeys:" + changedGroupKeys);
    
                    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 {//有更新的就获取一次配置
                            String[] ct = getServerConfig(dataId, group, tenant, 3000L);
                            CacheData cache = cacheMap.get().get(GroupKey.getKeyTenant(dataId, group, tenant));
                            cache.setContent(ct[0]);//设置配置内容
                            if (null != ct[1]) {
                                cache.setType(ct[1]);//设置配置类型
                            }
                           ...
                        } 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) {
    
                    LOGGER.error("longPolling error : ", e);
                    executorService.schedule(this, taskPenaltyTime, TimeUnit.MILLISECONDS);
                }
            }
        }

checkLocalConfig检查本地配置

如果不没用本地文件,但是文件存在或者如果用本地文件,本地文件并有更新,就获取内容,设置版本,设置setUseLocalConfigInfo(true)通知标记返回。如果用本地文件,但是不存在,不用设置通知,直接返回。

     //检查配置
        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;
            }
            // 用本地,存在,版本有变更,更新
            // 有变更
            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));
            }
        }

checkUpdateDataIds获取有改变的配置文件

这个是把所有的CacheData的相关信息都连起来,一次性批量请求。但是其中有个比较重要的就是inInitializingCacheList,这个表示里面是否有正在初始化的CacheData,如果有的话后面会设置一个标记,是的服务器不会挂起请求,会立即响应。 这里的响应只是告诉你哪些是有变化的,不会把内容给你,后面还得请求内容。

    List<String> checkUpdateDataIds(List<CacheData> cacheDatas, List<String> inInitializingCacheList) throws IOException {
            StringBuilder sb = new StringBuilder();//把配置信息都连起来,一次请求
            for (CacheData cacheData : cacheDatas) {
                if (!cacheData.isUseLocalConfigInfo()) {//不用本地的
                    sb.append(cacheData.dataId).append(WORD_SEPARATOR);
                    sb.append(cacheData.group).append(WORD_SEPARATOR);
                    if (StringUtils.isBlank(cacheData.tenant)) {
                        sb.append(cacheData.getMd5()).append(LINE_SEPARATOR);
                    } else {
                        sb.append(cacheData.getMd5()).append(WORD_SEPARATOR);
                        sb.append(cacheData.getTenant()).append(LINE_SEPARATOR);
                    }
                    if (cacheData.isInitializing()) {
                        // cacheData 首次出现在cacheMap中&首次check更新
                        inInitializingCacheList
                            .add(GroupKey.getKeyTenant(cacheData.dataId, cacheData.group, cacheData.tenant));
                    }
                }
            }
            boolean isInitializingCacheList = !inInitializingCacheList.isEmpty();//是否是初始化的获取标记
            return checkUpdateConfigStr(sb.toString(), isInitializingCacheList);
        }

checkUpdateDataIds服务器获取

这里就会根据isInitializingCacheList来设置一个标记,让服务器判断是否要挂起,请求的url/v1/cs/configs/listener,这里超时增加了,默认变成了45秒,就是为了应对挂起和检查配置文件变更。内部怎么请求的前面讲过了,就不多说了。这里根据情况还要设置服务器健康状态setHealthServer,然后拿到改变的配置文件结果解析后返回。

       List<String> checkUpdateConfigStr(String probeUpdateString, boolean isInitializingCacheList) throws IOException {
    
            List<String> params = new ArrayList<String>(2);
            params.add(Constants.PROBE_MODIFY_REQUEST);
            params.add(probeUpdateString);
    
            List<String> headers = new ArrayList<String>(2);
            headers.add("Long-Pulling-Timeout");
            headers.add("" + timeout);
    
    
            if (isInitializingCacheList) {//是初始化的会设置一个请求头标记
                headers.add("Long-Pulling-Timeout-No-Hangup");
                headers.add("true");
            }
    
            if (StringUtils.isBlank(probeUpdateString)) {
                return Collections.emptyList();
            }
    
            try {
              
                // 增加超时时间,防止被挂起,只有初始化的时候isInitializingCacheList=true不会挂起,应该是服务器看了请求头Long-Pulling-Timeout-No-Hangup才不会挂起
                long readTimeoutMs = timeout + (long) Math.round(timeout >> 1);//45秒
                HttpResult result = agent.httpPost(Constants.CONFIG_CONTROLLER_PATH + "/listener", headers, params,
                    agent.getEncode(), readTimeoutMs);
    
                if (HttpURLConnection.HTTP_OK == result.code) {
                    setHealthServer(true);
                    return parseUpdateDataIdResponse(result.content);
                } else {
                    setHealthServer(false);
                    LOGGER.error("[{}] [check-update] get changed dataId error, code: {}", agent.getName(), result.code);
                }
            } catch (IOException e) {
                setHealthServer(false);
                LOGGER.error("[" + agent.getName() + "] [check-update] get changed dataId exception", e);
                throw e;
            }
            return Collections.emptyList();
        }

下篇继续。

好了,今天就到这里了,希望对学习理解有帮助,大神看见勿喷,仅为自己的学习理解,能力有限,请多包涵

阅读全文
  • 点赞