微服务架构 | *2.4 Nacos 获取配置与事件订阅机制的源码分析

前言
1. 客户端获取 Nacos 服务器里的配置
- 1.1 定位 Nacos 配置源 NacosPropertySourceLocator.locate()
2. Nacos 配置的事件订阅机制
- 2.1 监听 ApplicationReadyEvent 事件，注册监听器 NacosContextRefresher.onApplicationEvent()
- 2.2 注册 Nacos 监听器，监听配置变更 NacosContextRefresher.registerNacosListener()
- 2.3 监听配置变更，实施变更 RefreshEventListener.handle()
3. 源码结构图小结
- 3.1 客户端获取 Nacos 服务器上的配置源码结构图
- 3.2 Nacos 配置的事件订阅机制
最后

参考资料：

《Spring Microservices in Action》

《Spring Cloud Alibaba 微服务原理与实战》

《B站尚硅谷 SpringCloud 框架开发教程周阳》

为方便理解与表达，这里把 Nacos 控制台和 Nacos 注册中心称为 Nacos 服务器（就是 web 界面那个），我们编写的业务服务称为 Nacso 客户端；

由于篇幅有限，这里将源码分析分为上下两篇，其中上篇讲获取配置与事件订阅机制，下篇讲长轮询定时机制；

上篇《微服务架构 | *2.3 Spring Cloud 启动及加载配置文件源码分析（以 Nacos 为例）》中提到，读取 Nacos 服务器里的配置依靠的是

NacosPropertySourceLocator.locate()

方法，我们这次的源码之旅将从这个方法开始；

该方法的主要作用是：
- 初始化 ConfigService 对象，这是 Nacos 客户端提供的用于访问实现配置中心基本操作的类；
- 按照顺序分别加载共享配置、扩展配置、应用名称对应的配置；
方法源码如下：

@Override
public PropertySource<?> locate(Environment env) {
    //【断点步入 长轮询定时机制】获取配置服务器实例，这是 Nacos 客户端提供的用于访问实现配置中心基本操作的类
	ConfigService configService = nacosConfigProperties.configServiceInstance();
	if (null == configService) {
		log.warn("no instance of config service found, can't load config from nacos");
		return null;
	}
	long timeout = nacosConfigProperties.getTimeout();
	//Nacos 属性源生成器
	nacosPropertySourceBuilder = new NacosPropertySourceBuilder(configService, timeout);
	String name = nacosConfigProperties.getName();

    //DataId 前缀（这里是 nacos-config-client）
	String dataIdPrefix = nacosConfigProperties.getPrefix();
	if (StringUtils.isEmpty(dataIdPrefix)) {
		dataIdPrefix = name;
	}
    //没有配置 DataId 前缀则用 spring.application.name 属性的值
	if (StringUtils.isEmpty(dataIdPrefix)) {
		dataIdPrefix = env.getProperty("spring.application.name");
	}
    //创建复合属性源
	CompositePropertySource composite = new CompositePropertySource(NACOS_PROPERTY_SOURCE_NAME);

    //加载共享配置
	loadSharedConfiguration(composite);
	//加载外部配置
	loadExtConfiguration(composite);
	//【断点步入】加载 Nacos 服务器上应用程序名对应的的配置
	loadApplicationConfiguration(composite, dataIdPrefix, nacosConfigProperties, env);

	return composite;
}

进入 NacosPropertySourceLocator.loadApplicationConfiguration() 方法，根据 Data ID 加载配置；

private void loadApplicationConfiguration(CompositePropertySource compositePropertySource, String dataIdPrefix, NacosConfigProperties properties, Environment environment) {
    //获取配置格式（这里是 yaml）
	String fileExtension = properties.getFileExtension();
	//获取nacosGroup（这里是 DEFAULT_GROUP）
	String nacosGroup = properties.getGroup();
    //如果我们配置了前缀，则按前缀获取配置文件（由于我们没配置前缀，这里是 nacos-config-client.yaml 获取不到）
	loadNacosDataIfPresent(compositePropertySource, dataIdPrefix + DOT + fileExtension, nacosGroup, fileExtension, true);
	for (String profile : environment.getActiveProfiles()) {
	    //这里是 nacos-config-client-dev.yaml 可以获取
		String dataId = dataIdPrefix + SEP1 + profile + DOT + fileExtension;
		//【断点步入】加载配置
		loadNacosDataIfPresent(compositePropertySource, dataId, nacosGroup, fileExtension, true);
	}
}

NacosPropertySourceLocator.loadNacosDataIfPresent() 方法，判断是否更新情况；

private void loadNacosDataIfPresent(final CompositePropertySource composite, final String dataId, final String group, String fileExtension, boolean isRefreshable) {
    //获取更新的配置，需要注意，NacosContextRefresher 类与事件订阅机制相关，本篇第2点将重点讨论
	if (NacosContextRefresher.getRefreshCount() != 0) {
		NacosPropertySource ps;
		if (!isRefreshable) {
			ps = NacosPropertySourceRepository.getNacosPropertySource(dataId);
		}
		else {
			ps = nacosPropertySourceBuilder.build(dataId, group, fileExtension, true);
		}

		composite.addFirstPropertySource(ps);
	}
	else {
	    //【断点步入】如果我们没有更新配置，则走下面代码
		NacosPropertySource ps = nacosPropertySourceBuilder.build(dataId, group,fileExtension, isRefreshable);
		composite.addFirstPropertySource(ps);
	}
}

NacosPropertySourceBuilder.build() 方法，加载并封装配置；

NacosPropertySource build(String dataId, String group, String fileExtension, boolean isRefreshable) {
    //【断点步入】加载 Nacos 
	Properties p = loadNacosData(dataId, group, fileExtension);
	//将获取到的配置封装到 NacosPropertySource
	NacosPropertySource nacosPropertySource = new NacosPropertySource(group, dataId, propertiesToMap(p), new Date(), isRefreshable);	
	NacosPropertySourceRepository.collectNacosPropertySources(nacosPropertySource);
	return nacosPropertySource;
}

NacosPropertySourceBuilder.loadNacosData() 方法，

private Properties loadNacosData(String dataId, String group, String fileExtension) {
	
    //省略其他代码

	try {
	    //【断点步入】根据 dataId、group 等信息获取配置
		data = configService.getConfig(dataId, group, timeout);
		
	}
	catch (NacosException e) {
		log.error("get data from Nacos error,dataId:{}, ", dataId, e);
	}
	return EMPTY_PROPERTIES;
}

一直追下去发现在 NacosConfigService.getConfigInner() 方法里成功获取到配置；

上下文准备完毕后，程序运行，EventPublishingRunListener 发布 ApplicationReadyEvent 事件，详情请见《微服务架构 | *2.3 Spring Cloud 启动及加载配置文件源码分析（以 Nacos 为例）》中的《4. 程序运行事件》；
上面 1.1 提到，事件订阅机制与 NacosContextRefresher（Nacos上下文更新器）相关，这是因为里面有个 NacosContextRefresher.onApplicationEvent() 方法实现了对事件 ApplicationReadyEvent（上下文准备完毕事件）的监听，源码如下：

//监听 ApplicationReadyEvent 事件
@Override
public void onApplicationEvent(ApplicationReadyEvent event) {
	if (this.ready.compareAndSet(false, true)) {
	    //【断点步入】为应用程序注册 Nacos 监听器
		this.registerNacosListenersForApplications();
	}
}

当监听到 ApplicationReadyEvent 事件后，最终会调用 NacosContextRefresher.registerNacosListener() 方法来实现 Nacos 监听器的注册，源码如下：

private void registerNacosListener(final String group, final String dataId) {

	Listener listener = listenerMap.computeIfAbsent(dataId, i -> new Listener() {
	    //接受配置变更的回调
		@Override
		public void receiveConfigInfo(String configInfo) {
			refreshCountIncrement();
			String md5 = "";
			if (!StringUtils.isEmpty(configInfo)) {
				try {
					MessageDigest md = MessageDigest.getInstance("MD5");
					md5 = new BigInteger(1, md.digest(configInfo.getBytes("UTF-8"))).toString(16);
				}
				catch (NoSuchAlgorithmException | UnsupportedEncodingException e) {
					log.warn("[Nacos] unable to get md5 for dataId: " + dataId, e);
				}
			}
			refreshHistory.add(dataId, md5);
			//发布 RefreshEvent 配置变更事件
			applicationContext.publishEvent(new RefreshEvent(this, null, "Refresh Nacos config"));
			if (log.isDebugEnabled()) {
				log.debug("Refresh Nacos config group " + group + ",dataId" + dataId);
			}
		}

		@Override
		public Executor getExecutor() {
			return null;
		}
	});

	try {
	    //监听配置
		configService.addListener(dataId, group, listener);
	}
	catch (NacosException e) {
		e.printStackTrace();
	}
}

当收到配置变更的回调时，会通过 applicationContext.publishEvent() 发布一个 RefreshEvent 事件；
该事件又会被 RefreshEventListener（事件更新监听器）监听，源码如下：

public void onApplicationEvent(ApplicationEvent event){
    if (event instanceof ApplicationReadyEvent) {
        //监听 ApplicationReadyEvent 事件
        this.handle((ApplicationReadyEvent)event);
    } else if (event instanceof RefreshEvent) {
        //【断点步入 2.3】监听 RefreshEvent 事件
        this.handle((RefreshEvent)event);
    }
}

RefreshEventListener（事件更新监听器）类使用 RefreshEventListener.handle() 方法变更配置，源码如下：

public void handle(RefreshEvent event) {
    if (this.ready.get()) {
        log.debug("Event received " + event.getEventDesc());
        Set<String> keys = this.refresh.refresh();
        log.info("Refresh keys changed: " + keys);
    }
}

NacosPropertySourceLocator.locate()：初始化 ConfigService 对象，定位配置；
- NacosPropertySourceLocator.loadApplicationConfiguration()：根据 Data ID 加载配置；
  - NacosPropertySourceLocator.loadNacosDataIfPresent()：判断是否更新配置；
    - NacosPropertySourceBuilder.build()：加载并封装配置；
      - NacosPropertySourceBuilder.loadNacosData()：加载配置；
        
        NacosConfigService.getConfig()：使用配置服务获取配置；
        
        NacosConfigService.getConfigInner()：最终在这里获取到配置；

上下文准备完毕，程序运行，EventPublishingRunListener 发布 ApplicationReadyEvent 事件；
NacosContextRefresher.onApplicationEvent()：监听 ApplicationReadyEvent 事件；
- NacosContextRefresher.registerNacosListener()：注册 Nacos 监听器，监听配置变更；
变更发生时，NacosContextRefresher 发布一个 RefreshEvent
RefreshEventListener.onApplicationEvent()：同时监听 ApplicationReadyEvent 和 RefreshEvent 事件；
- RefreshEventListener.handle()：实施变更方法；

新人制作，如有错误，欢迎指出，感激不尽！

欢迎关注公众号，会分享一些更日常的东西！

如需转载，请标注出处！

微服务架构 | *2.4 Nacos 获取配置与事件订阅机制的源码分析

继续阅读

Kafka：Topic概念与API介绍

ZooKeeper ： Curator框架之分布式屏障DistributedDoubleBarrier

RabbitMQ：交换机（fanout exchange）

Doris SQL 原理解析

ZooKeeper ： Curator框架之分布式锁InterProcessMutex

阿里巴巴分布式服务框架 Dubbo 团队成员梁飞专访

数据迁移方法数据迁移原则数据迁移之双写方案数据迁移之级联同步方案

微服务-性能压测\缓存redis和分布式锁redisson和SpringCache

Nacos 2.0 升级前后性能对比压测

Spring数据和Redis

redis集群数据一致性_RedisRaft为Redis集群带来强大的数据一致性

Centos7 下 Hadoop 2.6.4 分布式集群环境搭建摘要集群准备安装JDK 安装 Hadoop 2.6.4 部署 slaver1-slaver4 启动 hadoop 集群成功了

niginx的location的location详细规则和优先级关系以及测试案例2.1匹配语法2.2优先级判断2.3练习3.1处理逻辑如何执行3.2功能测试3.3流程解析

celery使用入门

MapReduce的几个企业级经典面试案例MapReduce的几个企业级经典面试案例