深入理解SpringCloud一（3）Nacos配置中心

　　一、应用启动的时候如何从配置中心拉取配置文件
　　通过上一节的介绍，我们已经知道了配置加载的扩展点。下面我们已具体的Nacos配置中心来进行说明。
　　NacosConfigBootstrapConfiguration是@BootstrapConfiguration的配置类，在bootstrap 的SpringApplication创建的过程中，会加载这个类。这个Configuration类包括两个Bean，分别是NacosConfigManager，NacosPropertySourceLocator。
　　NacosConfigManager的核心作用是创建NacosConfigService，通过NacosConfigService从远程配置中心拉取配置。private String getConfigInner(String tenant, String dataId, String group, long timeoutMs) throws NacosException {     group = null2defaultGroup(group);     ParamUtils.checkKeyParam(dataId, group);     ConfigResponse cr = new ConfigResponse();          cr.setDataId(dataId);     cr.setTenant(tenant);     cr.setGroup(group);    	//省略一些代码     try {         String[] ct = worker.getServerConfig(dataId, group, tenant, timeoutMs);         cr.setContent(ct[0]);                  configFilterChainManager.doFilter(null, cr);         content = cr.getContent();                  return content;     } catch (NacosException ioe) {     }     content = cr.getContent();     return content; }
　　NacosPropertySourceLocator是PropertySourceLocator的实现类，上一节已经详细介绍过PropertySourceLocator。@Override public PropertySource<?> locate(Environment env) {    nacosConfigProperties.setEnvironment(env);    ConfigService configService = nacosConfigManager.getConfigService(); 	 //省略一些代码    nacosPropertySourceBuilder = new NacosPropertySourceBuilder(configService,          timeout);    CompositePropertySource composite = new CompositePropertySource(          NACOS_PROPERTY_SOURCE_NAME);    loadSharedConfiguration(composite);    loadExtConfiguration(composite);    loadApplicationConfiguration(composite, dataIdPrefix, nacosConfigProperties, env);    return composite; }二、配置中心的配置变动时，如何通知到应用
　　NacosConfigAutoConfiguration是@EnableAutoConfiguration的配置类，当application创建SpringApplication的过程中会被加载，会加载两个重要的Bean，NacosContextRefresher和NacosConfigManager，NacosConfigManager上面已经结束过。
　　NacosContextRefresher主要作用就是注册Nacos监听器。@Override public void onApplicationEvent(ApplicationReadyEvent event) {    // many Spring context    if (this.ready.compareAndSet(false, true)) {       this.registerNacosListenersForApplications();    } }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 {       //注册监听器       configService.addListener(dataKey, groupKey, listener);    }    catch (NacosException e) {    } }
　　监听器又是如何触发的？
　　Nacos使用长轮询方式获取配置，判断文件是否有变动，如果有变动则触发监听器。
　　NacosConfigService创建的时候，会创建ClientWorker对象，同时会创建长轮询任务。public NacosConfigService(Properties properties) throws NacosException {     //省略其他代码     this.worker = new ClientWorker(this.agent, this.configFilterChainManager, properties); }public ClientWorker(final HttpAgent agent, final ConfigFilterChainManager configFilterChainManager,         final Properties properties) {   //初始化一个延迟任务，10毫秒后调用checkConfigInfo   this.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); }public void checkConfigInfo() {     // Dispatch taskes.     int listenerSize = cacheMap.get().size();     // Round up the longingTaskCount.     int longingTaskCount = (int) Math.ceil(listenerSize / ParamUtil.getPerTaskConfigSize());     if (longingTaskCount > currentLongingTaskCount) {         for (int i = (int) currentLongingTaskCount; i < longingTaskCount; i++) {             // The task list is no order.So it maybe has issues when changing.             executorService.execute(new LongPollingRunnable(i));         }         currentLongingTaskCount = longingTaskCount;     } }
　　我们继续往下看LongPollingRunnable   @Override     public void run() {         //省略其他代码         // check server config  从服务器询问缓存的文件是否有变动         List changedGroupKeys = checkUpdateDataIds(cacheDatas, inInitializingCacheList);         if (!CollectionUtils.isEmpty(changedGroupKeys)) {              LOGGER.info(＂get changedGroupKeys:＂ + changedGroupKeys);         }           } }List checkUpdateDataIds(List cacheDatas, List inInitializingCacheList) throws Exception {     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()) {                 // It updates when cacheData occours in cacheMap by first time.                 inInitializingCacheList                         .add(GroupKey.getKeyTenant(cacheData.dataId, cacheData.group, cacheData.tenant));             }         }     }     boolean isInitializingCacheList = !inInitializingCacheList.isEmpty();     return checkUpdateConfigStr(sb.toString(), isInitializingCacheList); }List checkUpdateConfigStr(String probeUpdateString, boolean isInitializingCacheList) throws Exception {     Map params = new HashMap(2);     params.put(Constants.PROBE_MODIFY_REQUEST, probeUpdateString);     Map headers = new HashMap(2);     headers.put(＂Long-Pulling-Timeout＂, ＂＂ + timeout);     try {         // In order to prevent the server from handling the delay of the client＂s long task,         // increase the client＂s read timeout to avoid this problem.         //询问服务器，配置是否有变动，这里客户端读超时时间需要设置长一点         long readTimeoutMs = timeout + (long) Math.round(timeout >> 1);         HttpRestResult result = agent                 .httpPost(Constants.CONFIG_CONTROLLER_PATH + ＂/listener＂, headers, params, agent.getEncode(),                         readTimeoutMs);                  if (result.ok()) {             setHealthServer(true);             return parseUpdateDataIdResponse(result.getData());         } else {             setHealthServer(false);         }     } catch (Exception e) {         setHealthServer(false);     }     return Collections.emptyList(); }@Override public void run() {     try {     		// check server config         List changedGroupKeys = checkUpdateDataIds(cacheDatas, inInitializingCacheList);         if (!CollectionUtils.isEmpty(changedGroupKeys)) {             LOGGER.info(＂get changedGroupKeys:＂ + changedGroupKeys);         }         //如果服务端文件有变动，则从服务端重新拉取配置         for (String groupKey : changedGroupKeys) {             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))) {                 //如果文件MD5不一致，则触发监听器                 cacheData.checkListenerMd5();                 cacheData.setInitializing(false);             }         }         inInitializingCacheList.clear();                  executorService.execute(this);              } catch (Throwable e) {         executorService.schedule(this, taskPenaltyTime, TimeUnit.MILLISECONDS);     } }
　　如果文件MD5不一致，则触发监听器void checkListenerMd5() {     for (ManagerListenerWrap wrap : listeners) {         if (!md5.equals(wrap.lastCallMd5)) {             safeNotifyListener(dataId, group, content, type, md5, wrap);         }     } } private void safeNotifyListener(final String dataId, final String group, final String content, final String type,         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;             } catch (NacosException ex) {           }          }     };   try {         if (null != listener.getExecutor()) {             listener.getExecutor().execute(job);         } else {             job.run();         }     } catch (Throwable t) {   }  }
　　最终实现回调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));    } }三、总结
　　判断文件是否变动的核心是长轮询，客户端比较简单，只需要设置较长时间的读超时即可。后面我们会继续探究Nacos服务端的长轮询是如何时间的。