前言：
前面系列文章中我们分析了 Nacos 客户端的配置加载、配置热更新、Nacos 服务端的配置加载、配置转储到磁盘文件中等，本篇我们来分析一下 Nacos 配置的发布逻辑。
Nacos 系列文章传送门：
Nacos 配置管理模型 – 命名空间（Namespace）、配置分组（Group）和配置集ID（Data ID）
Nacos Server 是如何通知 Nacos Client 服务下线？
Nacos Client 是如何接受 Nacos Server 推送的数据？
Nacos 故障转移源码分析（FailoverReactor）
ConfigController#publishConfig 方法源码解析
ConfigController#publishConfig 方法是配置发布的入口，当我在 WEB 界面修改了配置之后，会点击发布配置，最终会调用到 ConfigController#publishConfig，该接口共做了一下几件事：
- 校验入参。
- 白名单处理。
- 将入参构造成配置信息对象。
- beta、tag 判断，不同情况，调用不通的处理逻辑（大同小异），我们重点关注无 tag 分支逻辑。
- 发布配置数据变更 ConfigDataChangeEvent 事件。
- 日志跟踪。
//com.alibaba.nacos.config.server.controller.ConfigController#publishConfig
@PostMapping
@Secured(action = ActionTypes.WRITE, parser = ConfigResourceParser.class)
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) throws NacosException {
//获取ip
final String srcIp = RequestUtil.getRemoteIp(request);
//app name
final String requestIpApp = RequestUtil.getAppName(request);
//用户
srcUser = RequestUtil.getSrcUserName(request);
//check type
//检查类型
if (!ConfigType.isValidType(type)) {
type = ConfigType.getDefaultType().getType();
}
// check tenant
//检查租户
ParamUtils.checkTenant(tenant);
//检查参数
ParamUtils.checkParam(dataId, group, "datumId", content);
//检查 tag
ParamUtils.checkParam(tag);
//构造配置信息
Map<String, Object> configAdvanceInfo = new HashMap<String, Object>(10);
MapUtils.putIfValNoNull(configAdvanceInfo, "config_tags", configTags);
MapUtils.putIfValNoNull(configAdvanceInfo, "desc", desc);
MapUtils.putIfValNoNull(configAdvanceInfo, "use", use);
MapUtils.putIfValNoNull(configAdvanceInfo, "effect", effect);
MapUtils.putIfValNoNull(configAdvanceInfo, "type", type);
MapUtils.putIfValNoNull(configAdvanceInfo, "schema", schema);
ParamUtils.checkParam(configAdvanceInfo);
//白名单
if (AggrWhitelist.isAggrDataId(dataId)) {
LOGGER.warn("[aggr-conflict] {} attempt to publish single data, {}, {}", RequestUtil.getRemoteIp(request),
dataId, group);
throw new NacosException(NacosException.NO_RIGHT, "dataId:" + dataId + " is aggr");
}
//获取时间戳
final Timestamp time = TimeUtils.getCurrentTime();
//获取 betaIps
String betaIps = request.getHeader("betaIps");
//创建配置信息对象
ConfigInfo configInfo = new ConfigInfo(dataId, group, tenant, appName, content);
//设置类型
configInfo.setType(type);
//beta 测试版本为空判断
if (StringUtils.isBlank(betaIps)) {
//tag 为空判断 大多数时候我们是不是设置 tag 的
if (StringUtils.isBlank(tag)) {
//重点关注的分支 因为 大多数时候我们是不是设置 tag 的
//auto tag 发布逻辑
persistService.insertOrUpdate(srcIp, srcUser, configInfo, time, configAdvanceInfo, true);
//发布配置数据变更 ConfigDataChangeEvent 事件
ConfigChangePublisher
.notifyConfigChange(new ConfigDataChangeEvent(false, dataId, group, tenant, time.getTime()));
} else {
//tag 发布逻辑
persistService.insertOrUpdateTag(configInfo, tag, srcIp, srcUser, time, true);
//发布配置数据变更 ConfigDataChangeEvent 事件
ConfigChangePublisher.notifyConfigChange(
new ConfigDataChangeEvent(false, dataId, group, tenant, tag, time.getTime()));
}
} else {
// beta publish
//beta 发布逻辑
persistService.insertOrUpdateBeta(configInfo, betaIps, srcIp, srcUser, time, true);
//发布配置数据变更 ConfigDataChangeEvent 事件
ConfigChangePublisher
.notifyConfigChange(new ConfigDataChangeEvent(true, dataId, group, tenant, time.getTime()));
}
//日志跟踪
ConfigTraceService
.logPersistenceEvent(dataId, group, tenant, requestIpApp, time.getTime(), InetUtils.getSelfIP(),
ConfigTraceService.PERSISTENCE_EVENT_PUB, content);
return true;
}
EmbeddedStoragePersistServiceImpl#insertOrUpdate 方法源码解析
EmbeddedStoragePersistServiceImpl#insertOrUpdate 方法是处理内嵌数据库的方法，会根据配置信息去查询，如果可以查到到结果就认为是更新配置信息，否则认为是新增配置信息。
//com.alibaba.nacos.config.server.service.repository.embedded.EmbeddedStoragePersistServiceImpl#insertOrUpdate(java.lang.String, java.lang.String, com.alibaba.nacos.config.server.model.ConfigInfo, java.sql.Timestamp, java.util.Map<java.lang.String,java.lang.Object>, boolean)
@Override
public void insertOrUpdate(String srcIp, String srcUser, ConfigInfo configInfo, Timestamp time,
Map<String, Object> configAdvanceInfo, boolean notify) {
//查询配置信息
if (Objects.isNull(findConfigInfo(configInfo.getDataId(), configInfo.getGroup(), configInfo.getTenant()))) {
//找不到 表示新增
addConfigInfo(srcIp, srcUser, configInfo, time, configAdvanceInfo, notify);
} else {
//可以找到 更新配置
updateConfigInfo(configInfo, srcIp, srcUser, time, configAdvanceInfo, notify);
}
}
EmbeddedStoragePersistServiceImpl#addConfigInfo 方法源码解析
EmbeddedStoragePersistServiceImpl#addConfigInfo 内嵌数据库的新增配置信息的方法，就是将配置信息入库，并处理 CP 集群的情况。
//com.alibaba.nacos.config.server.service.repository.embedded.EmbeddedStoragePersistServiceImpl#addConfigInfo(java.lang.String, java.lang.String, com.alibaba.nacos.config.server.model.ConfigInfo, java.sql.Timestamp, java.util.Map<java.lang.String,java.lang.Object>, boolean)
@Override
public void addConfigInfo(final String srcIp, final String srcUser, final ConfigInfo configInfo,
final Timestamp time, final Map<String, Object> configAdvanceInfo, final boolean notify) {
//新增配置
addConfigInfo(srcIp, srcUser, configInfo, time, configAdvanceInfo, notify, null);
}
//新增配置
//com.alibaba.nacos.config.server.service.repository.embedded.EmbeddedStoragePersistServiceImpl#addConfigInfo(java.lang.String, java.lang.String, com.alibaba.nacos.config.server.model.ConfigInfo, java.sql.Timestamp, java.util.Map<java.lang.String,java.lang.Object>, boolean, java.util.function.BiConsumer<java.lang.Boolean,java.lang.Throwable>)
private void addConfigInfo(final String srcIp, final String srcUser, final ConfigInfo configInfo,
final Timestamp time, final Map<String, Object> configAdvanceInfo, final boolean notify,
BiConsumer<Boolean, Throwable> consumer) {
try {
//临时租户信息
final String tenantTmp =
StringUtils.isBlank(configInfo.getTenant()) ? StringUtils.EMPTY : configInfo.getTenant();
configInfo.setTenant(tenantTmp);
//生成配置id
long configId = idGeneratorManager.nextId(RESOURCE_CONFIG_INFO_ID);
//配置历史id
long hisId = idGeneratorManager.nextId(RESOURCE_CONFIG_HISTORY_ID);
//配置信息入库
addConfigInfoAtomic(configId, srcIp, srcUser, configInfo, time, configAdvanceInfo);
//配置信息tag
String configTags = configAdvanceInfo == null ? null : (String) configAdvanceInfo.get("config_tags");
//配置信息tag入库
addConfigTagsRelation(configId, configTags, configInfo.getDataId(), configInfo.getGroup(),
configInfo.getTenant());
//配置历史信息入库
insertConfigHistoryAtomic(hisId, configInfo, srcIp, srcUser, time, "I");
//CP 集群处理
EmbeddedStorageContextUtils.onModifyConfigInfo(configInfo, srcIp, time);
databaseOperate.blockUpdate(consumer);
} finally {
//清除 SQL 上下文
EmbeddedStorageContextUtils.cleanAllContext();
}
}
EmbeddedStoragePersistServiceImpl#updateConfigInfo 方法源码解析
EmbeddedStoragePersistServiceImpl#updateConfigInfo 方法是内嵌数据库的更新配置信息的方法，就是将配置信息入库更新，如果当前配置信息之前有 tag 记录，删除并重新写入，最后会处理 CP 集群的情况。
//com.alibaba.nacos.config.server.service.repository.embedded.EmbeddedStoragePersistServiceImpl#updateConfigInfo
@Override
public void updateConfigInfo(final ConfigInfo configInfo, final String srcIp, final String srcUser,
final Timestamp time, final Map<String, Object> configAdvanceInfo, final boolean notify) {
try {
//获取旧的配置信息
ConfigInfo oldConfigInfo = findConfigInfo(configInfo.getDataId(), configInfo.getGroup(),
configInfo.getTenant());
//租户信息
final String tenantTmp =
StringUtils.isBlank(configInfo.getTenant()) ? StringUtils.EMPTY : configInfo.getTenant();
//重新设置租户信息
oldConfigInfo.setTenant(tenantTmp);
//app name
String appNameTmp = oldConfigInfo.getAppName();
// If the appName passed by the user is not empty, the appName of the user is persisted;
// otherwise, the appName of db is used. Empty string is required to clear appName
//app name 设置
if (configInfo.getAppName() == null) {
configInfo.setAppName(appNameTmp);
}
//更新配置信息
updateConfigInfoAtomic(configInfo, srcIp, srcUser, time, configAdvanceInfo);
//配置 tag
String configTags = configAdvanceInfo == null ? null : (String) configAdvanceInfo.get("config_tags");
//配置tag 为空判断
if (configTags != null) {
// Delete all tags and recreate them
//移除旧的tag
removeTagByIdAtomic(oldConfigInfo.getId());
//新增新的tag 数据到表中
addConfigTagsRelation(oldConfigInfo.getId(), configTags, configInfo.getDataId(), configInfo.getGroup(),
configInfo.getTenant());
}
//新增配置历史数据
insertConfigHistoryAtomic(oldConfigInfo.getId(), oldConfigInfo, srcIp, srcUser, time, "U");
//CP 集群处理
EmbeddedStorageContextUtils.onModifyConfigInfo(configInfo, srcIp, time);
databaseOperate.blockUpdate();
} finally {
//清除 SQL 上下文
EmbeddedStorageContextUtils.cleanAllContext();
}
}
ExternalStoragePersistServiceImpl#insertOrUpdate 方法源码解析
ExternalStoragePersistServiceImpl#insertOrUpdate 方法主要判断了配置信息是新增还是更新，这里使用了一致性约束，如果配置新增出错，就会去走配置更新逻辑。
//com.alibaba.nacos.config.server.service.repository.extrnal.ExternalStoragePersistServiceImpl#insertOrUpdate(java.lang.String, java.lang.String, com.alibaba.nacos.config.server.model.ConfigInfo, java.sql.Timestamp, java.util.Map<java.lang.String,java.lang.Object>, boolean)
@Override
public void insertOrUpdate(String srcIp, String srcUser, ConfigInfo configInfo, Timestamp time,
Map<String, Object> configAdvanceInfo, boolean notify) {
try {
//新增配置信息
addConfigInfo(srcIp, srcUser, configInfo, time, configAdvanceInfo, notify);
} catch (DataIntegrityViolationException ive) { // Unique constraint conflict
//更新配置信息
updateConfigInfo(configInfo, srcIp, srcUser, time, configAdvanceInfo, notify);
}
}
ExternalStoragePersistServiceImpl#addConfigInfo 方法源码解析
ExternalStoragePersistServiceImpl#addConfigInfo 是使用外部数据库存储的新增配置信息的处理方法，方法没有太多的逻辑，就是直接新增配置信息、tag 信息、配置历史信息。
//com.alibaba.nacos.config.server.service.repository.extrnal.ExternalStoragePersistServiceImpl#addConfigInfo
@Override
public void addConfigInfo(final String srcIp, final String srcUser, final ConfigInfo configInfo,
final Timestamp time, final Map<String, Object> configAdvanceInfo, final boolean notify) {
boolean result = tjt.execute(status -> {
try {
//新增配置信息
long configId = addConfigInfoAtomic(-1, srcIp, srcUser, configInfo, time, configAdvanceInfo);
String configTags = configAdvanceInfo == null ? null : (String) configAdvanceInfo.get("config_tags");
//新增 tags 信息
addConfigTagsRelation(configId, configTags, configInfo.getDataId(), configInfo.getGroup(),
configInfo.getTenant());
//新增配置信息历史
insertConfigHistoryAtomic(0, configInfo, srcIp, srcUser, time, "I");
} catch (CannotGetJdbcConnectionException e) {
LogUtil.FATAL_LOG.error("[db-error] " + e.toString(), e);
throw e;
}
return Boolean.TRUE;
});
}
ExternalStoragePersistServiceImpl#updateConfigInfo 方法源码解析
ExternalStoragePersistServiceImpl#updateConfigInfo 方法是使用外部存储的配置更新处理方法，该方法会先获取旧的配置信息，然后对象 APP Name 进行处理，然后进行配置更新，同时会对 tag 进行处理，如果 tag 信息存在，则直接删除后新增，同时会新增一条历史配置信息。
//com.alibaba.nacos.config.server.service.repository.extrnal.ExternalStoragePersistServiceImpl#updateConfigInfo
@Override
public void updateConfigInfo(final ConfigInfo configInfo, final String srcIp, final String srcUser,
final Timestamp time, final Map<String, Object> configAdvanceInfo, final boolean notify) {
boolean result = tjt.execute(status -> {
try {
//获取旧的配置信息
ConfigInfo oldConfigInfo = findConfigInfo(configInfo.getDataId(), configInfo.getGroup(),
configInfo.getTenant());
//app name
String appNameTmp = oldConfigInfo.getAppName();
/*
If the appName passed by the user is not empty, use the persistent user's appName,
otherwise use db; when emptying appName, you need to pass an empty string
*/
//app name 设置
if (configInfo.getAppName() == null) {
configInfo.setAppName(appNameTmp);
}
//更新配置信息
updateConfigInfoAtomic(configInfo, srcIp, srcUser, time, configAdvanceInfo);
String configTags = configAdvanceInfo == null ? null : (String) configAdvanceInfo.get("config_tags");
if (configTags != null) {
// delete all tags and then recreate
//tag 如果存在就删除 在重新生成 tag
removeTagByIdAtomic(oldConfigInfo.getId());
addConfigTagsRelation(oldConfigInfo.getId(), configTags, configInfo.getDataId(),
configInfo.getGroup(), configInfo.getTenant());
}
//新增配置历史信息
insertConfigHistoryAtomic(oldConfigInfo.getId(), oldConfigInfo, srcIp, srcUser, time, "U");
} catch (CannotGetJdbcConnectionException e) {
LogUtil.FATAL_LOG.error("[db-error] " + e.toString(), e);
throw e;
}
return Boolean.TRUE;
});
}
ConfigChangePublisher#notifyConfigChange 方法源码解析
ConfigChangePublisher#notifyConfigChange 方法会先判断 Nacos 是否内嵌存储且是单机模式，如果是就不发送配置数据变化 ConfigDataChangeEvent 事件了，否则会发送配置数据变化事件 ConfigDataChangeEvent。
//ConfigChangePublisher#notifyConfigChange 方法会先判断 Nacos 是否内嵌存储且是单机模式，如果是就不发送配置数据变化 ConfigDataChangeEvent 事件了，否则会发送配置数据变化事件 ConfigDataChangeEvent。
//com.alibaba.nacos.config.server.service.ConfigChangePublisher#notifyConfigChange
public static void notifyConfigChange(ConfigDataChangeEvent event) {
//是否是内嵌存储模式 切是单机模式
if (PropertyUtil.isEmbeddedStorage() && !EnvUtil.getStandaloneMode()) {
//不发布事件
return;
}
//发布配置数据变化 ConfigDataChangeEvent 事件
NotifyCenter.publishEvent(event);
}
通过源码可以知道，Nacos 还是使用了统一事件发布中心 NotifyCenter 来发布 ConfigDataChangeEvent 事件的，我们直接查找ConfigDataChangeEvent 的 onEvent 方法来查看具体的事件处理逻辑。
AsyncNotifyService 构造方法源码解析
AsyncNotifyService 构造方法中，将 ConfigDataChangeEvent 事件注册到 NotifyCenter 通知中心，同时注册一个订阅 ConfigDataChangeEvent事件的处理类，AsyncNotifyService 交给了 Spring 管理，在 Spring IOC 容器启动的时候，就会创建这个 AsyncNotifyService 对象，也就是说会执行 AsyncNotifyService 构造方法，我们重点关注一下 onEvent 方法，onEvent 方法主要做了一下事情:
- 获取 Nacos 集群的所有服务节点。
- 创建一个队列，将 Nacos 节点封装成一个 NotifySingleTask 对象存入队列中。
- 通过线程池异步执行，重点关注 AsyncTask，AsyncTask 实现了 Runnable 接口，关注其 run 方法即可。
//com.alibaba.nacos.config.server.service.notify.AsyncNotifyService#AsyncNotifyService
@Autowired
public AsyncNotifyService(ServerMemberManager memberManager) {
//集群成员
this.memberManager = memberManager;
// Register ConfigDataChangeEvent to NotifyCenter.
//将 ConfigDataChangeEvent 注册到 NotifyCenter
NotifyCenter.registerToPublisher(ConfigDataChangeEvent.class, NotifyCenter.ringBufferSize);
// Register A Subscriber to subscribe ConfigDataChangeEvent.
//注册一个订阅者来订阅 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;
//Nacos 集群成员
Collection<Member> ipList = memberManager.allMembers();
// In fact, any type of queue here can be
Queue<NotifySingleTask> queue = new LinkedList<NotifySingleTask>();
//遍历集群成员
for (Member member : ipList) {
// NotifySingleTask 任务加入到队列
queue.add(new NotifySingleTask(dataId, group, tenant, tag, dumpTs, member.getAddress(),
evt.isBeta));
}
//配置执行器执行异步通知 AsyncTask 实现了 Runnable 接口 关注 run 方法
ConfigExecutor.executeAsyncNotify(new AsyncTask(nacosAsyncRestTemplate, queue));
}
}
@Override
public Class<? extends Event> subscribeType() {
return ConfigDataChangeEvent.class;
}
});
}
AsyncTask#run 方法源码解析
AsyncTask#run 方法主要做了一下几件事：
- 判断队列是否为空，只要队列不为空，就会使用 while 循环从队列中取出 NotifySingleRpcTask 任务来执行。
- 判断 Nacos 集群中是否有当前成员，有才会继续执行。
- 判断当前 Nacos 节点是否健康，如果不健康，将其放入通知列表中，并重新加入队列，进行延迟处理。
- Nacos 节点健康，会构造配置变动集群同步的请求对象，然后通过 HTTP 方式（Nacos 2.X 使用 GRPC 协议）然后通知目标节点。
//com.alibaba.nacos.config.server.service.notify.AsyncNotifyService.AsyncTask#run
@Override
public void run() {
//执行异步调用
executeAsyncInvoke();
}
private void executeAsyncInvoke() {
//队列为空判断
while (!queue.isEmpty()) {
//获取任务对象
NotifySingleTask task = queue.poll();
//目标ip
String targetIp = task.getTargetIP();
//集群成员中是否有当前 targetIp
if (memberManager.hasMember(targetIp)) {
// start the health check and there are ips that are not monitored, put them directly in the notification queue, otherwise notify
//健康检查
boolean unHealthNeedDelay = memberManager.isUnHealth(targetIp);
if (unHealthNeedDelay) {
// target ip is unhealthy, then put it in the notification list
//目标id 不健康 将其放入通知列表中
ConfigTraceService.logNotifyEvent(task.getDataId(), task.getGroup(), task.getTenant(), null,
task.getLastModified(), InetUtils.getSelfIP(), ConfigTraceService.NOTIFY_EVENT_UNHEALTH,
0, task.target);
// get delay time and set fail count to the task
//重新加入队列 延迟处理
asyncTaskExecute(task);
} else {
//header
Header header = Header.newInstance();
header.addParam(NotifyService.NOTIFY_HEADER_LAST_MODIFIED, String.valueOf(task.getLastModified()));
header.addParam(NotifyService.NOTIFY_HEADER_OP_HANDLE_IP, InetUtils.getSelfIP());
//是否是 beta 测试版本
if (task.isBeta) {
header.addParam("isBeta", "true");
}
//构造 headder
AuthHeaderUtil.addIdentityToHeader(header);
//http 调用同步配置 如果调用成功 则执行 AsyncNotifyCallBack 的回调方法
restTemplate.get(task.url, header, Query.EMPTY, String.class, new AsyncNotifyCallBack(task));
}
}
}
}
Nacos 配置发布后通知集群内其他节点
CommunicationController#notifyConfigInfo 方法源码解析
CommunicationController#notifyConfigInfo 方法是集群配置信息通知的入口，该方法业务非常简单，先从请求中提取 dataId、group、lastModified、handleIp、beta 參數，然后就调用了 DumpService#dump 方法。
//com.alibaba.nacos.config.server.controller.CommunicationController#notifyConfigInfo
@GetMapping("/dataChange")
public Boolean notifyConfigInfo(HttpServletRequest request, @RequestParam("dataId") String dataId,
@RequestParam("group") String group,
@RequestParam(value = "tenant", required = false, defaultValue = StringUtils.EMPTY) String tenant,
@RequestParam(value = "tag", required = false) String tag) {
//配置集id
dataId = dataId.trim();
//分组
group = group.trim();
//最后一次修改时间
String lastModified = request.getHeader(NotifyService.NOTIFY_HEADER_LAST_MODIFIED);
long lastModifiedTs = StringUtils.isEmpty(lastModified) ? -1 : Long.parseLong(lastModified);
//处理 ip
String handleIp = request.getHeader(NotifyService.NOTIFY_HEADER_OP_HANDLE_IP);
//beta
String isBetaStr = request.getHeader("isBeta");
//isBetaStr 不为空 是否为true
if (StringUtils.isNotBlank(isBetaStr) && trueStr.equals(isBetaStr)) {
//beta 版本
dumpService.dump(dataId, group, tenant, lastModifiedTs, handleIp, true);
} else {
//非beta 版本
dumpService.dump(dataId, group, tenant, tag, lastModifiedTs, handleIp);
}
return true;
}
DumpService#dump 方法源码解析
Beta 版本的 dump 方法，该方法将 DumpTask 添加到 TaskManager 中，它将会异步执行。
//com.alibaba.nacos.config.server.service.dump.DumpService#dump(java.lang.String, java.lang.String, java.lang.String, long, java.lang.String, boolean)
public void dump(String dataId, String group, String tenant, long lastModified, String handleIp, boolean isBeta) {
//获取可以
String groupKey = GroupKey2.getKey(dataId, group, tenant);
//任务key
String taskKey = String.join("+", dataId, group, tenant, String.valueOf(isBeta));
//添加任务
dumpTaskMgr.addTask(taskKey, new DumpTask(groupKey, lastModified, handleIp, isBeta));
DUMP_LOG.info("[dump-task] add task. groupKey={}, taskKey={}", groupKey, taskKey);
}
DumpService#dump 方法源码解析
非 Beta 版本的 dump 方法，该方法将 DumpTask 添加到 TaskManager 中，它将会异步执行。
//com.alibaba.nacos.config.server.service.dump.DumpService#dump(java.lang.String, java.lang.String, java.lang.String, java.lang.String, long, java.lang.String)
public void dump(String dataId, String group, String tenant, String tag, long lastModified, String handleIp) {
dump(dataId, group, tenant, tag, lastModified, handleIp, false);
}
//com.alibaba.nacos.config.server.service.dump.DumpService#dump(java.lang.String, java.lang.String, java.lang.String, java.lang.String, long, java.lang.String, boolean)
public void dump(String dataId, String group, String tenant, String tag, long lastModified, String handleIp,
boolean isBeta) {
//获取 key
String groupKey = GroupKey2.getKey(dataId, group, tenant);
//任务key
String taskKey = String.join("+", dataId, group, tenant, String.valueOf(isBeta), tag);
//添加任务
dumpTaskMgr.addTask(taskKey, new DumpTask(groupKey, tag, lastModified, handleIp, isBeta));
DUMP_LOG.info("[dump-task] add task. groupKey={}, taskKey={}", groupKey, taskKey);
}
TaskManager#addTask 方法源码解析
TaskManager#addTask 方法只是调用 NacosDelayTaskExecuteEngine#addTask 方法，添加任务。
//com.alibaba.nacos.config.server.manager.TaskManager#addTask
@Override
public void addTask(Object key, AbstractDelayTask newTask) {
//调用父类添加任务
super.addTask(key, newTask);
//监控
MetricsMonitor.getDumpTaskMonitor().set(tasks.size());
}
NacosDelayTaskExecuteEngine#addTask 方法源码解析
NacosDelayTaskExecuteEngine#addTask 方法是真正添加任务的方法，会先判断任务是否存在，任务存在，则合并任务后添加到 tasks 中，否则直接添加到 tasks 中。
//com.alibaba.nacos.common.task.engine.NacosDelayTaskExecuteEngine#addTask
@Override
public void addTask(Object key, AbstractDelayTask newTask) {
//加锁
lock.lock();
try {
//protected final ConcurrentHashMap<Object, AbstractDelayTask> tasks;
//从 task 中获取任务
AbstractDelayTask existTask = tasks.get(key);
if (null != existTask) {
//任务存在 合并任务
newTask.merge(existTask);
}
//添加任务
tasks.put(key, newTask);
} finally {
//解锁
lock.unlock();
}
}
通过源码我们看到最终提交了一个 AbstractDelayTask 任务，这种异步执行的设计，一般都会采用生产者消费者模式来完成，因此一定有另外一个线程来执行。
NacosDelayTaskExecuteEngine 构造方法源码解析
NacosDelayTaskExecuteEngine 构造方法中初始了一个 tasks 队列和一个延期执行的任务，其中具体的任务是 ProcessRunnable，ProcessRunnable 实现了 Runnable 接口，重点关注 ProcessRunnable#run 方法即可 。
//com.alibaba.nacos.common.task.engine.NacosDelayTaskExecuteEngine.NacosDelayTaskExecuteEngine(java.lang.String, int, org.slf4j.Logger, long)
public NacosDelayTaskExecuteEngine(String name, int initCapacity, Logger logger, long processInterval) {
super(logger);
//保存 AbstractDelayTask 任务的队列
tasks = new ConcurrentHashMap<Object, AbstractDelayTask>(initCapacity);
//得到单线程执行器
processingExecutor = ExecutorFactory.newSingleScheduledExecutorService(new NameThreadFactory(name));
//定时执行任务 重点关注 ProcessRunnable ProcessRunnable 实现了 Runnable 接口
processingExecutor
.scheduleWithFixedDelay(new ProcessRunnable(), processInterval, processInterval, TimeUnit.MILLISECONDS);
}
ProcessRunnable#run 方法源码解析
ProcessRunnable#run 方法调用了 NacosDelayTaskExecuteEngine#processTasks 方法。
//com.alibaba.nacos.common.task.engine.NacosDelayTaskExecuteEngine.ProcessRunnable#run
@Override
public void run() {
try {
processTasks();
} catch (Throwable e) {
getEngineLog().error(e.toString(), e);
}
}
NacosDelayTaskExecuteEngine#processTasks 方法源码解析
NacosDelayTaskExecuteEngine#processTasks 方法会获取所有的任务，遍历所有任务，如果任务为空则跳过，然后获取任务处理器，这里获取的任务处理器是 DumpProcessor，调用处理器的 process 方法处理任务，如果处理失败，会重新加入队列。
//com.alibaba.nacos.common.task.engine.NacosDelayTaskExecuteEngine#processTasks
protected void processTasks() {
//获取所有任务
Collection<Object> keys = getAllTaskKeys();
//遍历任务
for (Object taskKey : keys) {
//移出任务
AbstractDelayTask task = removeTask(taskKey);
if (null == task) {
//task 为空 跳过
continue;
}
//获取任务处理器 这里返回的是 DumpProcessor
NacosTaskProcessor processor = getProcessor(taskKey);
if (null == processor) {
//处理器为空判断 处理器为空 表示获取处理器失败 跳过
getEngineLog().error("processor not found for task, so discarded. " + task);
continue;
}
try {
// ReAdd task if process failed
//处理任务
if (!processor.process(task)) {
//失败重试 也就是重新添加到任务队列中
retryFailedTask(taskKey, task);
}
} catch (Throwable e) {
getEngineLog().error("Nacos task execute error : " + e.toString(), e);
//失败重试 也就是重新添加到任务队列中
retryFailedTask(taskKey, task);
}
}
}
DumpProcessor#process 方法源码解析
DumpProcessor#process 方法会对主要是做的配置转储之前的准备工作，主要流程如下：
- 获取持久化 persistService 对象。
- 从任务中获取配置信息，如 dataId、group、tenant、lastModified、handleIp、isBeta、tag 等。
- 然后判断是否是 beta 版本，是 beta 版本则走 beta 版本的逻辑，否则再判断是否有 tag，然后分别执行有无 tag 的逻辑。
- 最终都会执行配置转储。
//com.alibaba.nacos.config.server.service.dump.processor.DumpProcessor#process
@Override
public boolean process(NacosTask task) {
//获取持久化 persistService
final PersistService persistService = dumpService.getPersistService();
//任务强转为 DumpTask
DumpTask dumpTask = (DumpTask) task;
//获取group key 信息
String[] pair = GroupKey2.parseKey(dumpTask.getGroupKey());
//dataId group tenant
String dataId = pair[0];
String group = pair[1];
String tenant = pair[2];
//最后一次修改时间
long lastModified = dumpTask.getLastModified();
//处理ip
String handleIp = dumpTask.getHandleIp();
//是否 beta 测试版本
boolean isBeta = dumpTask.isBeta();
//获取 tag 我们使用时候一般 无 tag
String tag = dumpTask.getTag();
//配置转储事件生成器
ConfigDumpEvent.ConfigDumpEventBuilder build = ConfigDumpEvent.builder().namespaceId(tenant).dataId(dataId)
.group(group).isBeta(isBeta).tag(tag).lastModifiedTs(lastModified).handleIp(handleIp);
//是否是 beta 版本
if (isBeta) {
// beta发布，则dump数据，更新beta缓存
ConfigInfo4Beta cf = persistService.findConfigInfo4Beta(dataId, group, tenant);
//赋值 cf 不为空 remove fasle 否则 true
build.remove(Objects.isNull(cf));
//赋值 betaIps
build.betaIps(Objects.isNull(cf) ? null : cf.getBetaIps());
//赋值 betaIps
build.content(Objects.isNull(cf) ? null : cf.getContent());
//执行配置转储
return DumpConfigHandler.configDump(build.build());
} else {
//非 beta 版本
//是否有 tag 判断
if (StringUtils.isBlank(tag)) {
//无 tag 查找配置信息
ConfigInfo cf = persistService.findConfigInfo(dataId, group, tenant);
//赋值 cf 不为空 remove fasle 否则 true
build.remove(Objects.isNull(cf));
//赋值 content
build.content(Objects.isNull(cf) ? null : cf.getContent());
//赋值 type
build.type(Objects.isNull(cf) ? null : cf.getType());
//执行配置转储
return DumpConfigHandler.configDump(build.build());
} else {
//获取配置信息
ConfigInfo4Tag cf = persistService.findConfigInfo4Tag(dataId, group, tenant, tag);
//赋值 remove cf 不为空 remove fasle 否则 true
build.remove(Objects.isNull(cf));
//赋值 content
build.content(Objects.isNull(cf) ? null : cf.getContent());
//执行配置转储
return DumpConfigHandler.configDump(build.build());
}
}
}
DumpConfigHandler#configDump 方法源码解析
DumpConfigHandler#configDump 方法会根据配置基本信息，进行 beta、tag、是否 remove 判断，不同分支执行不同的业务逻辑大同小异，最终都会调用 ConfigCacheService 的相关 dump 方法，其中 remove 操作实际上会发布一个 LocalDataChangeEvent，这个事件我们在配置中心配置加载的源码中中分析过，该事件最终触发的也是配置转储逻辑。
//com.alibaba.nacos.config.server.service.dump.DumpConfigHandler#configDump
public static boolean configDump(ConfigDumpEvent event) {
//配置基本信息
final String dataId = event.getDataId();
final String group = event.getGroup();
final String namespaceId = event.getNamespaceId();
final String content = event.getContent();
final String type = event.getType();
final long lastModified = event.getLastModifiedTs();
//是否 beta 测试版本
if (event.isBeta()) {
//是 beta 版本
boolean result = false;
//是否需要 remove
if (event.isRemove()) {
//是 则remove掉配置 会发布 LocalDataChangeEvent 事件
result = ConfigCacheService.removeBeta(dataId, group, namespaceId);
if (result) {
ConfigTraceService.logDumpEvent(dataId, group, namespaceId, null, lastModified, event.getHandleIp(),
ConfigTraceService.DUMP_EVENT_REMOVE_OK, System.currentTimeMillis() - lastModified, 0);
}
return result;
} else {
//否 转储文件
result = ConfigCacheService
.dumpBeta(dataId, group, namespaceId, content, lastModified, event.getBetaIps());
if (result) {
//日志记录
ConfigTraceService.logDumpEvent(dataId, group, namespaceId, null, lastModified, event.getHandleIp(),
ConfigTraceService.DUMP_EVENT_OK, System.currentTimeMillis() - lastModified,
content.length());
}
}
return result;
}
//tag 为空判断
if (StringUtils.isBlank(event.getTag())) {
//白名单处理
if (dataId.equals(AggrWhitelist.AGGRIDS_METADATA)) {
AggrWhitelist.load(content);
}
if (dataId.equals(ClientIpWhiteList.CLIENT_IP_WHITELIST_METADATA)) {
ClientIpWhiteList.load(content);
}
if (dataId.equals(SwitchService.SWITCH_META_DATAID)) {
SwitchService.load(content);
}
boolean result;
//是否需要移出
if (!event.isRemove()) {
//否 转储配置信息 配置中心加载的源码中分析过该方法
result = ConfigCacheService.dump(dataId, group, namespaceId, content, lastModified, type);
if (result) {
//日志记录
ConfigTraceService.logDumpEvent(dataId, group, namespaceId, null, lastModified, event.getHandleIp(),
ConfigTraceService.DUMP_EVENT_OK, System.currentTimeMillis() - lastModified,
content.length());
}
} else {
//是 移除配置信息 会发布 LocalDataChangeEvent 事件
result = ConfigCacheService.remove(dataId, group, namespaceId);
if (result) {
//日志记录
ConfigTraceService.logDumpEvent(dataId, group, namespaceId, null, lastModified, event.getHandleIp(),
ConfigTraceService.DUMP_EVENT_REMOVE_OK, System.currentTimeMillis() - lastModified, 0);
}
}
return result;
} else {
//tag 不为空
boolean result;
//是否需要移出
if (!event.isRemove()) {
//否 转储配置信息 配置中心加载的源码中分析过该方法
result = ConfigCacheService.dumpTag(dataId, group, namespaceId, event.getTag(), content, lastModified);
if (result) {
//日志记录
ConfigTraceService.logDumpEvent(dataId, group, namespaceId, null, lastModified, event.getHandleIp(),
ConfigTraceService.DUMP_EVENT_OK, System.currentTimeMillis() - lastModified,
content.length());
}
} else {
//是 移除配置信息 会发布 LocalDataChangeEvent 事件
result = ConfigCacheService.removeTag(dataId, group, namespaceId, event.getTag());
if (result) {
ConfigTraceService.logDumpEvent(dataId, group, namespaceId, null, lastModified, event.getHandleIp(),
ConfigTraceService.DUMP_EVENT_REMOVE_OK, System.currentTimeMillis() - lastModified, 0);
}
}
return result;
}
}
ConfigCacheService#removeBeta 方法源码解析
ConfigCacheService#removeBeta 方法的作用是删除配置信息，该方法会先获取写锁，防止并发操作，然后回判断是否是单机内嵌存储模式，如果不是则删除文件，然后发布 LocalDataChangeEvent 事件，最后释放锁。
//com.alibaba.nacos.config.server.service.ConfigCacheService#removeBeta
public static boolean removeBeta(String dataId, String group, String tenant) {
//获取 key
final String groupKey = GroupKey2.getKey(dataId, group, tenant);
//写锁
final int lockResult = tryWriteLock(groupKey);
// If data is non-existent.
if (0 == lockResult) {
//要 remove 的数据不存在
DUMP_LOG.info("[remove-ok] {} not exist.", groupKey);
return true;
}
// try to lock failed
if (lockResult < 0) {
//获取写锁失败
DUMP_LOG.warn("[remove-error] write lock failed. {}", groupKey);
return false;
}
try {
if (!PropertyUtil.isDirectRead()) {
//不是单机模式 且不是内嵌数据存储 删除配置文件
DiskUtil.removeConfigInfo4Beta(dataId, group, tenant);
}
//使用通知中心发布 LocalDataChangeEvent 事件
NotifyCenter.publishEvent(new LocalDataChangeEvent(groupKey, true, CACHE.get(groupKey).getIps4Beta()));
//设置缓存值
CACHE.get(groupKey).setBeta(false);
CACHE.get(groupKey).setIps4Beta(null);
CACHE.get(groupKey).setMd54Beta(Constants.NULL);
return true;
} finally {
//释放锁
releaseWriteLock(groupKey);
}
}
ConfigCacheService#dumpBeta 方法源码解析
ConfigCacheService#dumpBeta 方法会先获取写锁，然后判断 MD5 值是否发生变化，如果 MD5 值没有发生变化，则不做处理，否则将配置信息写入到磁盘，最后会释放写锁。
//com.alibaba.nacos.config.server.service.ConfigCacheService#dumpBeta
public static boolean dumpBeta(String dataId, String group, String tenant, String content, long lastModifiedTs,
String betaIps) {
//获取 key
final String groupKey = GroupKey2.getKey(dataId, group, tenant);
//缓存 CACHE 中存在更新 不存在 加入
makeSure(groupKey);
//获取写锁
final int lockResult = tryWriteLock(groupKey);
assert (lockResult != 0);
if (lockResult < 0) {
//获取锁失败
DUMP_LOG.warn("[dump-beta-error] write lock failed. {}", groupKey);
return false;
}
try {
//获取 MD5 值
final String md5 = MD5Utils.md5Hex(content, Constants.ENCODE);
if (md5.equals(ConfigCacheService.getContentBetaMd5(groupKey))) {
//MD5 值没有变化 不做处理
DUMP_LOG.warn("[dump-beta-ignore] ignore to save cache file. groupKey={}, md5={}, lastModifiedOld={}, "
+ "lastModifiedNew={}", groupKey, md5, ConfigCacheService.getLastModifiedTs(groupKey),
lastModifiedTs);
} else if (!PropertyUtil.isDirectRead()) {
//不是单机 且不是内嵌数据库 配置信息数据写入磁盘
DiskUtil.saveBetaToDisk(dataId, group, tenant, content);
}
String[] betaIpsArr = betaIps.split(",");
//更新 md5
updateBetaMd5(groupKey, md5, Arrays.asList(betaIpsArr), lastModifiedTs);
return true;
} catch (IOException ioe) {
DUMP_LOG.error("[dump-beta-exception] save disk error. " + groupKey + ", " + ioe.toString(), ioe);
return false;
} finally {
//释放锁
releaseWriteLock(groupKey);
}
}
ConfigCacheService#dump 方法源码解析
ConfigCacheService#dump方法是 tag 为空的配置文件转储的方法，会先获取写锁，来保证线程安全和不被重复操作，获取锁成功后，则会获取 MD5 值进行比较，如果 MD5 值一致且配置文件存在，不做处理，否则会判断是否是本地读取，如果是将配置信息写入本地磁盘，更新 MD5值，发布 LocalDataChangeEvent 事件，并释放锁。
//com.alibaba.nacos.config.server.service.ConfigCacheService#dump
public static boolean dump(String dataId, String group, String tenant, String content, long lastModifiedTs,
String type) {
//获取 group
String groupKey = GroupKey2.getKey(dataId, group, tenant);
//groupKey 存在则更新 不存在加入到 CACHE
CacheItem ci = makeSure(groupKey);
//设置类型
ci.setType(type);
//获取写锁
final int lockResult = tryWriteLock(groupKey);
//写锁判断
assert (lockResult != 0);
if (lockResult < 0) {
//获取写锁失败
DUMP_LOG.warn("[dump-error] write lock failed. {}", groupKey);
return false;
}
try {
//获取 md5 值
final String md5 = MD5Utils.md5Hex(content, Constants.ENCODE);
if (md5.equals(ConfigCacheService.getContentMd5(groupKey)) && DiskUtil.targetFile(dataId, group, tenant).exists()) {
//md5 值一样 且文件存在
DUMP_LOG.warn("[dump-ignore] ignore to save cache file. groupKey={}, md5={}, lastModifiedOld={}, "
+ "lastModifiedNew={}", groupKey, md5, ConfigCacheService.getLastModifiedTs(groupKey),
lastModifiedTs);
} else if (!PropertyUtil.isDirectRead()) {
//进入 表示不是单机模式 也不是内嵌数据库
//保存文件到本地磁盘
DiskUtil.saveToDisk(dataId, group, tenant, content);
}
//更新 md5 值 发布 LocalDataChangeEvent 事件
updateMd5(groupKey, md5, lastModifiedTs);
return true;
} catch (IOException ioe) {
DUMP_LOG.error("[dump-exception] save disk error. " + groupKey + ", " + ioe.toString(), ioe);
if (ioe.getMessage() != null) {
String errMsg = ioe.getMessage();
if (NO_SPACE_CN.equals(errMsg) || NO_SPACE_EN.equals(errMsg) || errMsg.contains(DISK_QUATA_CN) || errMsg
.contains(DISK_QUATA_EN)) {
// Protect from disk full.
FATAL_LOG.error("磁盘满自杀退出", ioe);
System.exit(0);
}
}
return false;
} finally {
//释放写锁
releaseWriteLock(groupKey);
}
}
ConfigCacheService#remove 方法源码解析
ConfigCacheService#remove 方法是 tag 为空的删除配置的方法，该方法会先获取写锁，防止并发操作，然后回判断是否是单机内嵌存储模式，如果不是则删除文件，然后从缓存 CACHE 中移除，发布 LocalDataChangeEvent 事件，最后释放锁。
//com.alibaba.nacos.config.server.service.ConfigCacheService#remove
public static boolean remove(String dataId, String group, String tenant) {
//获取 key
final String groupKey = GroupKey2.getKey(dataId, group, tenant);
//获取写锁
final int lockResult = tryWriteLock(groupKey);
// If data is non-existent.
//数据不存在
if (0 == lockResult) {
DUMP_LOG.info("[remove-ok] {} not exist.", groupKey);
return true;
}
// try to lock failed
if (lockResult < 0) {
//加写锁失败
DUMP_LOG.warn("[remove-error] write lock failed. {}", groupKey);
return false;
}
try {
if (!PropertyUtil.isDirectRead()) {
//不是单机模式 且不是内嵌数据存储 删除配置文件
DiskUtil.removeConfigInfo(dataId, group, tenant);
}
//从缓存中删除
CACHE.remove(groupKey);
//使用通知中心发布 LocalDataChangeEvent 事件
NotifyCenter.publishEvent(new LocalDataChangeEvent(groupKey));
return true;
} finally {
//释放写锁
releaseWriteLock(groupKey);
}
}
ConfigCacheService#dumpTag 方法源码解析
ConfigCacheService#dumpTag 方法会先获取写锁，然后判断 MD5 值是否发生变化，如果 MD5 值没有发生变化，则不做处理，否则将配置信息写入到磁盘，最后会释放写锁。
//com.alibaba.nacos.config.server.service.ConfigCacheService#dumpTag
public static boolean dumpTag(String dataId, String group, String tenant, String tag, String content,
long lastModifiedTs) {
//获取 key
final String groupKey = GroupKey2.getKey(dataId, group, tenant);
//缓存 CACHE 中存在更新 不存在 加入
makeSure(groupKey);
//获取写锁
final int lockResult = tryWriteLock(groupKey);
assert (lockResult != 0);
if (lockResult < 0) {
//获取锁失败
DUMP_LOG.warn("[dump-tag-error] write lock failed. {}", groupKey);
return false;
}
try {
//获取 MD5 值
final String md5 = MD5Utils.md5Hex(content, Constants.ENCODE);
if (md5.equals(ConfigCacheService.getContentTagMd5(groupKey, tag))) {
//MD5 值没有变化 不做处理
DUMP_LOG.warn("[dump-tag-ignore] ignore to save cache file. groupKey={}, md5={}, lastModifiedOld={}, "
+ "lastModifiedNew={}", groupKey, md5, ConfigCacheService.getLastModifiedTs(groupKey),
lastModifiedTs);
} else if (!PropertyUtil.isDirectRead()) {
//不是单机 且不是内嵌数据库 配置信息数据写入磁盘
DiskUtil.saveTagToDisk(dataId, group, tenant, tag, content);
}
//更新 md5
updateTagMd5(groupKey, tag, md5, lastModifiedTs);
return true;
} catch (IOException ioe) {
DUMP_LOG.error("[dump-tag-exception] save disk error. " + groupKey + ", " + ioe.toString(), ioe);
return false;
} finally {
//释放锁
releaseWriteLock(groupKey);
}
}
ConfigCacheService#removeTag 方法源码解析
方法的作用是删除有 tag 的配置信息，该方法会先获取写锁，防止并发操作，然后回判断是否是单机内嵌存储模式，如果不是则删除文件，然后从缓存 CACHE 中移除，发布 LocalDataChangeEvent 事件，最后释放锁。
//com.alibaba.nacos.config.server.service.ConfigCacheService#removeTag
public static boolean removeTag(String dataId, String group, String tenant, String tag) {
//获取 kye
final String groupKey = GroupKey2.getKey(dataId, group, tenant);
//获取写锁
final int lockResult = tryWriteLock(groupKey);
// If data is non-existent.
if (0 == lockResult) {
//数据不存在
DUMP_LOG.info("[remove-ok] {} not exist.", groupKey);
return true;
}
// try to lock failed
if (lockResult < 0) {
//获取锁失败
DUMP_LOG.warn("[remove-error] write lock failed. {}", groupKey);
return false;
}
try {
if (!PropertyUtil.isDirectRead()) {
//不是单机模式 且不是内嵌数据存储 删除配置文件
DiskUtil.removeConfigInfo4Tag(dataId, group, tenant, tag);
}
//删除缓存
CacheItem ci = CACHE.get(groupKey);
ci.tagMd5.remove(tag);
ci.tagLastModifiedTs.remove(tag);
//使用通知中心发布 LocalDataChangeEvent 事件
NotifyCenter.publishEvent(new LocalDataChangeEvent(groupKey, false, null, tag));
return true;
} finally {
//释放锁
releaseWriteLock(groupKey);
}
}
至此，Nacos 配置中心配置发布的源码分析完毕，希望可以帮助到有需要的伙伴。
如有不正确的地方请各位指出纠正。
标签：dataId,group,String,groupKey,配置,tag,Nacos,源码,final From： https://blog.csdn.net/weixin_42118323/article/details/140585838