全网第一篇把Nacos配置中心服务端讲明白的-编程知识

入口

getServerConfig对应：ConfigQueryRequestHandler�

getBatchServiceConfig对应：ConfigChangeBatchListenResponse�

admin对应：ConfigController

我们重点就要2个，一个是服务端如何完成客户端获取配置请求，一个是服务端更新配置，客户端如何更新, 也就是说ConfigQueryReustHandler和ConfigController

ConfigQueryRequestHandler

ConfigQueryRequestHandler.getContext

private ConfigQueryResponse getContext(ConfigQueryRequest configQueryRequest, RequestMeta meta, boolean notify)throws UnsupportedEncodingException {// todoString dataId = configQueryRequest.getDataId();String group = configQueryRequest.getGroup();String tenant = configQueryRequest.getTenant();String clientIp = meta.getClientIp();String tag = configQueryRequest.getTag();ConfigQueryResponse response = new ConfigQueryResponse();final String groupKey = GroupKey2.getKey(configQueryRequest.getDataId(), configQueryRequest.getGroup(), configQueryRequest.getTenant());String autoTag = configQueryRequest.getHeader(com.alibaba.nacos.api.common.Constants.VIPSERVER_TAG);String requestIpApp = meta.getLabels().get(CLIENT_APPNAME_HEADER);int lockResult = tryConfigReadLock(groupKey);boolean isBeta = false;boolean isSli = false;if (lockResult > 0) {//FileInputStream fis = null;try {String md5 = Constants.NULL;long lastModified = 0L;// todo 从缓存中获取CacheItem cacheItem = ConfigCacheService.getContentCache(groupKey);if (cacheItem != null) {if (cacheItem.isBeta()) {if (cacheItem.getIps4Beta().contains(clientIp)) {isBeta = true;}}String configType = cacheItem.getType();response.setContentType((null != configType) ? configType : "text");}File file = null;ConfigInfoBase configInfoBase = null;PrintWriter out = null;if (isBeta) {md5 = cacheItem.getMd54Beta();lastModified = cacheItem.getLastModifiedTs4Beta();if (PropertyUtil.isDirectRead()) {configInfoBase = persistService.findConfigInfo4Beta(dataId, group, tenant);} else {file = DiskUtil.targetBetaFile(dataId, group, tenant);}response.setBeta(true);} else {if (StringUtils.isBlank(tag)) {if (isUseTag(cacheItem, autoTag)) {if (cacheItem != null) {if (cacheItem.tagMd5 != null) {md5 = cacheItem.tagMd5.get(autoTag);}if (cacheItem.tagLastModifiedTs != null) {lastModified = cacheItem.tagLastModifiedTs.get(autoTag);}}if (PropertyUtil.isDirectRead()) {configInfoBase = persistService.findConfigInfo4Tag(dataId, group, tenant, autoTag);} else {file = DiskUtil.targetTagFile(dataId, group, tenant, autoTag);}response.setTag(URLEncoder.encode(autoTag, Constants.ENCODE));} else {md5 = cacheItem.getMd5();lastModified = cacheItem.getLastModifiedTs();// todo 是否是数据读if (PropertyUtil.isDirectRead()) {configInfoBase = persistService.findConfigInfo(dataId, group, tenant);} else {file = DiskUtil.targetFile(dataId, group, tenant);}if (configInfoBase == null && fileNotExist(file)) {// FIXME CacheItem// No longer exists. It is impossible to simply calculate the push delayed. Here, simply record it as - 1.ConfigTraceService.logPullEvent(dataId, group, tenant, requestIpApp, -1,ConfigTraceService.PULL_EVENT_NOTFOUND, -1, clientIp, false);// pullLog.info("[client-get] clientIp={}, {},// no data",// new Object[]{clientIp, groupKey});response.setErrorInfo(ConfigQueryResponse.CONFIG_NOT_FOUND, "config data not exist");return response;}}} else {if (cacheItem != null) {if (cacheItem.tagMd5 != null) {md5 = cacheItem.tagMd5.get(tag);}if (cacheItem.tagLastModifiedTs != null) {Long lm = cacheItem.tagLastModifiedTs.get(tag);if (lm != null) {lastModified = lm;}}}if (PropertyUtil.isDirectRead()) {configInfoBase = persistService.findConfigInfo4Tag(dataId, group, tenant, tag);} else {file = DiskUtil.targetTagFile(dataId, group, tenant, tag);}if (configInfoBase == null && fileNotExist(file)) {// FIXME CacheItem// No longer exists. It is impossible to simply calculate the push delayed. Here, simply record it as - 1.ConfigTraceService.logPullEvent(dataId, group, tenant, requestIpApp, -1,ConfigTraceService.PULL_EVENT_NOTFOUND, -1, clientIp, false);// pullLog.info("[client-get] clientIp={}, {},// no data",// new Object[]{clientIp, groupKey});response.setErrorInfo(ConfigQueryResponse.CONFIG_NOT_FOUND, "config data not exist");return response;}}}response.setMd5(md5);if (PropertyUtil.isDirectRead()) {response.setLastModified(lastModified);response.setContent(configInfoBase.getContent());response.setResultCode(ResponseCode.SUCCESS.getCode());} else {//read from fileString content = null;try {content = readFileContent(file);response.setContent(content);response.setLastModified(lastModified);response.setResultCode(ResponseCode.SUCCESS.getCode());} catch (IOException e) {response.setErrorInfo(ResponseCode.FAIL.getCode(), e.getMessage());return response;}}LogUtil.PULL_CHECK_LOG.warn("{}|{}|{}|{}", groupKey, clientIp, md5, TimeUtils.getCurrentTimeStr());final long delayed = System.currentTimeMillis() - lastModified;// TODO distinguish pull-get && push-get/*Otherwise, delayed cannot be used as the basis of push delay directly,because the delayed value of active get requests is very large.*/ConfigTraceService.logPullEvent(dataId, group, tenant, requestIpApp, lastModified,ConfigTraceService.PULL_EVENT_OK, notify ? delayed : -1, clientIp, notify);} finally {releaseConfigReadLock(groupKey);}} else if (lockResult == 0) {// FIXME CacheItem No longer exists. It is impossible to simply calculate the push delayed. Here, simply record it as - 1.ConfigTraceService.logPullEvent(dataId, group, tenant, requestIpApp, -1, ConfigTraceService.PULL_EVENT_NOTFOUND, -1,clientIp, notify);response.setErrorInfo(ConfigQueryResponse.CONFIG_NOT_FOUND, "config data not exist");} else {PULL_LOG.info("[client-get] clientIp={}, {}, get data during dump", clientIp, groupKey);response.setErrorInfo(ConfigQueryResponse.CONFIG_QUERY_CONFLICT,"requested file is being modified, please try later.");}return response;}

这一段代码很长，感觉写的也不是很好，很冗长，总结一下：

通过groupKey从ConfigCacheService.getContentCache(groupKey)获取cacheItem，这里的cacheItem要等到pushConfig在看一下，具体用来做什么，只是做一些数据的暂存吗？
然后看了一下是否从数据读，如果是就从数据库读出配置，如果不是，就从文件中读取

�

PropertyUtil.isDirectRead

persistService.findConfigInfo

ConfigController

ConfigController.pushConfig

  @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 {// todo 2.2以上这里有个插件思想学一下，基本上也是spi的思想final String srcIp = RequestUtil.getRemoteIp(request);final String requestIpApp = RequestUtil.getAppName(request);srcUser = RequestUtil.getSrcUserName(request);//check typeif (!ConfigType.isValidType(type)) {type = ConfigType.getDefaultType().getType();}// check tenant//todo  参数检查ParamUtils.checkTenant(tenant);ParamUtils.checkParam(dataId, group, "datumId", content);ParamUtils.checkParam(tag);Map<String, Object> configAdvanceInfo = new HashMap<String, Object>(10);MapUtil.putIfValNoNull(configAdvanceInfo, "config_tags", configTags);MapUtil.putIfValNoNull(configAdvanceInfo, "desc", desc);MapUtil.putIfValNoNull(configAdvanceInfo, "use", use);MapUtil.putIfValNoNull(configAdvanceInfo, "effect", effect);MapUtil.putIfValNoNull(configAdvanceInfo, "type", type);MapUtil.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();String betaIps = request.getHeader("betaIps");// todo // 构造配置信息ConfigInfo configInfo = new ConfigInfo(dataId, group, tenant, appName, content);configInfo.setType(type);if (StringUtils.isBlank(betaIps)) {if (StringUtils.isBlank(tag)) {persistService.insertOrUpdate(srcIp, srcUser, configInfo, time, configAdvanceInfo, false);// todo 由AsyncNotifyService处理ConfigChangePublisher.notifyConfigChange(new ConfigDataChangeEvent(false, dataId, group, tenant, time.getTime()));} else {persistService.insertOrUpdateTag(configInfo, tag, srcIp, srcUser, time, false);// todo 由AsyncNotifyService处理ConfigChangePublisher.notifyConfigChange(new ConfigDataChangeEvent(false, dataId, group, tenant, tag, time.getTime()));}} else {// beta publishpersistService.insertOrUpdateBeta(configInfo, betaIps, srcIp, srcUser, time, false);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;}

一共做了这么几件事：

persistsService.insertOrUpdate
发送ConfigDataChangeEvent，这个事件最终由AsyncNotifyService执行到

AsyncNotifyService.onEvent

这里做了这么几件事：

memberManager.allMembers()：获取到所有的服务节点，我这里因为是单机启动的，所以就一台

看对应的服务是否支持长链接，如果支持就放到rpcQueue中，如果不支持，就放到httpQueue，基本到这里就知道后面要做什么，肯定又是什么线程要消费这里的queue
将rpcQueue放到AsyncRpcTask里面去，然后用一个异步线程池调用它

其中ConfigExecutor：

不往下追了，就是一个简单的定时调度线程池，反正最终也是调度到AsyncRpcTask的run方法，这个更重要

其中AsyncRpcTask：

每一个AsyncRpcTask里面有一个队列，这个队列里面存放的是NotifySingleRpcTask：就是rpc通知任务

AsyncRpcTask.run

这里主要做了几件事：

从queue中获取通知任务，如果任务目标是当前节点，就直接调用dumpService.dump
如果是其他服务节点的话，判断是否支持长链接，如果支持就发送rpc请求，如果不支持就发送http请求

DumpService.dump

比较简单，将task放到TaskManager中，然后TaskManager异步执行

TaskManager

首先DumpService里面有一个TaskManager，这个类是继承NacosDelayTaskExecuteEngine�，这个类之前是详细分析过的，就是延迟执行的，核心逻辑就是有一个线程不断的从一个map中获取任务，

如果获取到任务，就看这个taskKey有没有对应的处理器，如果没有设置，那就找一个默认的，

刚好DumpService为TaskManager设置了一个默认的Processor，位置在DumpService的构造方法上