Nacos14# 配置管理服务端流程
瓜农老梁
啃码、撸码、笔记
Official Account
引言
在上文分析中客户端会有长轮询，不断轮询阻塞队列「listenExecutebell」去比较客户端和服务端配置内容md5是否一致，不一致则通知我们注册的Listener完成回调。当阻塞队里有元素时会立即执行，没有元素等待5秒钟执行。那都在什么时候往队列中添加元素的从而触发立即执行。
长轮询回顾
-
监听配置内容的变更通过 长轮询实现 -
具体实现为轮询阻塞队列
阻塞队列添加时机
-
客户端添加Listener时添加 -
客户端删除Listener时添加 -
服务端通知内容变更时添加 -
建立gRPC连接时添加
服务端变更发布流程
-
通过配置中心发起变更请求 -
配置变更内容被写入数据库 -
向本节点连接的Client发送变更通知 -
向集群中的其他节点发送变更通知
向Client发送变更通知
-
首先构建DumpTask -
处理DumpTask时发布ConfigDumpEvent事件 -
处理ConfigDumpEvent时发布LocalDataChangeEvent事件 -
处理LocalDataChangeEvent事件：@1 先从缓存中拿出注册Client列表 @2 根据client从缓存获取gRPC连接 @3 构建ConfigChangeNotifyRequest @4 通过gRPC向客户端发送变更通知 -
客户端处理变更通知， 客户端接到请求后向阻塞队列添加new Object元素，详细客户端的长轮询流程见上篇
向其他节点发送变更通知
-
通过gRPC向集群中其他节点发送变更通知 -
集群节点收到变更通知后再下发给连接自己的Client
长轮询判断
public void startInternal() throws NacosException {
executor.schedule(new Runnable() {
@Override
public void run() {
while (true) { // 一直运行
try {
// 最长等待5秒
listenExecutebell.poll(5L, TimeUnit.SECONDS);
executeConfigListen();
} catch (Exception e) {
LOGGER.error("[ rpc listen execute ] [rpc listen] exception", e);
}
}
}
}, 0L, TimeUnit.MILLISECONDS);
}
队列添加元素
@Override
public void notifyListenConfig() {
listenExecutebell.offer(bellItem);
}
备注： 配置内容变更机制是通过长轮询实现的。
1.客户端添加Listener时添加
public void addListeners(String dataId, String group, List<? extends Listener> listeners) {
// ...
// 通过队列中增加一个对象，长轮询立即执行
agent.notifyListenConfig();
}
}
2.客户端删除Listener时添加
public void removeListener(String dataId, String group, Listener listener) {
group = null2defaultGroup(group);
CacheData cache = getCache(dataId, group);
if (null != cache) {
synchronized (cache) {
cache.removeListener(listener);
if (cache.getListeners().isEmpty()) {
cache.setSyncWithServer(false);
agent.removeCache(dataId, group);
}
}
}
}
public void removeCache(String dataId, String group) {
// Notify to rpc un listen ,and remove cache if success.
notifyListenConfig();
}
3.服务端通知内容变更时添加
在与服务端建立gRPC通道时会添加Handler用于处理服务端推送的请求，收到推送请求后向阻塞队列添加元素。
private void initRpcClientHandler(final RpcClient rpcClientInner) {
/*
* Register Config Change /Config ReSync Handler
*/
rpcClientInner.registerServerRequestHandler((request) -> {
if (request instanceof ConfigChangeNotifyRequest) {
ConfigChangeNotifyRequest configChangeNotifyRequest = (ConfigChangeNotifyRequest) request;
LOGGER.info("[{}] [server-push] config changed. dataId={}, group={},tenant={}",
rpcClientInner.getName(), configChangeNotifyRequest.getDataId(),
configChangeNotifyRequest.getGroup(), configChangeNotifyRequest.getTenant());
String groupKey = GroupKey
.getKeyTenant(configChangeNotifyRequest.getDataId(), configChangeNotifyRequest.getGroup(),
configChangeNotifyRequest.getTenant());
CacheData cacheData = cacheMap.get().get(groupKey);
if (cacheData != null) {
cacheData.setSyncWithServer(false);
notifyListenConfig(); // 向阻塞队列中添加元素
}
return new ConfigChangeNotifyResponse();
}
return null;
});
// ....
}
4.建立gRPC连接时添加
在建立gRPC连接时会向阻塞队列中添加元素。
rpcClientInner.registerConnectionListener(new ConnectionEventListener() {
@Override
public void onConnected() {
LOGGER.info("[{}] Connected,notify listen context...", rpcClientInner.getName());
notifyListenConfig(); // 向队列中添加元素
}
@Override
public void onDisConnect() {
String taskId = rpcClientInner.getLabels().get("taskId");
LOGGER.info("[{}] DisConnected,clear listen context...", rpcClientInner.getName());
Collection<CacheData> values = cacheMap.get().values();
for (CacheData cacheData : values) {
if (StringUtils.isNotBlank(taskId)) {
if (Integer.valueOf(taskId).equals(cacheData.getTaskId())) {
cacheData.setSyncWithServer(false);
}
} else {
cacheData.setSyncWithServer(false);
}
}
}
备注： 阻塞队列添加元素的四种情况，添加后长轮询立即执行配置内容校验。
1.发布配置变更请求
服务端的发布内容变更，通过变更请求跟踪下，后端处理请求通过ConfigController#publishConfig()实现。
2.处理配置变更请求
-
将变更配置更新到数据库 -
发布ConfigDataChangeEvent事件
3.订阅ConfigDataChangeEvent
翻到订阅该事件的处理逻辑，看看其干了什么事情，AsyncNotifyService构造函数中的onEvent()方法。从代码中可以看出当收到ConfigDataChangeEvent时，逻辑如下：
-
获取集群中所有节点列表 -
每个节点封装NotifySingleRpcTask -
异步向各个节点通知配置信息变更
4.NotifySingleRpcTask逻辑
在代码AsyncNotifyService#AsyncRpcTask部分查看逻辑框架，任务运行主要干了两件事：
-
向连接本节点的 Client发送配置变更通知 -
向集群中的其他节点发送变更通知
小结： 当我们从Nacos配置中心界面提交变更配置时：1.入库；2.向本节点连接的Client发送变更通知；3.向集群中的其他节点发送变更通知。
1.首先构建DumpTask
代码坐标DumpService#dump()
public void dump(String dataId, String group, String tenant, String tag, long lastModified, String handleIp,
boolean isBeta) {
String groupKey = GroupKey2.getKey(dataId, group, tenant);
dumpTaskMgr.addTask(groupKey, new DumpTask(groupKey, tag, lastModified, handleIp, isBeta));
}
2.处理DumpTask时发布ConfigDumpEvent事件
代码坐标DumpProcessor#process()
3.处理ConfigDumpEvent时发布LocalDataChangeEvent事件
4.处理LocalDataChangeEvent事件
关注RpcConfigChangeNotifier#onEvent处理过程
@Override
public void onEvent(LocalDataChangeEvent event) {
String groupKey = event.groupKey;
boolean isBeta = event.isBeta;
List<String> betaIps = event.betaIps;
String[] strings = GroupKey.parseKey(groupKey);
String dataId = strings[0];
String group = strings[1];
String tenant = strings.length > 2 ? strings[2] : "";
String tag = event.tag;
configDataChanged(groupKey, dataId, group, tenant, isBeta, betaIps, tag);
}
例如：注册在缓存的client=1627734181270_127.0.0.1_53522
5.客户端处理变更通知
小结： 走查了服务端如何将变更通知送达到连接该节点的客户端的。
1.接上面「服务端变更发布流程」最后步骤「向集群中的其他节点发送变更通知」
configClusterRpcClientProxy
.syncConfigChange(member, syncRequest, new AsyncRpcNotifyCallBack(task));
从缓存中获取与其他节点的rpcClient连接，key=Cluster-1.2.3.4:1111
public void asyncRequest(Member member, Request request, RequestCallBack callBack) throws NacosException {
RpcClient client = RpcClientFactory.getClient(memberClientKey(member));
if (client != null) {
client.asyncRequest(request, callBack);
} else {
throw new NacosException(CLIENT_INVALID_PARAM, "No rpc client related to member: " + member);
}
}
2.其他节点处理请求入口
GrpcRequestAcceptor#request只列出部分代码
// 执行RequestHandler逻辑
Response response = requestHandler.handleRequest(request, requestMeta);
3.其他节点由ConfigChangeClusterSyncRequestHandler处理ConfigChangeClusterSyncRequest
即：完成向客户端发送变更通知
小结： 当集群其他节点收到变更通知后，会向连接到本节点的Client发送变更通知，逻辑同上一小结「向Client发送变更通知」，从而完成全部向所有Client下发变更通知。
瓜农老梁
啃码、撸码、笔记
Official Account