配置中心原理

一：SpringCloud应用启动时拉取配置

Spring 提供了强大的扩展机制。其中包括ApplicationContextInitializer，该扩展是在上下文准备阶段（prepareContext）

容器刷新之前做一些初始化工作，比如我们常用的配置中心 client 基本都是继承该初始化器

在容器刷新前将配置从远程拉到本地，然后封装成 PropertySource 放到 Environment 中供使用。

SpringCloud 规范中提供了 PropertySourceBootstrapConfiguration实现了 ApplicationContextInitializer接口

另外还提供了个 PropertySourceLocator，二者配合完成配置中心的接入。

在 PropertySourceBootstrapConfiguration 这个单例对象初始化的时候会将 Spring 容器中所有的 PropertySourceLocator 实现注入进来。

然后在 initialize()方法中循环所有的 PropertySourceLocator 进行配置的获取

可以看出 SpringCloud 应用是支持我们引入多个配置中心实现的，获取到配置后调用 insertPropertySources 方法将所有的 PropertySource 封装的一个个配置文件添加到 Spring 的环境变量 environment 中。

如果我们要使用Nacos的配置中心功能，需要在pom.xml中引入对应的依赖：

<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId>
    <version>2.2.9.RELEASE</version>
</dependency>

跟nacos自动注册和服务发现一样，nacos配置中文也是利用了springboot的自动装配原理来实现的。

我们查看spring-cloud-starter-alibaba-nacos-config-2.2.9.RELEASE.jar的META-INF/spring.factories文件：

上图展示了在 spring-cloud-starter-alibaba-nacos-config 包提供的自动装配类中引入了NacosConfigBootstrapConfiguration类。

/**
 * Nacos配置启动配置类。
 * 该类用于在Spring Boot应用程序启动时，根据配置决定是否启用Nacos配置管理。
 * 
 * @ConditionalOnProperty 允许根据属性值决定是否加载这个配置类。
 * 如果"spring.cloud.nacos.config.enabled"属性存在且其值为true（或非空），则会启用Nacos配置管理相关Bean的创建。
 */
@Configuration(proxyBeanMethods = false)
@ConditionalOnProperty(name = "spring.cloud.nacos.config.enabled", matchIfMissing = true)
public class NacosConfigBootstrapConfiguration {

    /**
     * 创建NacosConfigProperties Bean。
     * 如果当前上下文中不存在NacosConfigProperties Bean，则会创建一个新的实例。
     * NacosConfigProperties用于存储和管理与Nacos配置相关的属性。
     * 
     * @return NacosConfigProperties的实例。
     */
    @Bean
    @ConditionalOnMissingBean
    public NacosConfigProperties nacosConfigProperties() {
        return new NacosConfigProperties();
    }

    /**
     * 创建NacosConfigManager Bean。
     * NacosConfigManager用于管理Nacos配置的加载和更新。
     * 它依赖于NacosConfigProperties来获取配置中心的相关配置。
     * 
     * @param nacosConfigProperties Nacos配置属性的来源。
     * @return NacosConfigManager的实例。
     */
    @Bean
    @ConditionalOnMissingBean
    public NacosConfigManager nacosConfigManager(NacosConfigProperties nacosConfigProperties) {
        return new NacosConfigManager(nacosConfigProperties);
    }

    /**
     * 创建NacosPropertySourceLocator Bean。
     * NacosPropertySourceLocator用于从Nacos配置中心加载配置，并将其作为属性源添加到Spring应用程序上下文中。
     * 它依赖于NacosConfigManager来实际加载和管理配置。
     * 
     * @param nacosConfigManager 管理Nacos配置的实例。
     * @return NacosPropertySourceLocator的实例。
     */
    @Bean
    public NacosPropertySourceLocator nacosPropertySourceLocator(NacosConfigManager nacosConfigManager) {
        return new NacosPropertySourceLocator(nacosConfigManager);
    }

    /**
     * 创建SmartConfigurationPropertiesRebinder Bean。
     * 该Bean用于在应用程序启动时重新绑定配置属性，以支持配置的动态更新。
     * 它只在非默认行为下创建，以避免不必要的操作，提高性能和稳定性。
     * 
     * @param beans 存储所有@ConfigurationProperties实例的bean。
     * @return ConfigurationPropertiesRebinder的实例，用于配置属性的重新绑定。
     */
    @Bean
    @ConditionalOnMissingBean(search = SearchStrategy.CURRENT)
    @ConditionalOnNonDefaultBehavior
    public ConfigurationPropertiesRebinder smartConfigurationPropertiesRebinder(ConfigurationPropertiesBeans beans) {
        return new SmartConfigurationPropertiesRebinder(beans);
    }

}

从源码可以看到，NacosConfigBootstrapConfiguration自动配置类注入了NacosPropertySourceLocator这个bean

该类继承自上述说的 PropertySourceLocator，重写了 locate 方法进行配置的读取。

@Order(0)
public class NacosPropertySourceLocator implements PropertySourceLocator {
    // others
}

分析下 NacosPropertySourceLocator#locate()的代码：

@Override
public PropertySource<?> locate(Environment env) {
    nacosConfigProperties.setEnvironment(env);
    // 通过反射创建出一个NacosConfigService实例
    ConfigService configService = nacosConfigManager.getConfigService();

    if (null == configService) {
        log.warn("no instance of config service found, can't load config from nacos");
        return null;
    }
    long timeout = nacosConfigProperties.getTimeout();
    // 配置获取（使用 configService）、配置封装、配置缓存等操作
    nacosPropertySourceBuilder = new NacosPropertySourceBuilder(configService, timeout);
    String name = nacosConfigProperties.getName();

    String dataIdPrefix = nacosConfigProperties.getPrefix();
    if (StringUtils.isEmpty(dataIdPrefix)) {
        dataIdPrefix = name;
    }

    if (StringUtils.isEmpty(dataIdPrefix)) {
        dataIdPrefix = env.getProperty("spring.application.name");
    }

    CompositePropertySource composite = new CompositePropertySource(
        NACOS_PROPERTY_SOURCE_NAME);
    // 加载共享的配置信息
    loadSharedConfiguration(composite);
    // 加载扩展的配置信息
    loadExtConfiguration(composite);
    // 加载应用自身的配置信息
    loadApplicationConfiguration(composite, dataIdPrefix, nacosConfigProperties, env);
    return composite;
}

可以看到 Nacos 启动会加载以下三种配置文件，加载到配置文件后会封装成 NacosPropertySource 返回。

loadSharedConfiguration()：加载共享配置

private void loadSharedConfiguration(
        CompositePropertySource compositePropertySource) {
	// 获取共享配置
    List<NacosConfigProperties.Config> sharedConfigs = nacosConfigProperties
            .getSharedConfigs();
    if (!CollectionUtils.isEmpty(sharedConfigs)) {
        checkConfiguration(sharedConfigs, "shared-configs");
		// 加载配置
        loadNacosConfiguration(compositePropertySource, sharedConfigs);
    }
}

loadExtConfiguration()：加载扩展配置

private void loadExtConfiguration(CompositePropertySource compositePropertySource) {
	// 获取扩展配置
    List<NacosConfigProperties.Config> extConfigs = nacosConfigProperties
            .getExtensionConfigs();
    if (!CollectionUtils.isEmpty(extConfigs)) {
        checkConfiguration(extConfigs, "extension-configs");
        // 加载配置
        loadNacosConfiguration(compositePropertySource, extConfigs);
    }
}

loadApplicationConfiguration()：加载应用自身配置

private void loadApplicationConfiguration(
        CompositePropertySource compositePropertySource, String dataIdPrefix,
        NacosConfigProperties properties, Environment environment) {
    // 配置文件扩展名
    String fileExtension = properties.getFileExtension();
    // 配置组名
    String nacosGroup = properties.getGroup();
    // 不带扩展名后缀查询，application
    loadNacosDataIfPresent(compositePropertySource, dataIdPrefix, nacosGroup,
            fileExtension, true);
    // 带扩展名后缀查询，application.yml
    loadNacosDataIfPresent(compositePropertySource,
            dataIdPrefix + DOT + fileExtension, nacosGroup, fileExtension, true);
    // 带环境，带扩展名后缀查询，application-prod.yml
    for (String profile : environment.getActiveProfiles()) {
        String dataId = dataIdPrefix + SEP1 + profile + DOT + fileExtension;
        loadNacosDataIfPresent(compositePropertySource, dataId, nacosGroup, fileExtension, true);
    }
}

加载应用自身的配置信息loadApplicationConfiguration()时，同时会加载以下三种配置，分别是：

• 不带扩展名后缀查询，application
• 带扩展名后缀查询，application.yml
• 带环境，带扩展名后缀查询，application-prod.yml

依据后面加载的属性会覆盖掉之前加载的，可知，从上到下，优先级依次增高。

// 加载配置的核心方法是loadNacosDataIfPresent()：
private void loadNacosDataIfPresent(final CompositePropertySource composite,
        final String dataId, final String group, String fileExtension,
        boolean isRefreshable) {
    // 判断各种空的情况，直接返回
    if (null == dataId || dataId.trim().length() < 1) {
        return;
    }
    if (null == group || group.trim().length() < 1) {
        return;
    }
    // 
    NacosPropertySource propertySource = this.loadNacosPropertySource(dataId, group,fileExtension, isRefreshable);
    this.addFirstPropertySource(composite, propertySource, false);
}
 
private NacosPropertySource loadNacosPropertySource(final String dataId,
        final String group, String fileExtension, boolean isRefreshable) {
    if (NacosContextRefresher.getRefreshCount() != 0) {
        if (!isRefreshable) {
            return NacosPropertySourceRepository.getNacosPropertySource(dataId, group);
        }
    }
    // 调用 loadNacosData 获取配置
    return nacosPropertySourceBuilder.build(dataId, group, fileExtension, isRefreshable);
}

nacosPropertySourceBuilder.build()方法调用 loadNacosData 获取配置

然后封装成 NacosPropertySource，并且将该对象缓存到 NacosPropertySourceRepository中，后续会用到。

来看看如何获取配置的loadNacosData()：

private List<PropertySource<?>> loadNacosData(String dataId, String group, String fileExtension) {
    String data = null;
    try {
        // 通过NacosConfigService加载配置内容
        data = configService.getConfig(dataId, group, timeout);
        if (StringUtils.isEmpty(data)) {
            log.warn(
                    "Ignore the empty nacos configuration and get it based on dataId[{}] & group[{}]",
                    dataId, group);
            return Collections.emptyList();
        }
        if (log.isDebugEnabled()) {
            log.debug(String.format(
                    "Loading nacos data, dataId: '%s', group: '%s', data: %s", dataId,
                    group, data));
        }
        // 解析返回的字符串
        return NacosDataParserHandler.getInstance().parseNacosData(dataId, data, fileExtension);
    }
    catch (NacosException e) {
        log.error("get data from Nacos error,dataId:{} ", dataId, e);
    }
    catch (Exception e) {
        log.error("parse data from Nacos error,dataId:{},data:{}", dataId, data, e);
    }
    return Collections.emptyList();
}

loadNacosData 方法中会将实际配置加载请求委托给NacosConfigService去做，然后解析返回的字符串

解析器实现了 PropertySourceLoader 接口，支持 yml、properties、xml、json 这几种。

public String getConfig(String dataId, String group, long timeoutMs) throws NacosException {
    return getConfigInner(namespace, dataId, group, timeoutMs);
}
 
private String getConfigInner(String tenant, String dataId, String group, long timeoutMs) throws NacosException {
    // 默认组名：DEFAULT_GROUP
    group = blank2defaultGroup(group);
    // 校验组名、dataId
    ParamUtils.checkKeyParam(dataId, group);
    ConfigResponse cr = new ConfigResponse();
    // 设置配置信息，通过 namespace, dataId, group能唯一定位到一个配置文件
    cr.setDataId(dataId);
    cr.setTenant(tenant);
    cr.setGroup(group);
 
    // We first try to use local failover content if exists.
    // A config content for failover is not created by client program automatically,
    // but is maintained by user.
    // This is designed for certain scenario like client emergency reboot,
    // changing config needed in the same time, while nacos server is down.
 
    // 首先尝试使用本地如果存在的故障转移的配置内容，如果能读到失败转移的配置信息，则直接返回了。
    // 优先使用失败转移，设计的目的是当server挂后，又需要修改配置，就可以读本地目录
    String content = LocalConfigInfoProcessor.getFailover(worker.getAgentName(), dataId, group, tenant);
    if (content != null) {
        LOGGER.warn("[{}] [get-config] get failover ok, dataId={}, group={}, tenant={}, config={}", worker.getAgentName(), dataId, group, tenant, ContentUtils.truncateContent(content));
        cr.setContent(content);
        String encryptedDataKey = LocalEncryptedDataKeyProcessor.getEncryptDataKeyFailover(agent.getName(), dataId, group, tenant);
        cr.setEncryptedDataKey(encryptedDataKey);
        configFilterChainManager.doFilter(null, cr);
        content = cr.getContent();
        return content;
    }
 
    try {
        // 去服务端拉取，这个就是正常逻辑，获取服务端存储的配置信息
        ConfigResponse response = worker.getServerConfig(dataId, group, tenant, timeoutMs, false);
        cr.setContent(response.getContent());
        cr.setEncryptedDataKey(response.getEncryptedDataKey());
        configFilterChainManager.doFilter(null, cr);
        content = cr.getContent();
 
        return content;
    } catch (NacosException ioe) {
        if (NacosException.NO_RIGHT == ioe.getErrCode()) {
            throw ioe;
        }
        LOGGER.warn("[{}] [get-config] get from server error, dataId={}, group={}, tenant={}, msg={}", worker.getAgentName(), dataId, group, tenant, ioe.toString());
    }
 
    // 非鉴权失败的异常（NacosException.NO_RIGHT）的，尝试从本地快照中获取配置
    content = LocalConfigInfoProcessor.getSnapshot(worker.getAgentName(), dataId, group, tenant);
    if (content != null) {
        LOGGER.warn("[{}] [get-config] get snapshot ok, dataId={}, group={}, tenant={}, config={}", worker.getAgentName(), dataId, group, tenant, ContentUtils.truncateContent(content));
    }
    cr.setContent(content);
    String encryptedDataKey = LocalEncryptedDataKeyProcessor.getEncryptDataKeySnapshot(agent.getName(), dataId, group, tenant);
    cr.setEncryptedDataKey(encryptedDataKey);
    configFilterChainManager.doFilter(null, cr);
    content = cr.getContent();
    return content;
}

从源码可以看到，首先获取本地缓存文件的配置内容（故障转移的配置内容）：

• 如果有直接返回；
• 如果从本地没找到相应配置文件，则通过grpc请求服务端拉取；

如果grpc请求报除了鉴权失败之外的异常，例如超时，还会尝试从本地快照中获取配置，不至于偶尔超时了就配置没了。

二：注册配置监听器

除了引入了NacosConfigBootstrapConfiguration自动配置类，还引入了NacosConfigAutoConfiguration这个自动配置类

这个配置类就负责配置监听器的注册

@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);
	}
 
	@Bean
	@ConditionalOnMissingBean(search = SearchStrategy.CURRENT)
	@ConditionalOnNonDefaultBehavior
	public ConfigurationPropertiesRebinder smartConfigurationPropertiesRebinder(
			ConfigurationPropertiesBeans beans) {
		// If using default behavior, not use SmartConfigurationPropertiesRebinder.
		// Minimize te possibility of making mistakes.
		return new SmartConfigurationPropertiesRebinder(beans);
	}
 
}

它往spring IOC容器中注入了NacosContextRefresher这个bean：

public class NacosContextRefresher implements ApplicationListener<ApplicationReadyEvent>, ApplicationContextAware {
 
   public void onApplicationEvent(ApplicationReadyEvent event) {
		// many Spring context
		if (this.ready.compareAndSet(false, true)) {
			this.registerNacosListenersForApplications();
		}
   }
}

NacosContextRefresher类实现了ApplicationListener监听器接口，并且监听了ApplicationReadyEvent事件

在事件监听回调方法onApplicationEvent()执行配置监听器注册。

private void registerNacosListenersForApplications() {
    // 是否开启配置刷新 spring.cloud.nacos.config.refreshEnabled属性的值为true, 表示开启配置刷新
    if (isRefreshEnabled()) {
        // 这个就是从缓存中拿到之前获取完配置内容后，封装成NacosPropertySource之后，缓存起来的内容
		// ConcurrentHashMap<String, NacosPropertySource> NACOS_PROPERTY_SOURCE_REPOSITORY
        for (NacosPropertySource propertySource : NacosPropertySourceRepository
                .getAll()) {
            if (!propertySource.isRefreshable()) {
                continue;
            }
            // 获取配置dataId
            String dataId = propertySource.getDataId();
        	// 注册nacos配置监听器
            registerNacosListener(propertySource.getGroup(), dataId);
        }
    }
}

registerNacosListenersForApplications()方法里会进行判断，如果自动刷新机制是开启的，则进行监听器注册。

private void registerNacosListener(final String groupKey, final String dataKey) {
    // key = {dataId,group}
    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);
                	// 通过Spring上下文发布一个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 {
        // 注册配置监听器，以 dataId + groupId + namespace 为维度进行注册的
        configService.addListener(dataKey, groupKey, listener);
    }
    catch (NacosException e) {
        log.warn(String.format(
                "register fail for nacos listener ,dataId=[%s],group=[%s]", dataKey,
                groupKey), e);
    }
}

private final ClientWorker worker;
 
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 {
    // 如果为空，取默认组名DEFAULT_GROUP
    group = blank2defaultGroup(group);
    // 命名空间ID
    String tenant = agent.getTenant();
    // 根据dataId,group和tenant获取一个cacheData
    CacheData cache = addCacheDataIfAbsent(dataId, group, tenant);
 
    // 可以看到，一个配置文件可以有多个监听器
    synchronized (cache) {
        // 添加监听器
        for (Listener listener : listeners) {
            cache.addListener(listener);
        }
        // 非丢弃，删除类型
        cache.setDiscard(false);
        // 未同步到服务端
        cache.setConsistentWithServer(false);
        // 处理配置类
        agent.notifyListenConfig();
    }
    
}
 
public CacheData addCacheDataIfAbsent(String dataId, String group, String tenant) throws NacosException {
    // 从缓存中获取cacheData
    CacheData cache = getCache(dataId, group, tenant);
    if (null != cache) {
        // 缓存中存在，则直接返回
        return cache;
    }
    String key = GroupKey.getKeyTenant(dataId, group, tenant);
    synchronized (cacheMap) {
        CacheData cacheFromMap = getCache(dataId, group, tenant);
        // 双重检查
        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);
            int taskId = calculateTaskId();
            increaseTaskIdCount(taskId);
            cache.setTaskId(taskId);
            // fix issue # 1317
            if (enableRemoteSyncConfig) {
                ConfigResponse response = getServerConfig(dataId, group, tenant, 3000L, false);
                cache.setEncryptedDataKey(response.getEncryptedDataKey());
                cache.setContent(response.getContent());
            }
        }
 
        // 写时复制思想
        Map<String, CacheData> copy = new HashMap<>(this.cacheMap.get());
        // 存入缓存
        copy.put(key, cache);
        cacheMap.set(copy);
    }
    LOGGER.info("[{}] [subscribe] {}", agent.getName(), key);
    
    MetricsMonitor.getListenConfigCountMonitor().set(cacheMap.get().size());
    
    return cache;
}

注册操作经过 ConfigService，在 ClientWorker 中处理，这里会创建一个 CacheData 对象

所有的配置信息和监听器都存放在CacheData中，包括一些操作方法。

所以它不仅仅是一个存储的配置信息的实体，也包括了操作，获取，更新这个信息的功能。

CacheData的一些主要属性：

// 是否初始化快照
static boolean initSnapshot;
 
// 静态块，获取属性值
static {
    initSnapshot = NacosClientProperties.PROTOTYPE.getBoolean("nacos.cache.data.init.snapshot", true);
    LOGGER.info("nacos.cache.data.init.snapshot = {} ", initSnapshot);
}
 
private final String envName;
 
private final ConfigFilterChainManager configFilterChainManager;
 
// dataId
public final String dataId;
 
// groupId
public final String group;
 
// namespaceId
public final String tenant;
 
/**
 * cacheData的监听器列表
 * 监听配置是在cacheData中配置上监听器，等待触发条件后，进行本地的内容和远程内容的比对，
 * 如果不一致，调用监听器上的回调逻辑，完成配置的更新通知。
 */
private final CopyOnWriteArrayList<ManagerListenerWrap> listeners;
 
// 配置信息的md5值，用来判断配置有没有变更
private volatile String md5;
 
// 是否使用本地缓存
private volatile boolean isUseLocalConfig = false;
 
// 上次配置修改时间
private volatile long localConfigLastModified;
 
// 具体的配置内容
private volatile String content;
 
// 加密数据的key
private volatile String encryptedDataKey;
 
// 修改时间
private volatile AtomicLong lastModifiedTs = new AtomicLong(0);
 
/**
 * notify change flag,for notify&sync concurrent control. 1.reset to false if starting to sync with server. 2.update
 * to true if receive config change notification.
 */
private volatile AtomicBoolean receiveNotifyChanged = new AtomicBoolean(false);
 
// 任务ID
private int taskId;
 
// 是否初始化
private volatile boolean isInitializing = true;
 
// 是否和服务端同步
private volatile AtomicBoolean isConsistentWithServer = new AtomicBoolean();
 
// 是否丢失
private volatile boolean isDiscard = false;
 
// 类型
private String type;

并且在 addCacheDataIfAbsent 方法中会将刚才创建的 CacheData 缓存到 ClientWorker 中的一个 Map 中，后续会用到。

private final AtomicReference<Map<String, CacheData>> cacheMap = new AtomicReference<>(new HashMap<>());

三：配置热更新

前面说了服务启动的时候从远程 Nacos 服务端拉到配置，以及服务启动后对需要支持热更新的配置都注册了一个监听器

下面主要说下配置变动后具体是怎么处理的

回到NacosPropertySourceLocator#locate()方法，该方法首先会通过NacosConfigManager获取一个 ConfigService

// 看下ConfigService是如何创建的
public ConfigService getConfigService() {
    if (Objects.isNull(service)) {
        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.config.ConfigFactory#createConfigService(java.util.Properties)
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);
    }
}

从源码可以看到，NacosConfigManager 中会进行一个 ConfigService 单例对象的创建

创建流程最终会委托给 ConfigFactory，使用反射方式创建一个 NacosConfigService 的实例对象

NacosConfigService 是一个很核心的类，配置的获取，监听器的注册都需要经此。

我们看下 NacosConfigService 的构造函数，会去创建一个 ClientWorker 类的对象，这个类是实现配置热更新的核心类。

/**
 * 长轮训
 */
private final ClientWorker worker;
 
public NacosConfigService(Properties properties) throws NacosException {
    final NacosClientProperties clientProperties = NacosClientProperties.PROTOTYPE.derive(properties);
    ValidatorUtils.checkInitParam(clientProperties);
 
    // 初始化Namespace
    initNamespace(clientProperties);
    // 创建了一个配置过滤器链，可以采用SPI扩展机制加载对应的过滤器实现类
    this.configFilterChainManager = new ConfigFilterChainManager(clientProperties.asProperties());
    // 创建了一个服务管理器，内含一个定时轮询线程池，每隔30s拉取一次服务
    ServerListManager serverListManager = new ServerListManager(clientProperties);
    serverListManager.start();
    // 创建了一个客户端工作者，包含了一个代理对象
    this.worker = new ClientWorker(this.configFilterChainManager, serverListManager, clientProperties);
    // will be deleted in 2.0 later versions
    agent = new ServerHttpAgent(serverListManager);
}

// ClientWorker的构造方法
public ClientWorker(final ConfigFilterChainManager configFilterChainManager, ServerListManager serverListManager,
        final NacosClientProperties properties) throws NacosException {
    this.configFilterChainManager = configFilterChainManager;
 
    // 初始化timeout、taskPenaltyTime、enableRemoteSyncConfig属性
    init(properties);
 
    // 创建一个用于配置服务端的Rpc通信客户端
    agent = new ConfigRpcTransportClient(properties, serverListManager);
    // 计算合适的核心线程数
    int count = ThreadUtils.getSuitableThreadCount(THREAD_MULTIPLE);
    // 初始化一个线程池
    ScheduledExecutorService executorService = Executors.newScheduledThreadPool(Math.max(count, MIN_THREAD_NUM),
            r -> {
                Thread t = new Thread(r);
                t.setName("com.alibaba.nacos.client.Worker");
                t.setDaemon(true);
                return t;
            });
    // 配置了一个异步处理线程池
    agent.setExecutor(executorService);
    // 调用start方法
    agent.start();
    
}

首先创建一个用于配置服务端的Rpc通信客户端，然后初始化了一个线程池，配置到rpc客户端中，调用ConfigRpcTransportClient#start()启动：

public void start() throws NacosException {
    // 认证服务，主要是通过Secret Key和Access Key做认证用
    securityProxy.login(this.properties);
    // 每隔5s执行一次认证
    this.executor.scheduleWithFixedDelay(() -> securityProxy.login(properties), 0,
            this.securityInfoRefreshIntervalMills, TimeUnit.MILLISECONDS);
    // 内部启动
    startInternal();
}

/**
 * 阻塞队列
 */
private final BlockingQueue<Object> listenExecutebell = new ArrayBlockingQueue<>(1);
 
public void startInternal() {
    // 线程池在阻塞等到信号的到来
    executor.schedule(() -> {
        while (!executor.isShutdown() && !executor.isTerminated()) {
            try {
                // 获取到listenExecutebell.offer(bellItem)的信号
                // 如果没有监听器的变动，则等待5s处理一次
                listenExecutebell.poll(5L, TimeUnit.SECONDS);
                if (executor.isShutdown() || executor.isTerminated()) {
                    continue;
                }
                // 执行配置监听
                executeConfigListen();
            } catch (Throwable e) {
                LOGGER.error("[rpc listen execute] [rpc listen] exception", e);
                try {
                    Thread.sleep(50L);
                } catch (InterruptedException interruptedException) {
                    //ignore
                }
                notifyListenConfig();
            }
        }
    }, 0L, TimeUnit.MILLISECONDS);
    
}

核心代码就是executeConfigListen()执行配置监听：

public void executeConfigListen() {
    // 存放含有listen的cacheData
    Map<String, List<CacheData>> listenCachesMap = new HashMap<>(16);
    // 存放不含有listen的cacheData
    Map<String, List<CacheData>> removeListenCachesMap = new HashMap<>(16);
    long now = System.currentTimeMillis();
    // 当前时间 减去 上一次全量同步的时间，如果大于3分钟，表示到了全量同步的时间了
    boolean needAllSync = now - lastAllSyncTime >= ALL_SYNC_INTERNAL;
    for (CacheData cache : cacheMap.get().values()) {
        
        synchronized (cache) {
            
            // 是否和服务端一致
            if (cache.isConsistentWithServer()) {
                // 一致则检查md5值，若md5值和上一个不一样，则说明变动了，需要通知监听器
                cache.checkListenerMd5();
                // 是否到全量同步时间了，未到则直接跳过
                if (!needAllSync) {
                    continue;
                }
            }
            
            if (!cache.isDiscard()) {
                // 非丢弃型，即新增，放入listenCachesMap
                //get listen  config
                if (!cache.isUseLocalConfigInfo()) {
                    List<CacheData> cacheDatas = listenCachesMap.get(String.valueOf(cache.getTaskId()));
                    if (cacheDatas == null) {
                        cacheDatas = new LinkedList<>();
                        listenCachesMap.put(String.valueOf(cache.getTaskId()), cacheDatas);
                    }
                    cacheDatas.add(cache);
                    
                }
            } else if (cache.isDiscard() && CollectionUtils.isEmpty(cache.getListeners())) {
                // 丢弃型，即删除， 放入removeListenCachesMap
                if (!cache.isUseLocalConfigInfo()) {
                    List<CacheData> cacheDatas = removeListenCachesMap.get(String.valueOf(cache.getTaskId()));
                    if (cacheDatas == null) {
                        cacheDatas = new LinkedList<>();
                        removeListenCachesMap.put(String.valueOf(cache.getTaskId()), cacheDatas);
                    }
                    cacheDatas.add(cache);
                    
                }
            }
        }
        
    }
    
    // 如果需要和服务端数据同步，则listenCachesMap和removeListenCachesMap存放了本地数据，需要和服务端对比
    boolean hasChangedKeys = checkListenCache(listenCachesMap);
    
    //execute check remove listen.
    checkRemoveListenCache(removeListenCachesMap);
    
    if (needAllSync) {
        // 更新同步时间
        lastAllSyncTime = now;
    }
    //If has changed keys,notify re sync md5.
    if (hasChangedKeys) {
        // 服务端告知了有数据变动，则需要再同步一次
        notifyListenConfig();
    }
    
}

上面主要做了两件事：

• 首先遍历之前缓存到cacheMap的所有CacheData，判断是否和服务端保持一致
  • 如果一致，则检查配置的md5值
  • 如果md5值和上一个不一样，则说明变动了，需要通知监听器

// cacheData的校验方法
void checkListenerMd5() {
    // 遍历这个配置所有的监听者
    for (ManagerListenerWrap wrap : listeners) {
        if (!md5.equals(wrap.lastCallMd5)) {
            // 如果内容变动了，直接通知监听器处理，并且更新 listenerWrap 中的 content、Md5
            safeNotifyListener(dataId, group, content, type, md5, encryptedDataKey, wrap);
        }
    }
}

// 遍历这个配置对应的所有的监听者，判断内容是否变动，如果变动了，调用safeNotifyListener()直接通知监听器处理。
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={},tenant={}, md5={}, listener={}, listener is not finish yet,will try next time.",
                envName, dataId, group, tenant, md5, listener);
        return;
    }
 
    // 定义一个通知任务
    NotifyTask job = new NotifyTask() {
        
        @Override
        public void run() {
            long start = System.currentTimeMillis();
            ClassLoader myClassLoader = Thread.currentThread().getContextClassLoader();
            ClassLoader appClassLoader = listener.getClass().getClassLoader();
            ScheduledFuture<?> timeSchedule = null;
            
            try {
                if (listener instanceof AbstractSharedListener) {
                    AbstractSharedListener adapter = (AbstractSharedListener) listener;
                    adapter.fillContext(dataId, group);
                    LOGGER.info("[{}] [notify-context] dataId={}, group={},tenant={}, md5={}", envName, dataId,
                            group, tenant, 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();
                timeSchedule = getNotifyBlockMonitor().schedule(
                        new LongNotifyHandler(listener.getClass().getSimpleName(), dataId, group, tenant, md5,
                                notifyWarnTimeout, Thread.currentThread()), notifyWarnTimeout,
                        TimeUnit.MILLISECONDS);
                listenerWrap.inNotifying = true;
 
                // 回调通知，也就是通知变动的内容
                // 这里就是执行前面说到的注册监听时的一个回调函数，里面其实最主要的就是发布了一个RefreshEvent事件，springcloud会处理这个事件
                listener.receiveConfigInfo(contentTmp);
 
                // compare lastContent and content
                if (listener instanceof AbstractConfigChangeListener) {
                    // 扩展点，告知配置内容的变动
                    Map<String, ConfigChangeItem> data = ConfigChangeHandler.getInstance()
                            .parseChangeData(listenerWrap.lastContent, contentTmp, type);
                    ConfigChangeEvent event = new ConfigChangeEvent(data);
                    ((AbstractConfigChangeListener) listener).receiveConfigChange(event);
                    listenerWrap.lastContent = contentTmp;
                }
 
                // 赋值最新的md5
                listenerWrap.lastCallMd5 = md5;
                LOGGER.info(
                        "[{}] [notify-ok] dataId={}, group={},tenant={}, md5={}, listener={} ,job run cost={} millis.",
                        envName, dataId, group, tenant, md5, listener, (System.currentTimeMillis() - start));
            } catch (NacosException ex) {
                LOGGER.error(
                        "[{}] [notify-error] dataId={}, group={},tenant={},md5={}, listener={} errCode={} errMsg={},stackTrace :{}",
                        envName, dataId, group, tenant, md5, listener, ex.getErrCode(), ex.getErrMsg(),
                        getTrace(ex.getStackTrace(), 3));
            } catch (Throwable t) {
                LOGGER.error("[{}] [notify-error] dataId={}, group={},tenant={}, md5={}, listener={} tx={}",
                        envName, dataId, group, tenant, md5, listener, getTrace(t.getStackTrace(), 3));
            } finally {
                listenerWrap.inNotifying = false;
                Thread.currentThread().setContextClassLoader(myClassLoader);
                if (timeSchedule != null) {
                    timeSchedule.cancel(true);
                }
            }
        }
    };
    
    try {
        // 监听器配置了异步执行器，就异步执行
        if (null != listener.getExecutor()) {
            LOGGER.info(
                    "[{}] [notify-listener] task submitted to user executor, dataId={}, group={},tenant={}, md5={}, listener={} ",
                    envName, dataId, group, tenant, md5, listener);
            job.async = true;
            listener.getExecutor().execute(job);
        } else {
            // 同步执行
            LOGGER.info(
                    "[{}] [notify-listener] task execute in nacos thread, dataId={}, group={},tenant={}, md5={}, listener={} ",
                    envName, dataId, group, tenant, md5, listener);
            job.run();
        }
    } catch (Throwable t) {
        LOGGER.error("[{}] [notify-listener-error] dataId={}, group={},tenant={}, md5={}, listener={} throwable={}",
                envName, dataId, group, tenant, md5, listener, t.getCause());
    }
}

• 服务端告知了有数据变动，则需要再同步一次，其实就是往阻塞队列中存放内容，再来一遍前面的同步流程

public void notifyListenConfig() {
    // listenExecutebell是一个阻塞队列，放入bellItem，即一个触发条件，相当于生产者
    listenExecutebell.offer(bellItem);
}

四：RefreshEvent 事件处理

回调监听器的方法中，主要就是发布了一个RefreshEvent事件，这个事件主要由 SpringCloud 相关类来处理

RefreshEvent 事件会由 RefreshEventListener 来处理，该 listener 含有一个 ContextRefresher 的对象。

public void onApplicationEvent(ApplicationEvent event) {
    if (event instanceof ApplicationReadyEvent) {
        handle((ApplicationReadyEvent) event);
    }
    else if (event instanceof RefreshEvent) {
        handle((RefreshEvent) event);
    }
}
 
public void handle(RefreshEvent event) {
    if (this.ready.get()) { // don't handle events before app is ready
        log.debug("Event received " + event.getEventDesc());
        // private ContextRefresher refresh
        Set<String> keys = this.refresh.refresh();
        log.info("Refresh keys changed: " + keys);
    }
}
 
// org.springframework.cloud.context.refresh.ContextRefresher#refresh
public synchronized Set<String> refresh() {
    // 刷新Environment环境信息
    Set<String> keys = refreshEnvironment();
    this.scope.refreshAll();
    return keys;
}
 
public synchronized Set<String> refreshEnvironment() {
    Map<String, Object> before = extract(
            this.context.getEnvironment().getPropertySources());
    // 添加配置内容到环境对象中
    addConfigFilesToEnvironment();
    Set<String> keys = changes(before,
            extract(this.context.getEnvironment().getPropertySources())).keySet();
    // 发布EnvironmentChangeEvent事件（环境变更事件）
    this.context.publishEvent(new EnvironmentChangeEvent事件(this.context, keys));
    return keys;
}

从源码可以看到，refreshEnvironment 会去刷新 Spring 环境变量，具体刷新思想就是重新创建一个Environment

然后将这个新的环境信息设置到原有的 Spring 环境中。

拿到所有变化的配置项后，发布一个环境变化的 EnvironmentChangeEvent（环境变更事件）事件。

ConfigurationPropertiesRebinder 会监听 EnvironmentChangeEvent 事件

监听到事件后会对所有的标注有 ConfigurationProperties 注解的配置类进行销毁后重新初始化的操作，完成后我们的配置类中的属性就是最新的了。

// org.springframework.cloud.context.properties.ConfigurationPropertiesRebinder#onApplicationEvent
public void onApplicationEvent(EnvironmentChangeEvent event) {
    if (this.applicationContext.equals(event.getSource())
            // Backwards compatible
            || event.getKeys().equals(event.getSource())) {
        rebind();
    }
}

// 所有的标注有 ConfigurationProperties 注解的配置类
private ConfigurationPropertiesBeans beans;
public ConfigurationPropertiesRebinder(ConfigurationPropertiesBeans beans) {
    this.beans = beans;
}
 
public void rebind() {
    this.errors.clear();
    for (String name : this.beans.getBeanNames()) {
        rebind(name);
    }
}
 
public boolean rebind(String name) {
    if (!this.beans.getBeanNames().contains(name)) {
        return false;
    }
    if (this.applicationContext != null) {
        try {
            Object bean = this.applicationContext.getBean(name);
            if (AopUtils.isAopProxy(bean)) {
                bean = ProxyUtils.getTargetObject(bean);
            }
            if (bean != null) {
                // TODO: determine a more general approach to fix this.
                // see https://github.com/spring-cloud/spring-cloud-commons/issues/571
                if (getNeverRefreshable().contains(bean.getClass().getName())) {
                    return false; // ignore
                }
 
                // 销毁bean
                this.applicationContext.getAutowireCapableBeanFactory()
                        .destroyBean(bean);
                // 重新初始化bean
                this.applicationContext.getAutowireCapableBeanFactory()
                        .initializeBean(bean, name);
                return true;
            }
        }
        catch (RuntimeException e) {
            this.errors.put(name, e);
            throw e;
        }
        catch (Exception e) {
            this.errors.put(name, e);
            throw new IllegalStateException("Cannot rebind to " + name, e);
        }
    }
    return false;
}

上述代码只会对标有 ConfigurationProperties 注解的配置类进行rebind，那对于普通组件类里标有 @Value 注解的属性要怎么生效呢？

这个其实需要配合 @RefreshScope 注解来生效的。

我们继续回到前面处理RefreshEvent事件的ContextRefresher#refresh()方法，接着会有一步 refreshAll 的操作，会调用父类的destroy()方法。

public synchronized Set<String> refresh() {
    Set<String> keys = refreshEnvironment();
    // private RefreshScope scope;
    this.scope.refreshAll();
    return keys;
}
 
// org.springframework.cloud.context.scope.refresh.RefreshScope#refreshAll
public void refreshAll() {
    // 调用父类的销毁方法：org.springframework.cloud.context.scope.GenericScope#destroy()
    super.destroy();
    this.context.publishEvent(new RefreshScopeRefreshedEvent());
}

父类就是 GenericScope，我们知道 Spring 中的 Bean 是有Scope 的概念的

Spring 默认 Scope 有单例和原型两种，同时提供了 Scope 扩展接口，通过实现该接口我们可以定义自己的 Scope。

在Spring IOC容器初始化的时候，doGetBean()方法中，这些自定义 Scope 类型对象的管理会交给相应的 Scope 实现去管理。

SpringCloud 实现的 RefreshScope 就是用来在运行时动态刷新 Bean 用的，RefreshScope 继承 GenericScope，提供 get()和 destroy()方法。

回到refreshAll()方法，在 refreshAll()中调用 super.destroy()方法时会将该 scope 的这些 Bean 都销毁掉

在下次 get()的时候会重新新触发spring的createBean，创建出一个新的bean对象，新创建的 Bean 就有了我们最新的配置。

// org.springframework.cloud.context.scope.GenericScope#get
public Object get(String name, ObjectFactory<?> objectFactory) {
    BeanLifecycleWrapper value = this.cache.put(name,
            new BeanLifecycleWrapper(name, objectFactory));
    this.locks.putIfAbsent(name, new ReentrantReadWriteLock());
    try {
        // 会重新触发spring的createBean，创建出一个新的bean对象，填充进去的属性就是最新配置的内容
        return value.getBean();
    }
    catch (RuntimeException e) {
        this.errors.put(name, e);
        throw e;
    }
}

1：动态刷新机制

nacos使用的长轮询机制【直到超时或者有变更才返回】，向nacos server查询配置更新的API

然后通过判断配置有没有变化进行比较从而判断是否有更新。

而如果发现差异，就说明nacos server有配置更新，就将nacos server的最新配置拉取下来

在上面这个场景中，有时候可能因为客户端的配置比较多，导致比较的时间比较长，使得配置的同步效率非常的低，于是nacos针对这样一个场景，进行了优化

主要进行了两点优化：

• 减少网络通信的数据量，客户端将需要进行比较的配置进行分片，每一个分片的大小是3000，每一次最多拿3000个配置，去nacos server比较
• 分阶段进行比较和更新

客户端将这3000个配置的key以及对应的value的md5拼接成为一个字符串，然后发送给nacos server端进行判断，服务端会依次比较不同的md5的key，将存在更新的key返回给客户端，客户端拿到这些key之后，循环逐个去调用服务单获取这些key的value值。

这两个优化的核心的目的就是减少网络通信中数据包的大小，将一个大的数据包的通信拆分成多个小的数据包进行通信，虽然会增加网络通信的次数，但是对于整个性能的提升还是非常的大的，最后，再加上长轮询的方式，既减少了pull的轮询次数，又利用了长轮询的优势，很好的实现爱你了配置的动态更新的一个同步功能。

2：在Nacos中配置更新的手段

• web界面 -> 可以登录到Nacos的控制台界面，找到对应的配置项，进行编辑的一个保存，这种方式非常的简单直观，适合手动操作
• API -> nacos提供了一套配置管理相关的API，可以通过编程的方式更新配置，在适合自动化的脚本或者集成到CI/CD流中
• 命令行工具 -> 如果有提供，可以使用命令行工具来管理nacos配置

3：应用中接收配置的手段

• 自动刷新：在支持自动配置刷新的应用框架中(比如spring cloud)，可以使用@RefreshScope或者相似的机制来标记需要动态更新的配置Bean，当配置发生变更的时候，Nacos客户端会自动更新这些bean的状态。
• 监听器：可以在应用中注册监听器来监听特定配置的变更事件，当配置发生变更的时候，Nacos客户端会通知这些监听器，从而可以编写自定义的逻辑来处理这些变更。
• 长轮询：Nacos客户端使用长轮询的机制来检测配置变更，当你通过web界面或者API更新了配置，Nacos会通知所有监听该配置的客户端，客户端接到通知之后会获取新的配置并应用。

4：应用配置更新的注意事项

• 安全和权限：确保只有授权用户能够更改配置，避免未经授权的配置变更导致的风险。
• 版本控制和备份：在进行配置更改之前，确保配置有备份，并且可以追踪不同版本，以便于必要时进行回滚操作。
• 灰度发布：在更新关键配置时，可以使用nacos的灰度发布功能，逐渐将变更推送给部分服务实例，从而降低风险。
• 变更通知：确保相关的开发和运维团队成员都意识到配置的变更，避免变更引发意外的问题。
• 兼容性检查：更新配置前要确保新配置与应用兼容，尤其是在公共配置环境中，一个配置的变更可能影响多个服务。
• 测试：在将配置推送到生产环境字前，在开发或顶生产环境进行充分的测试。

使用NacoS作为配置中心，可以带来配置管理的灵活性和动态性，但同时也需要合理的策略和实践来确保配置更新的流程既安全又高效。

五：配置发布原理

一般情况下，我们是通过Nacos提供的Web控制台登录，然后通过界面新增配置信息。

后续客户端只要配置了对应的NameSpace，Group，DataId就可以在客户端获取到对应的配置信息。

既然这样，Nacos服务端肯定会存储在Web控制台配置的配置信息。

Web控制台发布配置的入口肯定也是一个controller -> nacos.config.server.controller.ConfigController#publishConfig

@PostMapping
@TpsControl(pointName = "ConfigPublish")
@Secured(action = ActionTypes.WRITE, signType = SignType.CONFIG)
public Boolean publishConfig(HttpServletRequest request, HttpServletResponse response,
        @RequestParam(value = "dataId") String dataId, @RequestParam(value = "group") String group,
        @RequestParam(value = "tenant", required = false, defaultValue = StringUtils.EMPTY) String tenant,
        @RequestParam(value = "content") String content, @RequestParam(value = "tag", required = false) String tag,
        @RequestParam(value = "appName", required = false) String appName,
        @RequestParam(value = "src_user", required = false) String srcUser,
        @RequestParam(value = "config_tags", required = false) String configTags,
        @RequestParam(value = "desc", required = false) String desc,
        @RequestParam(value = "use", required = false) String use,
        @RequestParam(value = "effect", required = false) String effect,
        @RequestParam(value = "type", required = false) String type,
        @RequestParam(value = "schema", required = false) String schema,
        @RequestParam(required = false) String encryptedDataKey) throws NacosException {
    
    String encryptedDataKeyFinal = null;
    // 内容加密
    if (StringUtils.isNotBlank(encryptedDataKey)) {
        encryptedDataKeyFinal = encryptedDataKey;
    } else {
        // 使用到插件化的思想进行加密
        Pair<String, String> pair = EncryptionHandler.encryptHandler(dataId, content);
        content = pair.getSecond();
        encryptedDataKeyFinal = pair.getFirst();
    }
 
    // 参数检查
    ParamUtils.checkTenant(tenant);
    ParamUtils.checkParam(dataId, group, "datumId", content);
    ParamUtils.checkParam(tag);
    // 构造配置信息，包括namespaceId、groupId、dataId、配置内容、描述信息等
    ConfigForm configForm = new ConfigForm();
    configForm.setDataId(dataId);
    configForm.setGroup(group);
    configForm.setNamespaceId(tenant);
    configForm.setContent(content);
    configForm.setTag(tag);
    configForm.setAppName(appName);
    configForm.setSrcUser(srcUser);
    configForm.setConfigTags(configTags);
    configForm.setDesc(desc);
    configForm.setUse(use);
    configForm.setEffect(effect);
    configForm.setType(type);
    configForm.setSchema(schema);
    
    if (StringUtils.isBlank(srcUser)) {
        configForm.setSrcUser(RequestUtil.getSrcUserName(request));
    }
    if (!ConfigType.isValidType(type)) {
        configForm.setType(ConfigType.getDefaultType().getType());
    }
    // 构造请求对象
    ConfigRequestInfo configRequestInfo = new ConfigRequestInfo();
    configRequestInfo.setSrcIp(RequestUtil.getRemoteIp(request));
    configRequestInfo.setRequestIpApp(RequestUtil.getAppName(request));
    configRequestInfo.setBetaIps(request.getHeader("betaIps"));
    // 发布配置
    return configOperationService.publishConfig(configForm, configRequestInfo, encryptedDataKeyFinal);
}

上述的代码主要完成了五件事情：

1：加密处理

加密处理使用了插件化思想。我们分析下插件化的思想，看看是如何使用插件或者扩展来进行加解密的

public static Pair<String, String> encryptHandler(String dataId, String content) {
    // 检查是否需要加密
    if (!checkCipher(dataId)) {
        return Pair.with("", content);
    }
    Optional<String> algorithmName = parseAlgorithmName(dataId);
    // 获取加密的处理类
    // EncryptionPluginManager.instance(): 返回单例实例
    Optional<EncryptionPluginService> optional = algorithmName
        .flatMap(EncryptionPluginManager.instance()::findEncryptionService);
    if (!optional.isPresent()) {
        LOGGER.warn("[EncryptionHandler] [encryptHandler] No encryption program with the corresponding name found");
        // 获取不到，还是走非加密型
        return Pair.with("", content);
    }
    EncryptionPluginService encryptionPluginService = optional.get();
    // 根据扩展的插件类，获取密钥
    String secretKey = encryptionPluginService.generateSecretKey();
    // 利用密钥加密
    String encryptContent = encryptionPluginService.encrypt(secretKey, content);
    return Pair.with(encryptionPluginService.encryptSecretKey(secretKey), encryptContent);
}

首先判断是否需要处理加密，如果需要的话，去插件里面获取对应的处理类，如果获取不到则打日志，然后使用非加密方式进行处理；

获取到加密插件，利用插件获取秘钥，然后再加密

我们来分析下如何获取加密处理类的：

Optional<EncryptionPluginService> optional = 
    algorithmName.flatMap(EncryptionPluginManager.instance()::findEncryptionService)

这个EncryptionPluginManager.instance()执行返回的是一个单例对象，看看它的构造方法：

private EncryptionPluginManager() {
    // 初始化: 根据自己写的扩展机制，获取EncryptionPluginService，然后再进行反射初始化。
    loadInitial();
}
 
private void loadInitial() {
    // 通过NacosServiceLoader扩展机制，获取EncryptionPluginService加密处理类的集合
    Collection<EncryptionPluginService> encryptionPluginServices = NacosServiceLoader.load(
            EncryptionPluginService.class);
    for (EncryptionPluginService encryptionPluginService : encryptionPluginServices) {
        if (StringUtils.isBlank(encryptionPluginService.algorithmName())) {
            LOGGER.warn("[EncryptionPluginManager] Load EncryptionPluginService({}) algorithmName(null/empty) fail."
                    + " Please Add algorithmName to resolve.", encryptionPluginService.getClass());
            continue;
        }
        // 放入集合
        ENCRYPTION_SPI_MAP.put(encryptionPluginService.algorithmName(), encryptionPluginService);
        LOGGER.info("[EncryptionPluginManager] Load EncryptionPluginService({}) algorithmName({}) successfully.",
                encryptionPluginService.getClass(), encryptionPluginService.algorithmName());
    }
}

因为是单例，所以获取单例的时候通过loadInitial()进行初始化，初始化的时候会根据自己写的扩展机制

获取EncryptionPluginService加密处理类集合，然后再进行反射初始化，并缓存起来。

重要的还是这种插件化的思想，它仅仅依赖于原生JDK的SPI机制，可以按需扩展和定制：

1. 提供给插件化的接口，由第三方去实现（自定义功能）
2. 在初始化的时候，Nacos去加载处理类

2：发布配置

发布配置调用的是ConfigOperationService#publishConfig()方法

public Boolean publishConfig(ConfigForm configForm, ConfigRequestInfo configRequestInfo, String encryptedDataKey)
        throws NacosException {
    // 将配置高级信息转成Map键值对
    Map<String, Object> configAdvanceInfo = getConfigAdvanceInfo(configForm);
    // 检查参数
    ParamUtils.checkParam(configAdvanceInfo);
    
    if (AggrWhitelist.isAggrDataId(configForm.getDataId())) {
        LOGGER.warn("[aggr-conflict] {} attempt to publish single data, {}, {}", configRequestInfo.getSrcIp(),
                configForm.getDataId(), configForm.getGroup());
        throw new NacosApiException(HttpStatus.FORBIDDEN.value(), ErrorCode.INVALID_DATA_ID,
                "dataId:" + configForm.getDataId() + " is aggr");
    }
 
    // 构建ConfigInfo配置信息，发布配置最基本的五个参数： nameSpaceId、groupId、dataId、应用名称、配置内容
    ConfigInfo configInfo = new ConfigInfo(configForm.getDataId(), configForm.getGroup(),
            configForm.getNamespaceId(), configForm.getAppName(), configForm.getContent());
    
    configInfo.setType(configForm.getType());
    configInfo.setEncryptedDataKey(encryptedDataKey);
    ConfigOperateResult configOperateResult = null;
    
    String persistEvent = ConfigTraceService.PERSISTENCE_EVENT;
 
    // 判断是否是beta测试版本
    if (StringUtils.isBlank(configRequestInfo.getBetaIps())) {
        // 正常发布，大部分情况下，我们都没有指定tag
        if (StringUtils.isBlank(configForm.getTag())) {
            // 1、插入 or 更新配置信息
            // 这里分为内置数据库（EmbeddedConfigInfoPersistServiceImpl）和外置数据库（ExternalConfigInfoPersistServiceImpl）操作，通常我们都是使用MySQL进行持久化存储
            configOperateResult = configInfoPersistService.insertOrUpdate(configRequestInfo.getSrcIp(),
                    configForm.getSrcUser(), configInfo, configAdvanceInfo);
 
            // 2、发布配置数据变动事件
            ConfigChangePublisher.notifyConfigChange(
                    new ConfigDataChangeEvent(false, configForm.getDataId(), configForm.getGroup(),
                            configForm.getNamespaceId(), configOperateResult.getLastModified()));
        } else {
            // 指定tag
            persistEvent = ConfigTraceService.PERSISTENCE_EVENT_TAG + "-" + configForm.getTag();
            configOperateResult = configInfoTagPersistService.insertOrUpdateTag(configInfo, configForm.getTag(),
                    configRequestInfo.getSrcIp(), configForm.getSrcUser());
            ConfigChangePublisher.notifyConfigChange(
                    new ConfigDataChangeEvent(false, configForm.getDataId(), configForm.getGroup(),
                            configForm.getNamespaceId(), configForm.getTag(),
                            configOperateResult.getLastModified()));
        }
    } else {
        persistEvent = ConfigTraceService.PERSISTENCE_EVENT_BETA;
        
        // beta publish
        configOperateResult = configInfoBetaPersistService.insertOrUpdateBeta(configInfo,
                configRequestInfo.getBetaIps(), configRequestInfo.getSrcIp(), configForm.getSrcUser());
        ConfigChangePublisher.notifyConfigChange(
                new ConfigDataChangeEvent(true, configForm.getDataId(), configForm.getGroup(),
                        configForm.getNamespaceId(), configOperateResult.getLastModified()));
    }
    // 日志跟踪
    ConfigTraceService.logPersistenceEvent(configForm.getDataId(), configForm.getGroup(),
            configForm.getNamespaceId(), configRequestInfo.getRequestIpApp(), configOperateResult.getLastModified(),
            InetUtils.getSelfIP(), persistEvent, ConfigTraceService.PERSISTENCE_TYPE_PUB, configForm.getContent());
    
    return true;
}

首先组装好一些参数，我们需要重点关注的是构建ConfigInfo配置信息

发布配置最基本的五个参数： nameSpaceId、groupId、dataId、应用名称、配置内容。

然后包含一些测试版本和tag的分支逻辑判断，我们关注最常用的正常发布流程。

通常情况下，我们发布配置，都不指定tag，其实就做了两件事：

1. 插入或更新配置信息
2. 发布配置数据变动事件

2.1：插入或更新配置信息

插入或更新配置信息，其实就是操作数据库，数据库操作分为了内置数据库和外置数据库

我们通常使用外置数据库MySQL来存储配置信息，也就是ExternalConfigInfoPersistServiceImpl

内置数据库对应的操作类是EmbeddedConfigInfoPersistServiceImpl。

这里主要分析外置数据库MySQL的方式：ExternalConfigInfoPersistServiceImpl#insertOrUpdate()

public ConfigOperateResult insertOrUpdate(String srcIp, String srcUser, ConfigInfo configInfo,
                                          Map<String, Object> configAdvanceInfo) {
    // 没有直接判断是新增还是更新，而且依赖数据库唯一性做检查，重复了（报主键冲突，说明已存在）就做更新。
    try {
        // 添加配置信息
        return addConfigInfo(srcIp, srcUser, configInfo, configAdvanceInfo);
    } catch (DuplicateKeyException ive) { // Unique constraint conflict
        // 如果报唯一约束冲突，则更新配置内容
        return updateConfigInfo(configInfo, srcIp, srcUser, configAdvanceInfo);
    }
}

从源码可以看到，这里没有直接判断是新增还是更新配置，而且依赖数据库唯一性做检查，重复了（报主键冲突，说明已存在）就做更新

public ConfigOperateResult addConfigInfo(final String srcIp, final String srcUser, final ConfigInfo configInfo,
        final Map<String, Object> configAdvanceInfo) {
    return tjt.execute(status -> {
        try {
            // jdbcTemplate操作，自动插入到数据库表（config_info）中，返回主键id
            long configId = addConfigInfoAtomic(-1, srcIp, srcUser, configInfo, configAdvanceInfo);
            String configTags = configAdvanceInfo == null ? null : (String) configAdvanceInfo.get("config_tags");
            // 新增tag管理
            addConfigTagsRelation(configId, configTags, configInfo.getDataId(), configInfo.getGroup(),
                    configInfo.getTenant());
            Timestamp now = new Timestamp(System.currentTimeMillis());
            // 插入历史数据到表中（his_config_info）
            historyConfigInfoPersistService.insertConfigHistoryAtomic(0, configInfo, srcIp, srcUser, now, "I");
            ConfigInfoStateWrapper configInfoCurrent = this.findConfigInfoState(configInfo.getDataId(),
                    configInfo.getGroup(), configInfo.getTenant());
            if (configInfoCurrent == null) {
                return new ConfigOperateResult(false);
            }
            return new ConfigOperateResult(configInfoCurrent.getId(), configInfoCurrent.getLastModified());
            
        } catch (CannotGetJdbcConnectionException e) {
            LogUtil.FATAL_LOG.error("[db-error] " + e, e);
            throw e;
        }
    });
}

插入数据库的操作是在addConfigInfoAtomic()方法：

public long addConfigInfoAtomic(final long configId, final String srcIp, final String srcUser,
        final ConfigInfo configInfo, Map<String, Object> configAdvanceInfo) {
    // 取出配置信息
    final String appNameTmp =
            StringUtils.isBlank(configInfo.getAppName()) ? StringUtils.EMPTY : configInfo.getAppName();
    final String tenantTmp =
            StringUtils.isBlank(configInfo.getTenant()) ? StringUtils.EMPTY : configInfo.getTenant();
    
    final String desc = configAdvanceInfo == null ? null : (String) configAdvanceInfo.get("desc");
    final String use = configAdvanceInfo == null ? null : (String) configAdvanceInfo.get("use");
    final String effect = configAdvanceInfo == null ? null : (String) configAdvanceInfo.get("effect");
    final String type = configAdvanceInfo == null ? null : (String) configAdvanceInfo.get("type");
    final String schema = configAdvanceInfo == null ? null : (String) configAdvanceInfo.get("schema");
    final String encryptedDataKey =
            configInfo.getEncryptedDataKey() == null ? StringUtils.EMPTY : configInfo.getEncryptedDataKey();
 
    // 将配置内容进行MD5加密
    final String md5Tmp = MD5Utils.md5Hex(configInfo.getContent(), Constants.ENCODE);
    
    KeyHolder keyHolder = new GeneratedKeyHolder();
 
    // 根据数据库表获取对应的mapper, 通过插件化的形式, 灵活应对使用不同数据库的场景
    ConfigInfoMapper configInfoMapper = mapperManager.findMapper(dataSourceService.getDataSourceType(),
            TableConstant.CONFIG_INFO);
    // 将参数转换成对应数据库类型的sql语句，拼接insert into config_info values(....)插入语句
    final String sql = configInfoMapper.insert(
            Arrays.asList("data_id", "group_id", "tenant_id", "app_name", "content", "md5", "src_ip", "src_user",
                    "gmt_create", "gmt_modified", "c_desc", "c_use", "effect", "type", "c_schema",
                    "encrypted_data_key"));
    // 获取主键名称，默认值为id
    String[] returnGeneratedKeys = configInfoMapper.getPrimaryKeyGeneratedKeys();
    try {
        jt.update(new PreparedStatementCreator() {
            @Override
            public PreparedStatement createPreparedStatement(Connection connection) throws SQLException {
                Timestamp now = new Timestamp(System.currentTimeMillis());
 
                // 通过预编译的PreparedStatement，设置每个字段的值
                PreparedStatement ps = connection.prepareStatement(sql, returnGeneratedKeys);
                ps.setString(1, configInfo.getDataId());
                ps.setString(2, configInfo.getGroup());
                ps.setString(3, tenantTmp);
                ps.setString(4, appNameTmp);
                ps.setString(5, configInfo.getContent());
                ps.setString(6, md5Tmp);
                ps.setString(7, srcIp);
                ps.setString(8, srcUser);
                ps.setTimestamp(9, now);
                ps.setTimestamp(10, now);
                ps.setString(11, desc);
                ps.setString(12, use);
                ps.setString(13, effect);
                ps.setString(14, type);
                ps.setString(15, schema);
                ps.setString(16, encryptedDataKey);
                return ps;
            }
        }, keyHolder);
        Number nu = keyHolder.getKey();
        if (nu == null) {
            throw new IllegalArgumentException("insert config_info fail");
        }
        return nu.longValue();
    } catch (CannotGetJdbcConnectionException e) {
        LogUtil.FATAL_LOG.error("[db-error] " + e, e);
        throw e;
    }
}

首先取出配置信息，对配置的内容进行MD5加密，然后根据数据库表获取对应的mapper，这里还是通过插件化的形式，灵活应对使用不同数据库的场景。

获取到mapper之后，将参数转换成对应数据库类型的sql语句，拼接insert into config_info values(....)插入语句

最后通过JdbcTemplate执行sql，完成配置的插入。

我们再来分析下如何利用插件化思想完成对mapper的获取的：

ConfigInfoMapper configInfoMapper = mapperManager.findMapper(dataSourceService.getDataSourceType(),
                TableConstant.CONFIG_INFO);
 
public <R extends Mapper> R findMapper(String dataSource, String tableName) {
    LOGGER.info("[MapperManager] findMapper dataSource: {}, tableName: {}", dataSource, tableName);
    if (StringUtils.isBlank(dataSource) || StringUtils.isBlank(tableName)) {
        throw new NacosRuntimeException(FIND_DATASOURCE_ERROR_CODE, "dataSource or tableName is null");
    }
    // 从SPI缓存中获取，这个是在MapperManager构造方法中初始化的
    Map<String, Mapper> tableMapper = MAPPER_SPI_MAP.get(dataSource);
    if (Objects.isNull(tableMapper)) {
        throw new NacosRuntimeException(FIND_DATASOURCE_ERROR_CODE,
                "[MapperManager] Failed to find the datasource,dataSource:" + dataSource);
    }
    // 根据表名称获取mapper
    Mapper mapper = tableMapper.get(tableName);
    if (Objects.isNull(mapper)) {
        throw new NacosRuntimeException(FIND_TABLE_ERROR_CODE,
                "[MapperManager] Failed to find the table ,tableName:" + tableName);
    }
    if (dataSourceLogEnable) {
        return (R) MapperProxy.createSingleProxy(mapper);
    }
    return (R) mapper;
}

首先从MAPPER_SPI_MAP缓存中获取，这个是在MapperManager构造方法中初始化的。然后根据表名称获取到对应的mapper。

这个MAPPER_SPI_MAP初始化也和之前EncryptionPluginService的一样，在单例的构造方法中加载：

private MapperManager() {
    loadInitial();
}
 
public void loadInitial() {
    Collection<Mapper> mappers = NacosServiceLoader.load(Mapper.class);
    for (Mapper mapper : mappers) {
        Map<String, Mapper> mapperMap = MAPPER_SPI_MAP.computeIfAbsent(mapper.getDataSource(), (r) -> new HashMap<>(16));
        mapperMap.put(mapper.getTableName(), mapper);
        LOGGER.info("[MapperManager] Load Mapper({}) datasource({}) tableName({}) successfully.",
                mapper.getClass(), mapper.getDataSource(), mapper.getTableName());
    }
}

获取到插件配置的具体mapper实现类后，在调用mapper.Mapper#insert()方法时，就可以根据插件的扩展

通过不同的实现类去处理了，就能解决不同数据库类型中sql存在差异的问题。

更新配置的大体流程跟新增一样，首先查出旧的配置信息，然后做一些判断，最后根据dataType和表名称获取对应的mapper

然后组装好sql，通过JdbcTemplate执行。

2.2：发布配置数据变动事件

ConfigChangePublisher.notifyConfigChange(
    new ConfigDataChangeEvent(false, configForm.getDataId(), configForm.getGroup(), configForm.getNamespaceId(), configOperateResult.getLastModified()));

public static void notifyConfigChange(ConfigDataChangeEvent event) {
    // 如果是内部存储并且Nacos非单机模式启动，就不处理了
    if (DatasourceConfiguration.isEmbeddedStorage() && !EnvUtil.getStandaloneMode()) {
        return;
    }
    NotifyCenter.publishEvent(event);
}

可以看到，还是利用了Nacos的事件统一发布中心NotifyCenter类，我们直接查找ConfigDataChangeEvent的onEvent方法来查看处理逻辑。

public AsyncNotifyService(ServerMemberManager memberManager) {
    this.memberManager = memberManager;
    
    // 注册ConfigDataChangeEvent到NotifyCenter.
    NotifyCenter.registerToPublisher(ConfigDataChangeEvent.class, NotifyCenter.ringBufferSize);
    
    // 注册一个订阅ConfigDataChangeEvent事件的处理类
    NotifyCenter.registerSubscriber(new Subscriber() {
        
        @Override
        public void onEvent(Event event) {
            // Generate ConfigDataChangeEvent concurrently
            if (event instanceof ConfigDataChangeEvent) {
                ConfigDataChangeEvent evt = (ConfigDataChangeEvent) event;
                long dumpTs = evt.lastModifiedTs;
                String dataId = evt.dataId;
                String group = evt.group;
                String tenant = evt.tenant;
                String tag = evt.tag;
                MetricsMonitor.incrementConfigChangeCount(tenant, group, dataId);
 
                // 获取所有的Nacos服务节点（包括当前客户端）
                Collection<Member> ipList = memberManager.allMembers();
                
                // 创建一个队列，将相关配置的其他服务节点都存放进来
                Queue<NotifySingleRpcTask> rpcQueue = new LinkedList<>();
                
                for (Member member : ipList) {
                    // grpc report data change only
                    rpcQueue.add(
                            new NotifySingleRpcTask(dataId, group, tenant, tag, dumpTs, evt.isBeta, evt.isBatch,
                                    member));
                }
                if (!rpcQueue.isEmpty()) {
                    // 通过线程池执行异步通知
                    // AsyncRpcTask实现了runnable接口，关注其run方法
                    ConfigExecutor.executeAsyncNotify(new AsyncRpcTask(rpcQueue));
                }
                
            }
        }
        
        @Override
        public Class<? extends Event> subscribeType() {
            return ConfigDataChangeEvent.class;
        }
    });
}

在AsyncNotifyService构造方法中，将ConfigDataChangeEvent事件注册到NotifyCenter通知中心

然后还注册一个订阅ConfigDataChangeEvent事件的处理类。

AsyncNotifyService使用spring进行托管，在IOC容器启动的时候，就会创建这个bean对象，就会执行AsyncNotifyService构造方法。

重点关注onEvent()具体的事件处理逻辑：

1. 获取所有的Nacos服务节点（包括当前客户端）
2. 创建一个队列，将相关配置的其他服务节点都存放进来
3. 通过线程池执行异步通知

// 获取所有的Nacos服务节点（包括当前客户端）
Collection<Member> ipList = memberManager.allMembers();
 
// 创建一个队列，将相关配置的其他服务节点都存放进来
Queue<NotifySingleRpcTask> rpcQueue = new LinkedList<>();
 
for (Member member : ipList) {
    // grpc report data change only
    rpcQueue.add(
            new NotifySingleRpcTask(dataId, group, tenant, tag, dumpTs, evt.isBeta, evt.isBatch,
                    member));
}
if (!rpcQueue.isEmpty()) {
    // 通过线程池执行异步通知
    // AsyncRpcTask实现了runnable接口，关注其run方法
    ConfigExecutor.executeAsyncNotify(new AsyncRpcTask(rpcQueue));
}

获取到服务列表后，通过线程池调用异步任务AsyncRpcTask，AsyncRpcTask实现了Runnable接口，看看run()的逻辑：

class AsyncRpcTask implements Runnable {
    
    private Queue<NotifySingleRpcTask> queue;
    
    public AsyncRpcTask(Queue<NotifySingleRpcTask> queue) {
        // 构造方法放入rpcTask的队列
        this.queue = queue;
    }
    
    @Override
    public void run() {
        while (!queue.isEmpty()) {
            // 从队列中取出任务
            NotifySingleRpcTask task = queue.poll();
            // 构造配置变动集群同步请求
            ConfigChangeClusterSyncRequest syncRequest = new ConfigChangeClusterSyncRequest();
            syncRequest.setDataId(task.getDataId());
            syncRequest.setGroup(task.getGroup());
            syncRequest.setBeta(task.isBeta);
            syncRequest.setLastModified(task.getLastModified());
            syncRequest.setTag(task.tag);
            syncRequest.setBatch(task.isBatch);
            syncRequest.setTenant(task.getTenant());
 
            // 通知的目标节点
            Member member = task.member;
            // 如果是当前节点，直接调用dumpService执行dump操作
            if (memberManager.getSelf().equals(member)) {
                if (syncRequest.isBeta()) {
                    dumpService.dumpBeta(syncRequest.getDataId(), syncRequest.getGroup(), syncRequest.getTenant(),
                            syncRequest.getLastModified(), NetUtils.localIP());
                } else if (syncRequest.isBatch()) {
                    dumpService.dumpBatch(syncRequest.getDataId(), syncRequest.getGroup(), syncRequest.getTenant(),
                            syncRequest.getLastModified(), NetUtils.localIP());
                } else if (StringUtils.isNotBlank(syncRequest.getTag())) {
                    dumpService.dumpTag(syncRequest.getDataId(), syncRequest.getGroup(), syncRequest.getTenant(),
                            syncRequest.getTag(), syncRequest.getLastModified(), NetUtils.localIP());
                } else {
                    dumpService.dumpFormal(syncRequest.getDataId(), syncRequest.getGroup(), syncRequest.getTenant(),
                            syncRequest.getLastModified(), NetUtils.localIP());
                }
                continue;
            }
            String event = getNotifyEvent(task);
            
            if (memberManager.hasMember(member.getAddress())) {
                // 启动健康检查，有IP未被监控，直接放入通知队列，否则通知
                boolean unHealthNeedDelay = isUnHealthy(member.getAddress());
                if (unHealthNeedDelay) {
                    // 目标 IP 运行状况不健康，然后将其放入通知列表中
                    ConfigTraceService.logNotifyEvent(task.getDataId(), task.getGroup(), task.getTenant(), null,
                            task.getLastModified(), InetUtils.getSelfIP(), event,
                            ConfigTraceService.NOTIFY_TYPE_UNHEALTH, 0, member.getAddress());
 
                    // 异步任务执行
                    // 可延迟的处理，因为是不健康的节点，不知道什么时候能恢复
                    asyncTaskExecute(task);
                } else {
                    // 发送grpc请求
                    try {
                        configClusterRpcClientProxy.syncConfigChange(member, syncRequest,
                                new AsyncRpcNotifyCallBack(task));
                    } catch (Exception e) {
                        MetricsMonitor.getConfigNotifyException().increment();
                        asyncTaskExecute(task);
                    }
                    
                }
            } else {
                //No nothing if  member has offline.
            }
            
        }
    }
}

只要队列不为空，就会从队列中取出NotifySingleRpcTask任务来执行

然后构造配置变动集群同步的请求对象，包括namespaceId、dataId、groupId、标签等，然后通知目标节点。

目标节点是当前节点

如果目标节点是当前节点，则会直接调用dumpService执行dump操作，其实就是更新本地内存和磁盘中的配置信息为最新的配置信息。

public void dumpFormal(String dataId, String group, String tenant, long lastModified, String handleIp) {
    String groupKey = GroupKey2.getKey(dataId, group, tenant);
    String taskKey = dataId + group + tenant;
    // 将DumpTask添加到TaskManager任务管理器，它将异步执行
    dumpTaskMgr.addTask(taskKey, new DumpTask(groupKey, false, false, false, null, lastModified, handleIp));
    DUMP_LOG.info("[dump] add formal task. groupKey={}", groupKey);
}

目标节点是其他节点

如果目标节点是其它节点，还会区分是否健康实例，进行不同的处理。

• 如果节点不健康，则会延迟处理同步的逻辑；

// 节点不健康的情况，采用异步定时任务去执行
// 但是这个定时并不是严格意义的定时，因为他会有个延迟的过程，会随着失败次数的增加，延迟不断加大
// 不过当达到最大失败次数后，就不会再增加，以一个固定的时间去触发。
// 最大时间间隔是500ms + 7 * 7 * 1000ms。
private void asyncTaskExecute(NotifySingleRpcTask task) {
    // 随着失败次数的增加，延迟不断加大，不过当达到最大失败次数后，就不会再增加，以一个固定的时间去触发。最大时间间隔是500ms + 7 * 7 * 1000ms
    int delay = getDelayTime(task);
    Queue<NotifySingleRpcTask> queue = new LinkedList<>();
    queue.add(task);
    AsyncRpcTask asyncTask = new AsyncRpcTask(queue);
    ConfigExecutor.scheduleAsyncNotify(asyncTask, delay, TimeUnit.MILLISECONDS);
}
 
private static int getDelayTime(NotifyTask task) {
    int failCount = task.getFailCount();
    // 最大时间间隔是500ms + 7 * 7 * 1000ms
    int delay = MIN_RETRY_INTERVAL + failCount * failCount * INCREASE_STEPS;
    if (failCount <= MAX_COUNT) {
        task.setFailCount(failCount + 1);
    }
    return delay;
}

• 如果节点健康，则会发送grpc请求，目标服务进行配置的同步。

configClusterRpcClientProxy.syncConfigChange(member, syncRequest,
                                new AsyncRpcNotifyCallBack(task));
 
public void syncConfigChange(Member member, ConfigChangeClusterSyncRequest request, RequestCallBack callBack)
        throws NacosException {
    // 异步处理
    // grpc真正的处理类是在：com.alibaba.nacos.config.server.remote.ConfigChangeClusterSyncRequestHandler.handle
    clusterRpcClientProxy.asyncRequest(member, request, callBack);
}

public ConfigChangeClusterSyncResponse handle(ConfigChangeClusterSyncRequest configChangeSyncRequest,
        RequestMeta meta) throws NacosException {
    // 调用到其他节点，其他节点也是执行dump服务，然后通知和本机连接的客户端，通知他们进行配置更新。
    if (configChangeSyncRequest.isBeta()) {
        dumpService.dumpBeta(configChangeSyncRequest.getDataId(), configChangeSyncRequest.getGroup(),
                configChangeSyncRequest.getTenant(), configChangeSyncRequest.getLastModified(), meta.getClientIp());
    } else if (configChangeSyncRequest.isBatch()) {
        dumpService.dumpBatch(configChangeSyncRequest.getDataId(), configChangeSyncRequest.getGroup(),
                configChangeSyncRequest.getTenant(), configChangeSyncRequest.getLastModified(), meta.getClientIp());
    } else if (StringUtils.isNotBlank(configChangeSyncRequest.getTag())) {
        dumpService.dumpTag(configChangeSyncRequest.getDataId(), configChangeSyncRequest.getGroup(),
                configChangeSyncRequest.getTenant(), configChangeSyncRequest.getTag(),
                configChangeSyncRequest.getLastModified(), meta.getClientIp());
    } else {
		// 本机的dump服务
        dumpService.dumpFormal(configChangeSyncRequest.getDataId(), configChangeSyncRequest.getGroup(),
                configChangeSyncRequest.getTenant(), configChangeSyncRequest.getLastModified(), meta.getClientIp());
    }
    return new ConfigChangeClusterSyncResponse();
}