ToB企服应用市场:ToB评测及商务社交产业平台
标题:
Nacos源码 (4) 配置中心
[打印本页]
作者:
八卦阵
时间:
2023-8-30 21:48
标题:
Nacos源码 (4) 配置中心
本文阅读nacos-2.0.2的config源码,编写示例,分析推送配置、监听配置的原理。
客户端
创建NacosConfigService对象
Properties properties = new Properties();
properties.setProperty(PropertyKeyConst.SERVER_ADDR, NACOS_HOST);
NacosConfigService configService = new NacosConfigService(properties);
复制代码
构造方法:
public NacosConfigService(Properties properties) throws NacosException {
ValidatorUtils.checkInitParam(properties);
initNamespace(properties);
this.configFilterChainManager = new ConfigFilterChainManager(properties);
ServerListManager serverListManager = new ServerListManager(properties);
serverListManager.start();
this.worker = new ClientWorker(this.configFilterChainManager, serverListManager, properties);
// will be deleted in 2.0 later versions
agent = new ServerHttpAgent(serverListManager);
}
复制代码
创建ConfigFilterChainManager - 过滤器链
创建ServerListManager - 服务器列表管理
创建ClientWorker - 用来发送请求,内部封装了一个ConfigRpcTransportClient类型对象agent,它能够获取到RpcClient与服务端进行通信
推送配置
Properties properties = new Properties();
properties.setProperty(PropertyKeyConst.SERVER_ADDR, NACOS_HOST);
NacosConfigService configService = new NacosConfigService(properties);StringWriter out = new StringWriter();properties.store(out, "test config");// 推送配置到nacos服务器configService.publishConfig( ORDER_SERVICE, Constants.DEFAULT_GROUP, out.toString(), "properties");
复制代码
推送配置到nacos服务器:
public boolean publishConfig(String dataId,
String group,
String content,
String type) throws NacosException {
return publishConfigInner(namespace, dataId, group, null, null, null, content, type, null);
}
private boolean publishConfigInner(String tenant, String dataId, String group, String tag, String appName,
String betaIps, String content, String type, String casMd5) throws NacosException {
group = blank2defaultGroup(group);
ParamUtils.checkParam(dataId, group, content);
ConfigRequest cr = new ConfigRequest();
cr.setDataId(dataId);
cr.setTenant(tenant);
cr.setGroup(group);
cr.setContent(content);
cr.setType(type);
configFilterChainManager.doFilter(cr, null);
content = cr.getContent();
String encryptedDataKey = (String) cr.getParameter("encryptedDataKey");
return worker.publishConfig(
dataId, group, tenant, appName, tag, betaIps, content, encryptedDataKey, casMd5, type);
}
复制代码
worker使用agent推送配置:
// 1. 封装ConfigPublishRequest对象
ConfigPublishRequest request = new ConfigPublishRequest(dataId, group, tenant, content);
request.setCasMd5(casMd5);
request.putAdditionalParam(TAG_PARAM, tag);
request.putAdditionalParam(APP_NAME_PARAM, appName);
request.putAdditionalParam(BETAIPS_PARAM, betaIps);
request.putAdditionalParam(TYPE_PARAM, type);
request.putAdditionalParam(ENCRYPTED_DATA_KEY_PARAM, encryptedDataKey);
// 2. 获取RpcClient对象
// 3. 使用RpcClient发请求
ConfigPublishResponse response = (ConfigPublishResponse) requestProxy(getOneRunningClient(), request);
复制代码
监听配置
Properties properties = new Properties();
properties.setProperty(PropertyKeyConst.SERVER_ADDR, NACOS_HOST);
NacosConfigService configService = new NacosConfigService(properties);// 添加监听器configService.addListener(ORDER_SERVICE, Constants.DEFAULT_GROUP, new AbstractListener() { @Override public void receiveConfigInfo(String configInfo) { System.out.printf(">> config: \n%s\n\n", configInfo); }});
复制代码
将监听器注册到本地,本地使用CacheData作为监听器管理器,封装配置名、dataId和监听器集合,内部使用CopyOnWriteArrayList保存监听器,如果是第一次监听,会先拉取一次配置。本地注册表为Map结构,使用dataId+group+tenant作为key,value是CacheData对象
在client初始化阶段,会注册一个ServerRequestHandler,专门处理服务器端的ConfigChangeNotifyRequest请求,该请求只会推送变化了的配置的基本信息,而不包括内容,所以此处还会触发一次ConfigBatchListenRequest请求
之后查找到变化的配置后,再发一个查询请求拉取配置
服务端
推送配置处理器
ConfigPublishRequestHandler处理器
配置中心使用ConfigPublishRequestHandler类处理客户端配置推送请求:
验证请求参数
封装configAdvanceInfo配置扩展信息
创建ConfigInfo封装配置信息
使用持久层对象insert或者update配置
使用ConfigChangePublisher推送一个ConfigDataChangeEvent事件
ConfigChangePublisher.notifyConfigChange(
new ConfigDataChangeEvent(false, dataId, group, tenant, time.getTime()));
复制代码
ConfigDataChangeEvent事件处理
ConfigDataChangeEvent事件会触发数据dump操作和集群同步操作。
此处先介绍数据dump操作:
dumpService.dump(syncRequest.getDataId(), syncRequest.getGroup(), syncRequest.getTenant(),
syncRequest.getTag(), syncRequest.getLastModified(), NetUtils.localIP());
复制代码
dumpService是Dump data service用于备份数据:
// Add DumpTask to TaskManager, it will execute asynchronously.
public void dump(String dataId, String group, String tenant, String tag, long lastModified, String handleIp,
boolean isBeta) {
String groupKey = GroupKey2.getKey(dataId, group, tenant);
String taskKey = String.join("+", dataId, group, tenant, String.valueOf(isBeta), tag);
// 推送一个DumpTask
dumpTaskMgr.addTask(taskKey, new DumpTask(groupKey, tag, lastModified, handleIp, isBeta));
}
复制代码
DumpTask将使用DumpProcessor类处理:
把数据写到磁盘
推送LocalDataChangeEvent事件
LocalDataChangeEvent事件会被以下几个类处理:
LongPollingService$1会触发DataChangeTask,响应长轮询订阅者
RpcConfigChangeNotifier将数据变化推送给rpc订阅者
InternalConfigChangeNotifier处理内部配置文件变化
监听配置处理器
ConfigChangeBatchListenRequestHandler处理器
处理客户端的配置监听请求:
public class ConfigChangeBatchListenRequestHandler
extends RequestHandler<ConfigBatchListenRequest, ConfigChangeBatchListenResponse> {
@Autowired
private ConfigChangeListenContext configChangeListenContext;
@TpsControl(pointName = "ConfigListen")
@Secured(action = ActionTypes.READ, parser = ConfigResourceParser.class)
public ConfigChangeBatchListenResponse handle(
ConfigBatchListenRequest configChangeListenRequest, RequestMeta meta)
throws NacosException {
String connectionId = StringPool.get(meta.getConnectionId());
String tag = configChangeListenRequest.getHeader(Constants.VIPSERVER_TAG);
ConfigChangeBatchListenResponse configChangeBatchListenResponse =
new ConfigChangeBatchListenResponse();
for (ConfigBatchListenRequest.ConfigListenContext listenContext : configChangeListenRequest
.getConfigListenContexts()) {
String groupKey = GroupKey2
.getKey(listenContext.getDataId(), listenContext.getGroup(), listenContext.getTenant());
groupKey = StringPool.get(groupKey);
String md5 = StringPool.get(listenContext.getMd5());
// 监听
if (configChangeListenRequest.isListen()) {
// 注册监听器,维护groupKey->connectionId集关系和groupKey->md5关系
configChangeListenContext.addListen(groupKey, md5, connectionId);
// 判断变化
boolean isUptoDate = ConfigCacheService.isUptodate(groupKey, md5, meta.getClientIp(), tag);
if (!isUptoDate) {
// 把变化的配置基本信息添加到响应
configChangeBatchListenResponse
.addChangeConfig(listenContext.getDataId(), listenContext.getGroup(),
listenContext.getTenant());
}
} else {
// 取消监听
configChangeListenContext.removeListen(groupKey, connectionId);
}
}
return configChangeBatchListenResponse;
}
}
复制代码
免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!
欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)
Powered by Discuz! X3.4