Spring Cloud Nacos 配置变更感知

上一篇项目启动时执行locate方法中会生成ConfigService并启动线程池。

1、获取ConfigService

//com.alibaba.cloud.nacos.NacosConfigManager#getConfigService
public ConfigService getConfigService() {
   if (Objects.isNull(service)) {
       //创建ConfigService，和构造器中想删除的代码一样，这里是做了一个懒加载
      createConfigService(this.nacosConfigProperties);
   }
   return service;
}

static ConfigService createConfigService(
      NacosConfigProperties nacosConfigProperties) {
   if (Objects.isNull(service)) {
      synchronized (NacosConfigManager.class) {
         try {
            if (Objects.isNull(service)) {
               service = NacosFactory.createConfigService(
                     nacosConfigProperties.assembleConfigServiceProperties());
            }
         }
         catch (NacosException e) {
            log.error(e.getMessage());
            throw new NacosConnectionFailureException(
                  nacosConfigProperties.getServerAddr(), e.getMessage(), e);
         }
      }
   }
   return service;
}

//com.alibaba.nacos.api.NacosFactory#createConfigService(java.util.Properties)
public static ConfigService createConfigService(Properties properties) throws NacosException {
    return ConfigFactory.createConfigService(properties);
}

反射生成ConfigService

//com.alibaba.nacos.api.config.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 {
    ValidatorUtils.checkInitParam(properties);
    
    initNamespace(properties);
    //生成ConfigFilterChainManager，会根据配置加载实现了IConfigFilter接口的过滤器
    this.configFilterChainManager = new ConfigFilterChainManager(properties);
    //构建servers，会配置server的属性，包括组装url
    ServerListManager serverListManager = new ServerListManager(properties);
    //启动,因为在实际运行时，线程池不会启动，下面就不贴代码了
    serverListManager.start();
    //轮询配置变化的线程池
    this.worker = new ClientWorker(this.configFilterChainManager, serverListManager, properties);
    // will be deleted in 2.0 later versions
    agent = new ServerHttpAgent(serverListManager);
    
}

ClientWorker的构造器

public ClientWorker(final ConfigFilterChainManager configFilterChainManager, ServerListManager serverListManager,
        final Properties properties) throws NacosException {
    this.configFilterChainManager = configFilterChainManager;
    
    init(properties);
    //创建ConfigRpcTransportClient
    agent = new ConfigRpcTransportClient(properties, serverListManager);
    //定义线程池
    ScheduledExecutorService executorService = Executors
            .newScheduledThreadPool(ThreadUtils.getSuitableThreadCount(1), r -> {
                Thread t = new Thread(r);
                t.setName("com.alibaba.nacos.client.Worker");
                t.setDaemon(true);
                return t;
            });
    //将线程池赋值给agent
    agent.setExecutor(executorService);
    //启动agent
    agent.start();
    
}

//com.alibaba.nacos.client.config.impl.ConfigTransportClient#start
public void start() throws NacosException {
    //如果启动了安全验证，启动一个线程，5s定时
    if (securityProxy.isEnabled()) {
        //登录
        securityProxy.login(serverListManager.getServerUrls());
        
        this.executor.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                securityProxy.login(serverListManager.getServerUrls());
            }
        }, 0, this.securityInfoRefreshIntervalMills, TimeUnit.MILLISECONDS);
        
    }
    //启动轮询线程
    startInternal();
}

1.1、登录

//com.alibaba.nacos.client.security.SecurityProxy#login
public boolean login(List<String> servers) {
    
    try {
        //如果当前时间减去上一次刷新token时间小于token刷新时间窗口，返回
        if ((System.currentTimeMillis() - lastRefreshTime) < TimeUnit.SECONDS
                .toMillis(tokenTtl - tokenRefreshWindow)) {
            return true;
        }
        //遍历server，登录，然后刷新最后刷新时间
        for (String server : servers) {
            //登录实用的 nacosRestTemplate post请求  登陆成功刷新这三个值
            //accessToken = obj.get(Constants.ACCESS_TOKEN).asText();
            //tokenTtl = obj.get(Constants.TOKEN_TTL).asInt();
            //tokenRefreshWindow = tokenTtl / 10;
            if (login(server)) {
                lastRefreshTime = System.currentTimeMillis();
                return true;
            }
        }
    } catch (Throwable throwable) {
        SECURITY_LOGGER.warn("[SecurityProxy] login failed, error: ", throwable);
    }
    
    return false;
}

1.2、轮询

//com.alibaba.nacos.client.config.impl.ClientWorker.ConfigRpcTransportClient#startInternal
public void startInternal() throws NacosException {
    executor.schedule(new Runnable() {
        @Override
        public void run() {
            while (!executor.isShutdown() && !executor.isTerminated()) {
                try {
                    //阻塞队列，实现5秒定时
                    listenExecutebell.poll(5L, TimeUnit.SECONDS);
                    if (executor.isShutdown() || executor.isTerminated()) {
                        continue;
                    }
                    executeConfigListen();
                } catch (Exception e) {
                    LOGGER.error("[ rpc listen execute ] [rpc listen] exception", e);
                }
            }
        }
    }, 0L, TimeUnit.MILLISECONDS);
    
}

执行配置监听

public void executeConfigListen() {
    //定义缓存数据
    Map<String, List<CacheData>> listenCachesMap = new HashMap<String, List<CacheData>>(16);
    //定义移除的缓存数据
    Map<String, List<CacheData>> removeListenCachesMap = new HashMap<String, List<CacheData>>(16);
    //保存时间
    long now = System.currentTimeMillis();
    //判断本次刷新距离上次刷新全部数据是否超过了窗口期  
    //是否需要全同步 ALL_SYNC_INTERNAL = 5 * 60 * 1000L;
    boolean needAllSync = now - lastAllSyncTime >= ALL_SYNC_INTERNAL;
    //cacheMap: application,application.yml,application-dev.yml
    for (CacheData cache : cacheMap.get().values()) {
        
        synchronized (cache) {
            
            //check local listeners consistent.
            //检查标志位，下面的代码会设置为true，其他情况都是false，cacheData的监听器不为空
            if (cache.isSyncWithServer()) {
                //检查md5的值并启动任务
                cache.checkListenerMd5();
                //如果不需要全刷新，执行下一个cache
                if (!needAllSync) {
                    continue;
                }
            }
            //cacheData中有监听	NacosContextRefresh
            if (!CollectionUtils.isEmpty(cache.getListeners())) {
                //get listen  config
                if (!cache.isUseLocalConfigInfo()) {
                    List<CacheData> cacheDatas = listenCachesMap.get(String.valueOf(cache.getTaskId()));
                    if (cacheDatas == null) {
                        cacheDatas = new LinkedList<CacheData>();
                        //添加到listenCachesMap
                        listenCachesMap.put(String.valueOf(cache.getTaskId()), cacheDatas);
                    }
                    cacheDatas.add(cache);
                    
                }
                //cacheData没有监听
            } else if (CollectionUtils.isEmpty(cache.getListeners())) {
                
                if (!cache.isUseLocalConfigInfo()) {
                    List<CacheData> cacheDatas = removeListenCachesMap.get(String.valueOf(cache.getTaskId()));
                    if (cacheDatas == null) {
                        cacheDatas = new LinkedList<CacheData>();
                       //添加到removeListenCachesMap
                        removeListenCachesMap.put(String.valueOf(cache.getTaskId()), cacheDatas);
                    }
                    cacheDatas.add(cache);
                    
                }
            }
        }
        
    }
    //有没有key的配置变化了
    boolean hasChangedKeys = false;
    //cacheData中有监听
    if (!listenCachesMap.isEmpty()) {
        for (Map.Entry<String, List<CacheData>> entry : listenCachesMap.entrySet()) {
            String taskId = entry.getKey();
            // key:application  value:最后更新毫秒数
            // key:application.yml  value:最后更新毫秒数
            // key:application-dev.yml  value:最后更新毫秒数
            Map<String, Long> timestampMap = new HashMap<>(listenCachesMap.size() * 2);
            
            List<CacheData> listenCaches = entry.getValue();
            for (CacheData cacheData : listenCaches) {
                timestampMap.put(GroupKey.getKeyTenant(cacheData.dataId, cacheData.group, cacheData.tenant),
                        cacheData.getLastModifiedTs().longValue());
            }
            //构建请求
            ConfigBatchListenRequest configChangeListenRequest = buildConfigRequest(listenCaches);
            configChangeListenRequest.setListen(true);
            try {
                RpcClient rpcClient = ensureRpcClient(taskId);
                //发送请求	会增加一个 listenExecutebell 元素
                ConfigChangeBatchListenResponse configChangeBatchListenResponse = (ConfigChangeBatchListenResponse) requestProxy(
                        rpcClient, configChangeListenRequest);
                //请求成功
                if (configChangeBatchListenResponse != null && configChangeBatchListenResponse.isSuccess()) {
                    
                    Set<String> changeKeys = new HashSet<String>();
                    //handle changed keys,notify listener
                    //处理变化的key，启动监听
                    if (!CollectionUtils.isEmpty(configChangeBatchListenResponse.getChangedConfigs())) {
                        hasChangedKeys = true;
                        for (ConfigChangeBatchListenResponse.ConfigContext changeConfig : configChangeBatchListenResponse
                                .getChangedConfigs()) {
                            String changeKey = GroupKey
                                    .getKeyTenant(changeConfig.getDataId(), changeConfig.getGroup(),
                                            changeConfig.getTenant());
                            changeKeys.add(changeKey);
                            boolean isInitializing = cacheMap.get().get(changeKey).isInitializing();
                            //刷新cacheData的值，并启动监听
                            refreshContentAndCheck(changeKey, !isInitializing);
                        }
                        
                    }
                    
                    //handler content configs
                    //处理缓存
                    for (CacheData cacheData : listenCaches) {
                        //cloud-dubbo-gatway.yml+DEFAULT_GROUP
                        String groupKey = GroupKey
                                .getKeyTenant(cacheData.dataId, cacheData.group, cacheData.getTenant());
                        if (!changeKeys.contains(groupKey)) {
                            //sync:cache data md5 = server md5 && cache data md5 = all listeners md5.
                            synchronized (cacheData) {
                                //如果缓存的监听器不是空，将标志位设为true
                                if (!cacheData.getListeners().isEmpty()) {
                                    
                                    Long previousTimesStamp = timestampMap.get(groupKey);
                                    if (previousTimesStamp != null) {
                                        if (!cacheData.getLastModifiedTs().compareAndSet(previousTimesStamp,
                                                System.currentTimeMillis())) {
                                            continue;
                                        }
                                    }
                                    cacheData.setSyncWithServer(true);
                                }
                            }
                        }
                        
                        cacheData.setInitializing(false);
                    }
                    
                }
            } catch (Exception e) {
                
                LOGGER.error("Async listen config change error ", e);
                try {
                    Thread.sleep(50L);
                } catch (InterruptedException interruptedException) {
                    //ignore
                }
            }
        }
    }
    //cacheData中没有监听，将cacheData从监听器中移除
    if (!removeListenCachesMap.isEmpty()) {
        for (Map.Entry<String, List<CacheData>> entry : removeListenCachesMap.entrySet()) {
            String taskId = entry.getKey();
            List<CacheData> removeListenCaches = entry.getValue();
            ConfigBatchListenRequest configChangeListenRequest = buildConfigRequest(removeListenCaches);
            configChangeListenRequest.setListen(false);
            try {
                RpcClient rpcClient = ensureRpcClient(taskId);
                boolean removeSuccess = unListenConfigChange(rpcClient, configChangeListenRequest);
                if (removeSuccess) {
                    for (CacheData cacheData : removeListenCaches) {
                        synchronized (cacheData) {
                            if (cacheData.getListeners().isEmpty()) {
                                ClientWorker.this
                                        .removeCache(cacheData.dataId, cacheData.group, cacheData.tenant);
                            }
                        }
                    }
                }
                
            } catch (Exception e) {
                LOGGER.error("async remove listen config change error ", e);
            }
            try {
                Thread.sleep(50L);
            } catch (InterruptedException interruptedException) {
                //ignore
            }
        }
    }
    //需要全部刷新，设置全刷新时间
    if (needAllSync) {
        lastAllSyncTime = now;
    }
    //If has changed keys,notify re sync md5.
    //如果有key变化了，将阻塞队列增加一个，跳过5s定时，继续执行这个方法
    if (hasChangedKeys) {
        notifyListenConfig();
    }
}

检查配置md5

//com.alibaba.nacos.client.config.impl.CacheData#checkListenerMd5
void checkListenerMd5() {
    for (ManagerListenerWrap wrap : listeners) {
        //如果md5不相同，即配置有变化
        if (!md5.equals(wrap.lastCallMd5)) {
            safeNotifyListener(dataId, group, content, type, md5, encryptedDataKey, wrap);
        }
    }
}

notify

//com.alibaba.nacos.client.config.impl.CacheData#safeNotifyListener
//cacheData的任务线程池	最大5个线程，60秒过期
//static ThreadFactory internalNotifierFactory = r -> {
//        Thread t = new Thread(r);
//        t.setName("nacos.client.cachedata.internal.notifier");
//        t.setDaemon(true);
//        return t;
//    };
    
//    static final ThreadPoolExecutor INTERNAL_NOTIFIER = new ThreadPoolExecutor(0, CONCURRENCY, 60L, TimeUnit.SECONDS,
//            new SynchronousQueue<>(), internalNotifierFactory);
private void safeNotifyListener(final String dataId, final String group, final String content, final String type,
        final String md5, final String encryptedDataKey, final ManagerListenerWrap listenerWrap) {
    final Listener listener = listenerWrap.listener;
    if (listenerWrap.inNotifying) {
        LOGGER.warn(
                "[{}] [notify-currentSkip] dataId={}, group={}, md5={}, listener={}, listener is not finish yet,will try next time.",
                name, dataId, group, md5, listener);
        return;
    }
    //生成job
    Runnable job = new Runnable() {
        @Override
        public void run() {
            long start = System.currentTimeMillis();
            ClassLoader myClassLoader = Thread.currentThread().getContextClassLoader();
            ClassLoader appClassLoader = listener.getClass().getClassLoader();
            try {
                if (listener instanceof AbstractSharedListener) {
                    //为listener添加信息
                    AbstractSharedListener adapter = (AbstractSharedListener) listener;
                    adapter.fillContext(dataId, group);
                    LOGGER.info("[{}] [notify-context] dataId={}, group={}, md5={}", name, dataId, group, md5);
                }
                // Before executing the callback, set the thread classloader to the classloader of
                // the specific webapp to avoid exceptions or misuses when calling the spi interface in
                // the callback method (this problem occurs only in multi-application deployment).
                Thread.currentThread().setContextClassLoader(appClassLoader);
                
                ConfigResponse cr = new ConfigResponse();
                cr.setDataId(dataId);
                cr.setGroup(group);
                cr.setContent(content);
                cr.setEncryptedDataKey(encryptedDataKey);
                configFilterChainManager.doFilter(null, cr);
                String contentTmp = cr.getContent();
                listenerWrap.inNotifying = true;
                //这里调用AbstractSharedListener的实现方法，实现刷新配置
                listener.receiveConfigInfo(contentTmp);
                // compare lastContent and content
                if (listener instanceof AbstractConfigChangeListener) {
                    Map data = ConfigChangeHandler.getInstance()
                            .parseChangeData(listenerWrap.lastContent, content, type);
                    ConfigChangeEvent event = new ConfigChangeEvent(data);
                    ((AbstractConfigChangeListener) listener).receiveConfigChange(event);
                    listenerWrap.lastContent = content;
                }
                
                listenerWrap.lastCallMd5 = md5;
                LOGGER.info("[{}] [notify-ok] dataId={}, group={}, md5={}, listener={} ,cost={} millis.", name,
                        dataId, group, md5, listener, (System.currentTimeMillis() - start));
            } catch (NacosException ex) {
                LOGGER.error("[{}] [notify-error] dataId={}, group={}, md5={}, listener={} errCode={} errMsg={}",
                        name, dataId, group, md5, listener, ex.getErrCode(), ex.getErrMsg());
            } catch (Throwable t) {
                LOGGER.error("[{}] [notify-error] dataId={}, group={}, md5={}, listener={} tx={}", name, dataId,
                        group, md5, listener, t.getCause());
            } finally {
                listenerWrap.inNotifying = false;
                Thread.currentThread().setContextClassLoader(myClassLoader);
            }
        }
    };
    
    final long startNotify = System.currentTimeMillis();
    try {
        if (null != listener.getExecutor()) {
            listener.getExecutor().execute(job);
        } else {
            try {
                //提交任务
                INTERNAL_NOTIFIER.submit(job);
            } catch (RejectedExecutionException rejectedExecutionException) {
                LOGGER.warn(
                        "[{}] [notify-blocked] dataId={}, group={}, md5={}, listener={}, no available internal notifier,will sync notifier ",
                        name, dataId, group, md5, listener);
                job.run();
            } catch (Throwable throwable) {
                LOGGER.error(
                        "[{}] [notify-blocked] dataId={}, group={}, md5={}, listener={}, submit internal async task fail,throwable= ",
                        name, dataId, group, md5, listener, throwable);
                job.run();
            }
        }
    } catch (Throwable t) {
        LOGGER.error("[{}] [notify-error] dataId={}, group={}, md5={}, listener={} throwable={}", name, dataId,
                group, md5, listener, t.getCause());
    }
    final long finishNotify = System.currentTimeMillis();
    LOGGER.info("[{}] [notify-listener] time cost={}ms in ClientWorker, dataId={}, group={}, md5={}, listener={} ",
            name, (finishNotify - startNotify), dataId, group, md5, listener);
}

//com.alibaba.nacos.api.config.listener.AbstractSharedListener#receiveConfigInfo
public final void receiveConfigInfo(String configInfo) {
    innerReceive(dataId, group, configInfo);
}

innerReceive具体执行com.alibaba.cloud.nacos.refresh.NacosContextRefresher#registerNacosListener中匿名内部类添加的AbstractSharedListener。见下面。

这个匿名内部类的innerReceive方法中会发布RefreshEvent事件。根据这个事件实现配置刷新。

com.alibaba.cloud.nacos.refresh.NacosContextRefresher#onApplicationEvent

2、轮询器生效

注册

@Configuration(proxyBeanMethods = false)
@ConditionalOnProperty(name = "spring.cloud.nacos.config.enabled", matchIfMissing = true)
public class NacosConfigAutoConfiguration {

   @Bean
   public NacosConfigProperties nacosConfigProperties(ApplicationContext context) {
      if (context.getParent() != null
            && BeanFactoryUtils.beanNamesForTypeIncludingAncestors(
                  context.getParent(), NacosConfigProperties.class).length > 0) {
         return BeanFactoryUtils.beanOfTypeIncludingAncestors(context.getParent(),
               NacosConfigProperties.class);
      }
      return new NacosConfigProperties();
   }

   @Bean
   public NacosRefreshProperties nacosRefreshProperties() {
      return new NacosRefreshProperties();
   }

   @Bean
   public NacosRefreshHistory nacosRefreshHistory() {
      return new NacosRefreshHistory();
   }

   @Bean
   public NacosConfigManager nacosConfigManager(
         NacosConfigProperties nacosConfigProperties) {
      return new NacosConfigManager(nacosConfigProperties);
   }

   @Bean
   public NacosContextRefresher nacosContextRefresher(
         NacosConfigManager nacosConfigManager,
         NacosRefreshHistory nacosRefreshHistory) {
      // Consider that it is not necessary to be compatible with the previous
      // configuration
      // and use the new configuration if necessary.
      return new NacosContextRefresher(nacosConfigManager, nacosRefreshHistory);
   }

}

NacosContextRefresher 实现了ApplicationListener接口，监听ApplicationReadyEvent

//com.alibaba.cloud.nacos.refresh.NacosContextRefresher#onApplicationEvent
public void onApplicationEvent(ApplicationReadyEvent event) {
   // many Spring context
   if (this.ready.compareAndSet(false, true)) {
      this.registerNacosListenersForApplications();
   }
}

注册nacos监听

//com.alibaba.cloud.nacos.refresh.NacosContextRefresher#registerNacosListenersForApplications
private void registerNacosListenersForApplications() {
   if (isRefreshEnabled()) {
       //遍历propertySource，添加到轮询的cacheMap，一般是三个
       //application，application.yml，application-dev.yml
      for (NacosPropertySource propertySource : NacosPropertySourceRepository
            .getAll()) {
         if (!propertySource.isRefreshable()) {
            continue;
         }
         String dataId = propertySource.getDataId();
         registerNacosListener(propertySource.getGroup(), dataId);
      }
   }
}

注册listener，会发布refreshEvent

//com.alibaba.cloud.nacos.refresh.NacosContextRefresher#registerNacosListener
private void registerNacosListener(final String groupKey, final String dataKey) {
   String key = NacosPropertySourceRepository.getMapKey(dataKey, groupKey);
   Listener listener = listenerMap.computeIfAbsent(key,
         lst -> new AbstractSharedListener() {
            @Override
            public void innerReceive(String dataId, String group,
                  String configInfo) {
                //添加计数
               refreshCountIncrement();
                //添加刷新历史
               nacosRefreshHistory.addRefreshRecord(dataId, group, configInfo);
               // todo feature: support single refresh for listening
               //发布RefreshEvent
               applicationContext.publishEvent(
                     new RefreshEvent(this, null, "Refresh Nacos config"));
               if (log.isDebugEnabled()) {
                  log.debug(String.format(
                        "Refresh Nacos config group=%s,dataId=%s,configInfo=%s",
                        group, dataId, configInfo));
               }
            }
         });
   try {
       //添加listener
      configService.addListener(dataKey, groupKey, listener);
   }
   catch (NacosException e) {
      log.warn(String.format(
            "register fail for nacos listener ,dataId=[%s],group=[%s]", dataKey,
            groupKey), e);
   }
}

给轮询器添加listener

//com.alibaba.nacos.client.config.NacosConfigService#addListener
public void addListener(String dataId, String group, Listener listener) throws NacosException {
    worker.addTenantListeners(dataId, group, Arrays.asList(listener));
}

//com.alibaba.nacos.client.config.impl.ClientWorker#addTenantListeners
public void addTenantListeners(String dataId, String group, List<? extends Listener> listeners)
        throws NacosException {
    group = blank2defaultGroup(group);
    String tenant = agent.getTenant();
    //获取轮询器缓存，没有则增加再返回
    CacheData cache = addCacheDataIfAbsent(dataId, group, tenant);
    synchronized (cache) {
        for (Listener listener : listeners) {
            cache.addListener(listener);
        }
        //设置标志位
        cache.setSyncWithServer(false);
        //添加轮询器队列元素
        agent.notifyListenConfig();
    }
    
}

添加轮询器缓存

//com.alibaba.nacos.client.config.impl.ClientWorker#addCacheDataIfAbsent
public CacheData addCacheDataIfAbsent(String dataId, String group, String tenant) throws NacosException {
    //判断缓存有没有
    CacheData cache = getCache(dataId, group, tenant);
    if (null != cache) {
        return cache;
    }
    //获取key
    String key = GroupKey.getKeyTenant(dataId, group, tenant);
    synchronized (cacheMap) {
        //再次获取缓存，双重检查锁
        CacheData cacheFromMap = getCache(dataId, group, tenant);
        // multiple listeners on the same dataid+group and race condition,so
        // double check again
        // other listener thread beat me to set to cacheMap
        if (null != cacheFromMap) {
            cache = cacheFromMap;
            // reset so that server not hang this check
            //设置标志位
            cache.setInitializing(true);
        } else {
            //创建新的CacheData
            cache = new CacheData(configFilterChainManager, agent.getName(), dataId, group, tenant);
            //生成taskId		缓存个数/3000
            int taskId = cacheMap.get().size() / (int) ParamUtil.getPerTaskConfigSize();
            cache.setTaskId(taskId);
            // fix issue # 1317
            //启用远程同步
            if (enableRemoteSyncConfig) {
                //获取服务器数据并设置
                ConfigResponse response = getServerConfig(dataId, group, tenant, 3000L, false);
                cache.setContent(response.getContent());
            }
        }
        //设置缓存
        Map<String, CacheData> copy = new HashMap<String, CacheData>(this.cacheMap.get());
        copy.put(key, cache);
        cacheMap.set(copy);
    }
    LOGGER.info("[{}] [subscribe] {}", agent.getName(), key);
    
    MetricsMonitor.getListenConfigCountMonitor().set(cacheMap.get().size());
    
    return cache;
}

获取缓存

//com.alibaba.nacos.client.config.impl.ClientWorker#getCache
public CacheData getCache(String dataId, String group, String tenant) {
    if (null == dataId || null == group) {
        throw new IllegalArgumentException();
    }
    return cacheMap.get().get(GroupKey.getKeyTenant(dataId, group, tenant));
}

增加轮询器阻塞队列的个数

//com.alibaba.nacos.client.config.impl.ClientWorker.ConfigRpcTransportClient#notifyListenConfig
public void notifyListenConfig() {
    listenExecutebell.offer(bellItem);
}

发布RefreshEvent后，监听这个Event，可以实现配置的刷新。