前言
- 当配置中心的内容发生变更时，客户端是如何获取到最新内容的？
- 监听数据变更的 Long-Polling 长轮询是如何实现的？
- 在客户端集群模式中，如何做到只更改某一台客户端的配置内容？
- 当 Nacos 挂掉后，客户端还可以获取数据吗？
简介
动态配置服务是 Nacos 其中的关键特性之一，动态配置服务可以让您以中心化、外部化和动态化的方式管理所有环境的应用配置和服务配置。动态配置消除了配置变更时重新部署应用和服务的需要，让配置管理变得更加高效和敏捷。配置中心化管理让实现无状态服务变得更简单，让服务按需弹性扩展变得更容易。
我们可以通过 Nacos 控制台来发布配置，也可以使用 Nacos 提供的 REST 接口来发布配置。
发布配置
curl -X POST "http://127.0.0.1:8848/nacos/v1/cs/configs?dataId=nacos.cfg.dataId&group=test&content=HelloWorld"
获取配置
curl -X GET "http://127.0.0.1:8848/nacos/v1/cs/configs?dataId=nacos.cfg.dataId&group=test"
Nacos 配置中心分为服务端和客户端，服务端提供 REST 接口查询、更改配置，客户端SDK通过封装了服务端的 REST 接口来获取配置
客户端实现原理
使用SDK获取配置demo
String serverAddr = "{serverAddr}";
String dataId = "{dataId}";
String group = "{group}";
Properties properties = new Properties();
properties.put("serverAddr", serverAddr);
//创建ConfigService
ConfigService configService = NacosFactory.createConfigService(properties);
//获取dataId的配置
String content = configService.getConfig(dataId, group, 5000);
System.out.println(content);
//动态监听配置，当dataId数据发生变更时会调用receiveConfigInfo方法
configService.addListener(dataId, group, new Listener() {
@Override
public void receiveConfigInfo(String configInfo) {
System.out.println("recieve1:" + configInfo);
}
@Override
public Executor getExecutor() {
return null;
}
});demo 中首先获取了 dataId 的配置内容，并且为该内容添加了监听器，当dataId内容发生变化时会回调 receiveConfigInfo 方法获取最新的内容
获取配置原理解析
获取配置的主要方法是 NacosConfigService 类的 getConfigInner 方法，该方法优先从本地文件中获取配置，如果没有本地文件，则通过 HTTP REST 接口从服务端获取配置并将配置保存到本地快照文件中，如果从服务端获取配置失败，则会从快照文件中获取配置。
@Override
public String getConfig(String dataId, String group, long timeoutMs) throws NacosException {
return getConfigInner(namespace, dataId, group, timeoutMs);
}ConfigService 的 getConfig 方法调用的是 getConfigInner 方法
private String getConfigInner(String tenant, String dataId, String group, long timeoutMs) throws NacosException {
group = null2defaultGroup(group);
ParamUtils.checkKeyParam(dataId, group);
ConfigResponse cr = new ConfigResponse();
cr.setDataId(dataId);
cr.setTenant(tenant);
cr.setGroup(group);
// 1 优先使用本地配置
String content = LocalConfigInfoProcessor.getFailover(agent.getName(), dataId, group, tenant);
if (content != null) {
LOGGER.warn("[{}] [get-config] get failover ok, dataId={}, group={}, tenant={}, config={}", agent.getName(),
dataId, group, tenant, ContentUtils.truncateContent(content));
cr.setContent(content);
configFilterChainManager.doFilter(null, cr);
content = cr.getContent();
return content;
}
//2 如果没有本地配置，则获取服务器中的配置
try {
String[] ct = worker.getServerConfig(dataId, group, tenant, timeoutMs);
cr.setContent(ct[0]);
configFilterChainManager.doFilter(null, cr);
content = cr.getContent();
return content;
} catch (NacosException ioe) {
if (NacosException.NO_RIGHT == ioe.getErrCode()) {
throw ioe;
}
LOGGER.warn("[{}] [get-config] get from server error, dataId={}, group={}, tenant={}, msg={}",
agent.getName(), dataId, group, tenant, ioe.toString());
}
//3 如果获取服务器配置失败，则获取本地的快照数据
LOGGER.warn("[{}] [get-config] get snapshot ok, dataId={}, group={}, tenant={}, config={}", agent.getName(),
dataId, group, tenant, ContentUtils.truncateContent(content));
content = LocalConfigInfoProcessor.getSnapshot(agent.getName(), dataId, group, tenant);
cr.setContent(content);
configFilterChainManager.doFilter(null, cr);
content = cr.getContent();
return content;
}getConfigInner 方法优先从本地文件获取配置，本地文件默认是不存在的，因此如果想用本地配置覆盖远程配置只需要在本地新建配置文件即可，Nacos 会优先使用本地文件，本地文件配置的路径为：
/{user.home}/nacos/config/fixed-{serverName}/data/config-data/{group}/{dataId}如果没有本地配置，则调用 REST 接口从服务器中获取配置，调用接口为：
/v1/cs/configs?dataId={dataId}&group={group}如果从服务器获取配置失败，则从本地快照数据中获取，每次从服务器获取数据时都会更新本地快照数据，快照文件的路径为：
/{user.home}/nacos/config/fixed-{serverName}/snapshot/{group}/{dataId}getConfig 方法只是获取一次配置文件内容，当配置发生变更后还需要通过上面添加的监听器来获得最新的配置
监听配置原理解析
当通过 addListener 注册了监听器后，NacosConfigService 类会使用 ClientWorker 类的 checkConfigInfo 方法创建 LongPollingRunnable 长轮询线程去监听服务端的配置，默认3000个数据为一组创建一个 LongPollingRunnable 线程，长轮询连接默认超时时间为30秒，在30秒内如果监听的数据有任何变化会立即返回最新的数据，如果30秒内数据没有任何变化，则会结束当前的监听并开启下一轮监听。
在 NacosConfigService 类初始化时创建了 ClientWorke r对象，ClientWorker 负责获取 Nacos 的配置以及创建长轮询连接监听Client中所有使用过的配置。
this.worker = new ClientWorker(this.agent, this.configFilterChainManager, properties)NacosConfigService 中初始化 ClientWorker 对象，在 ClientWorker 构造方法中开启了以10毫秒的间隔去创建一个默认超时事件为30秒的长轮询连接去监听本地数据的变化，当数据发生变化时则更新本地数据，否则继续监听。
this.executor.scheduleWithFixedDelay(new Runnable() {
public void run() {
try {
ClientWorker.this.checkConfigInfo();
} catch (Throwable var2) {
ClientWorker.LOGGER.error("[" + agent.getName() + "] [sub-check] rotate check error", var2);
}
}
}, 1L, 10L, TimeUnit.MILLISECONDS);创建监听线程监听配置变化
public void checkConfigInfo() {
// Dispatch taskes.
int listenerSize = cacheMap.get().size();
//把本地数据进行分组，默认每3000个数据为一组，每组会开启一个长轮询监听线程
int longingTaskCount = (int) Math.ceil(listenerSize / ParamUtil.getPerTaskConfigSize());
if (longingTaskCount > currentLongingTaskCount) {
for (int i = (int) currentLongingTaskCount; i < longingTaskCount; i++) {
// 开启长轮询线程
executorService.execute(new LongPollingRunnable(i));
}
currentLongingTaskCount = longingTaskCount;
}
}LongPollingRunnable 是长轮询线程
// 使用长轮询获取变更的数据列表
List<String> changedGroupKeys = checkUpdateDataIds(cacheDatas, inInitializingCacheList);
for (String groupKey : changedGroupKeys) {
//…………
//通过变更的数据key调用REST接口获取最新的数据内容
String[] ct = getServerConfig(dataId, group, tenant, 3000L);
CacheData cache = cacheMap.get().get(GroupKey.getKeyTenant(dataId, group, tenant));
cache.setContent(ct[0]);
//…………
}
//开启下一轮30秒长轮询监听
executorService.execute(this);
LongPollingRunnable 线程首先通过 checkUpdateDataIds 中的长轮询连接监听数据，如果数据有变更则更新本地数据，否则开启下一轮的监听 checkUpdateDataIds方 法调用的是 checkUpdateConfigStr 开启长轮询监听
List<String> checkUpdateConfigStr(String probeUpdateString, boolean isInitializingCacheList) throws Exception {
Map<String, String> params = new HashMap<String, String>(2);
//监听配置的
params.put(Constants.PROBE_MODIFY_REQUEST, probeUpdateString);
Map<String, String> headers = new HashMap<String, String>(2);
//长轮询超时时间 默认30秒
headers.put("Long-Pulling-Timeout", "" + timeout);
//………………
try {
//调用 Nacos服务端的 /listener 接口开始监听配置变化
long readTimeoutMs = timeout + (long) Math.round(timeout >> 1);
HttpRestResult<String> result = agent
.httpPost(Constants.CONFIG_CONTROLLER_PATH + "/listener", headers, params, agent.getEncode(),
readTimeoutMs);
//返回监听的内容，如果配置发生了变化那么result就是最新的数据，如果配置没有发生变化那么result=null
if (result.ok()) {
setHealthServer(true);
return parseUpdateDataIdResponse(result.getData());
} else {
setHealthServer(false);
LOGGER.error("[{}] [check-update] get changed dataId error, code: {}", agent.getName(),
result.getCode());
}
} catch (Exception e) {
setHealthServer(false);
LOGGER.error("[" + agent.getName() + "] [check-update] get changed dataId exception", e);
throw e;
}
return Collections.emptyList();
}checkUpdateConfigStr 方法发起长轮询连接来监听 Nacos 的配置是否有变化，如果在30秒内配置发生了变化则会立即返回新的数据，如果在30秒内没有任何数据变化，则会返回 NULL，同时会开启下一轮30秒的监听。
服务端实现原理
ConfigController 类提供了发布&获取配置的 REST 接口，我们分别看下发布配置和获取配置的实现原理。
发布配置原理解析
发布配置时会先把数据持久化到存储引擎上，一般是mysql或者是Nacos内置的derby数据库，完成数据持久化之后会将数据变更包装成 ConfigDataChangeEvent 事件，通过 NotifyCenter.publishEvent 向外广播数据变更事件，所有订阅了 ConfigDataChangeEvent 事件的消费方会收到数据变更事件。
ConfigDataChangeEvent 事件订阅者在收到事件消息后，会先通过 HTTP REST 接口通知 Nacos 集群中的所有机器，集群在接收到通知后会先更新本地的内存数据，然后将数据变更事件包装成 LocalDataChangeEvent 事件通过 NotifyCenter.publishEvent 向外广播本地数据变更事件，所有订阅了 LocalDataChangeEvent 事件的消费方会收到数据变更事件。
LocalDataChangeEvent 事件订阅者在收到事件消息后，会创建一个线程来遍历所有的客户端长轮询连接监听的数据是否包含此次事件中的变更数据，如果变更的数据有客户端正在监听，则直接通过长连接把数据返回给客户端。
发布配置的 REST 接口为 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 {
/*
* 省略
*/
final String srcIp = RequestUtil.getRemoteIp(request);
final String requestIpApp = RequestUtil.getAppName(request);
/*
* 省略
*/
final Timestamp time = TimeUtils.getCurrentTime();
String betaIps = request.getHeader("betaIps");
ConfigInfo configInfo = new ConfigInfo(dataId, group, tenant, appName, content);
configInfo.setType(type);
//更新持久化数据
persistService.insertOrUpdateBeta(configInfo, betaIps, srcIp, srcUser, time, true);
//广播数据变更事件
ConfigChangePublisher
.notifyConfigChange(new ConfigDataChangeEvent(true, dataId, group, tenant, time.getTime()));
return true;
}publishConfig 方法先将数据进行持久化，然后将数据变更包装成 ConfigDataChangeEvent 事件通过 NotifyCenter.publishEvent 向外广播，ConfigDataChangeEvent 事件的消息订阅类是 AsyncNotifyService，它在构造方法中调用 NotifyCenter.registerSubscriber 注册了 Subscriber 事件处理类，因此onEvent调用的是 AsyncNotifyService.onEvent 方法
public AsyncNotifyService(ServerMemberManager memberManager) {
this.memberManager = memberManager;
// 订阅事件类型
NotifyCenter.registerToPublisher(ConfigDataChangeEvent.class, NotifyCenter.ringBufferSize);
// 订阅事件处理类
NotifyCenter.registerSubscriber(new Subscriber() {
@Override
public void onEvent(Event event) {
// 只处理事件为ConfigDataChangeEvent的类
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集群列表，因为我是单机运行模式，所以ipList是本机节点
Collection<Member> ipList = memberManager.allMembers();
// 将ConfigDataChangeEvent事件转换为NotifySingleTask任务并将任务放入到队列中
Queue<NotifySingleTask> queue = new LinkedList<NotifySingleTask>();
// 每个集群都创建一个NotifySingleTask任务
for (Member member : ipList) {
queue.add(new NotifySingleTask(dataId, group, tenant, tag, dumpTs, member.getAddress(),
evt.isBeta));
}
// 将队列数据保证成AsyncTask对象并使用线程池执行AsyncTask的run方法
ConfigExecutor.executeAsyncNotify(new AsyncTask(nacosAsyncRestTemplate, queue));
}
}
});
}AsyncNotifyService 的 onEvent 方法在接收到事件数据后将数据包装成 AsyncTask 任务，并使用线程池处理 AsyncTask，如果 Nacos 集群存在多个N个节点，则相应创建N个 AsyncTask 任务
class AsyncTask implements Runnable {
@Override
public void run() {
executeAsyncInvoke();
}
private void executeAsyncInvoke() {
while (!queue.isEmpty()) {
NotifySingleTask task = queue.poll();
String targetIp = task.getTargetIP();
if (memberManager.hasMember(targetIp)) {
// 检查集群中当前服务是否健康，如果服务是下线状态则延时执行任务
boolean unHealthNeedDelay = memberManager.isUnHealth(targetIp);
if (unHealthNeedDelay) {
ConfigTraceService.logNotifyEvent(task.getDataId(), task.getGroup(), task.getTenant(), null,
task.getLastModified(), InetUtils.getSelfIP(), ConfigTraceService.NOTIFY_EVENT_UNHEALTH,
0, task.target);
// 后台延时执行任务
asyncTaskExecute(task);
} else {
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());
if (task.isBeta) {
header.addParam("isBeta", "true");
}
//调用集群的dataChange REST接口
//task.url = http://127.0.0.1:8848/nacos/v1/cs/communication/dataChange?dataId=nacos.cfg.dataId.1&group=test
restTemplate.get(task.url, header, Query.EMPTY, String.class, new AsyncNotifyCallBack(task));
}
}
}
}
}AsyncTask 会根据当前集群节点的健康状态来延时或者直接调用集群节点的 /nacos/v1/cs/communication/dataChange REST接口来更新每个集群中的内存数据
@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) {
dataId = dataId.trim();
group = group.trim();
String lastModified = request.getHeader(NotifyService.NOTIFY_HEADER_LAST_MODIFIED);
long lastModifiedTs = StringUtils.isEmpty(lastModified) ? -1 : Long.parseLong(lastModified);
String handleIp = request.getHeader(NotifyService.NOTIFY_HEADER_OP_HANDLE_IP);
String isBetaStr = request.getHeader("isBeta");
//调用dump方法更新集群节点的
if (StringUtils.isNotBlank(isBetaStr) && trueStr.equals(isBetaStr)) {
dumpService.dump(dataId, group, tenant, lastModifiedTs, handleIp, true);
} else {
dumpService.dump(dataId, group, tenant, tag, lastModifiedTs, handleIp);
}
return true;
}dataChange REST 接口中调用了 dumpService.dump 方法来更新节点的内存数据
public void dump(String dataId, String group, String tenant, long lastModified, String handleIp, boolean isBeta) {
String groupKey = GroupKey2.getKey(dataId, group, tenant);
//将数据包装成DumpTask任务并加入到dumpTaskMgr任务列表中
dumpTaskMgr.addTask(groupKey, new DumpTask(groupKey, lastModified, handleIp, isBeta));
}dump 方法将数据包装成 DumpTask 任务并加入到 dumpTaskMgr 任务列表中，由父类 NacosDelayTaskExecuteEngine 的 processingExecutor 线程池按照100毫秒的间隔执行 processTasks 方法
protected void processTasks() {
Collection<Object> keys = getAllTaskKeys();
for (Object taskKey : keys) {
//删除task
AbstractDelayTask task = removeTask(taskKey);
if (null == task) {
continue;
}
//获取task对象应的Processor
NacosTaskProcessor processor = getProcessor(taskKey);
if (null == processor) {
getEngineLog().error("processor not found for task, so discarded. " + task);
continue;
}
try {
//如果执行失败 加入重试任务
if (!processor.process(task)) {
retryFailedTask(taskKey, task);
}
} catch (Throwable e) {
getEngineLog().error("Nacos task execute error : " + e.toString(), e);
retryFailedTask(taskKey, task);
}
}
}processTasks 每100毫秒会执行一次，获取 Task 对应的 Processor 处理类并执行其 process 方法，如果执行失败则重新加入到任务队列中，其中默认的 Processor 为 DumpProcessor，因此 processor.process 调用的是 DumpProcessor.process 方法
public class DumpProcessor implements NacosTaskProcessor {
public DumpProcessor(DumpService dumpService) {
this.dumpService = dumpService;
}
@Override
public boolean process(NacosTask task) {
final PersistService persistService = dumpService.getPersistService();
DumpTask dumpTask = (DumpTask) task;
String[] pair = GroupKey2.parseKey(dumpTask.getGroupKey());
String dataId = pair[0];
String group = pair[1];
String tenant = pair[2];
long lastModified = dumpTask.getLastModified();
String handleIp = dumpTask.getHandleIp();
boolean isBeta = dumpTask.isBeta();
String tag = dumpTask.getTag();
ConfigDumpEvent.ConfigDumpEventBuilder build = ConfigDumpEvent.builder().namespaceId(tenant).dataId(dataId)
.group(group).isBeta(isBeta).tag(tag).lastModifiedTs(lastModified).handleIp(handleIp);
/*
* 省略
*/
//根据dataId、group、tenant从存储上查询最新数据
ConfigInfo cf = persistService.findConfigInfo(dataId, group, tenant);
build.remove(Objects.isNull(cf));
//最新的Content数据
build.content(Objects.isNull(cf) ? null : cf.getContent());
build.type(Objects.isNull(cf) ? null : cf.getType());
return DumpConfigHandler.configDump(build.build());
}
}
final DumpService dumpService;
}在 DumpProcessor 的 process 方法会从存储上（mysql、derby）中查询最新的数据，然后调用 dump 方法更新内存中的md5值
public static boolean dump(String dataId, String group, String tenant, String content, long lastModifiedTs,
String type) {
String groupKey = GroupKey2.getKey(dataId, group, tenant);
//获取key对应的缓存对象
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);
//新md5和内存中的旧md5值如果一致 则不执行saveToDisk的判断
if (md5.equals(ConfigCacheService.getContentMd5(groupKey))) {
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值
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);
}
}dump 方法将更新配置的 md5 值
public static void updateMd5(String groupKey, String md5, long lastModifiedTs) {
CacheItem cache = makeSure(groupKey);
if (cache.md5 == null || !cache.md5.equals(md5)) {
cache.md5 = md5;
cache.lastModifiedTs = lastModifiedTs;
NotifyCenter.publishEvent(new LocalDataChangeEvent(groupKey));
}
}updateMd5 方法将更新的数据包装成 LocalDataChangeEvent 事件并向事件订阅者广播，LocalDataChangeEvent 事件的订阅者是 LongPollingService，因此会调用 LongPollingService.onEvent 方法
public LongPollingService() {
allSubs = new ConcurrentLinkedQueue<ClientLongPolling>();
ConfigExecutor.scheduleLongPolling(new StatTask(), 0L, 10L, TimeUnit.SECONDS);
// 注册订阅事件类型
NotifyCenter.registerToPublisher(LocalDataChangeEvent.class, NotifyCenter.ringBufferSize);
// 注册订阅事件处理类
NotifyCenter.registerSubscriber(new Subscriber() {
@Override
public void onEvent(Event event) {
if (isFixedPolling()) {
// Ignore.
} else {
if (event instanceof LocalDataChangeEvent) {
LocalDataChangeEvent evt = (LocalDataChangeEvent) event;
//将事件数据包装成DataChangeTask任务并使用线程池执行
ConfigExecutor.executeLongPolling(new DataChangeTask(evt.groupKey, evt.isBeta, evt.betaIps));
}
}
}
});
}LongPollingService 的构造方法中注册了 LocalDataChangeEvent 事件，并将事件包装成 DataChangeTask 交给 LongPolling 线程池处理
/**
* allSubs存储了与client端的所有长链接列表
*/
final Queue<ClientLongPolling> allSubs;
class DataChangeTask implements Runnable {
@Override
public void run() {
try {
ConfigCacheService.getContentBetaMd5(groupKey);
//循环所有client长连接
for (Iterator<ClientLongPolling> iter = allSubs.iterator(); iter.hasNext(); ) {
ClientLongPolling clientSub = iter.next();
//判断是监听列表中是否存在当前groupKey
if (clientSub.clientMd5Map.containsKey(groupKey)) {
if (isBeta && !CollectionUtils.contains(betaIps, clientSub.ip)) {
continue;
}
if (StringUtils.isNotBlank(tag) && !tag.equals(clientSub.tag)) {
continue;
}
getRetainIps().put(clientSub.ip, System.currentTimeMillis());
//删除监听关系
iter.remove();
LogUtil.CLIENT_LOG
.info("{}|{}|{}|{}|{}|{}|{}", (System.currentTimeMillis() - changeTime), "in-advance",
RequestUtil
.getRemoteIp((HttpServletRequest) clientSub.asyncContext.getRequest()),
"polling", clientSub.clientMd5Map.size(), clientSub.probeRequestSize, groupKey);
//向被监听的client长连接发送结果
clientSub.sendResponse(Arrays.asList(groupKey));
}
}
} catch (Throwable t) {
LogUtil.DEFAULT_LOG.error("data change error: {}", ExceptionUtil.getStackTrace(t));
}
}
}
void sendResponse(List<String> changedGroups) {
// Cancel time out task.
if (null != asyncTimeoutFuture) {
asyncTimeoutFuture.cancel(false);
}
//生成响应内容
generateResponse(changedGroups);
}
void generateResponse(List<String> changedGroups) {
//如果没有变更数据 结束Tomcat的异步请求
if (null == changedGroups) { // Tell web container to send http response.
asyncContext.complete();
return;
}
HttpServletResponse response = (HttpServletResponse) asyncContext.getResponse();
try {
final String respString = MD5Util.compareMd5ResultString(changedGroups);
// 禁用缓存
response.setHeader("Pragma", "no-cache");
response.setDateHeader("Expires", 0);
response.setHeader("Cache-Control", "no-cache,no-store");
response.setStatus(HttpServletResponse.SC_OK);
//返回变更的groupKey
response.getWriter().println(respString);
//结束Tomcat的异步请求
asyncContext.complete();
} catch (Exception ex) {
PULL_LOG.error(ex.toString(), ex);
asyncContext.complete();
}
}DataChangeTask 每次执行时会在所有的 Queue<ClientLongPolling> allSubs 长连接列表中查找有监听当前数据变更的 Client，并将变更数据推送给 Client，同时结束 Client 长连接轮询连接。
以上是整个发布配置的流程，代码比较长，需要仔细阅读，Long-Polling 长轮询也是 Nacos 比较核心的特性。
获取配置原理解析
Nacos 获取配置有2种方法，客户端可以通过 HTTP GET 获取一次配置内容，另外一种是客户端通过 HTTP GET /listener 长轮询的方式监听某个配置，当服务端配置发生变化时会将最新的配置推送给客户端。
1 短连接获取一次数据
根据 dataId、group 等参数获取 Nacos 中的内容，在 Nacos 中每个配置项都是一个 CacheItem 对象，每个 CacheItem 对象中都包含一把读写锁，当客户端来读取数据时，先根据 dataId 等参数获取 CacheItem，如果 CacheItem 不存在，则返回404，如果 CacheItem 存在会对其加read锁，如果加锁失败，则会重试，超过最大重试次数10次后仍然失败的，则返回409，如果加锁成功，则从数据库中读取数据返回给客户端，最后释放锁。
HTTP GET 获取一次配置的REST接口为 getConfig 方法
@GetMapping
@Secured(action = ActionTypes.READ, parser = ConfigResourceParser.class)
public void getConfig(HttpServletRequest request, HttpServletResponse response,
@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)
throws IOException, ServletException, NacosException {
// 检查多租户参数
ParamUtils.checkTenant(tenant);
tenant = NamespaceUtil.processNamespaceParameter(tenant);
// 验证参数
ParamUtils.checkParam(dataId, group, "datumId", "content");
ParamUtils.checkParam(tag);
final String clientIp = RequestUtil.getRemoteIp(request);
//调用doGetConfig获取配置
inner.doGetConfig(request, response, dataId, group, tenant, tag, clientIp);
}getConfig 方法是获取配置的 REST 接口，调用了 doGetConfig 方法
public String doGetConfig(HttpServletRequest request, HttpServletResponse response, String dataId, String group,
String tenant, String tag, String clientIp) throws IOException, ServletException {
//将参数拼接成组合成groupKey字符串
final String groupKey = GroupKey2.getKey(dataId, group, tenant);
String autoTag = request.getHeader("Vipserver-Tag");
String requestIpApp = RequestUtil.getAppName(request);
//对groupKey加读锁，Nacos会把所有的配置数据dump到内存中做缓存，每个缓存数据对象中都会包含一把读写锁
//lockResult = 0 内存中不存在groupKey对应的数据
//lockResult = 1 加锁成功
//lockResult = -1 加锁失败
int lockResult = tryConfigReadLock(groupKey);
final String requestIp = RequestUtil.getRemoteIp(request);
boolean isBeta = false;
//加锁成功
if (lockResult > 0) {
/**
*代码太长，省略非核心逻辑
*/
String md5 = Constants.NULL;
long lastModified = 0L;
//从缓存中获取CacheItem对象
CacheItem cacheItem = ConfigCacheService.getContentCache(groupKey);
//配置内容的md5
md5 = cacheItem.getMd5();
//配置内容最后修改时间
lastModified = cacheItem.getLastModifiedTs();
//如果是单机模式，直接从持久化的数据源读取数据，如mysql、derby，否则从文件系统读取数据
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, requestIp);
response.setStatus(HttpServletResponse.SC_NOT_FOUND);
response.getWriter().println("config data not exist");
return HttpServletResponse.SC_NOT_FOUND + "";
}
/**
*代码太长，省略非核心逻辑
*/
// 禁用缓存.
response.setHeader("Pragma", "no-cache");
response.setDateHeader("Expires", 0);
response.setHeader("Cache-Control", "no-cache,no-store");
if (PropertyUtil.isDirectRead()) {
response.setDateHeader("Last-Modified", lastModified);
} else {
fis = new FileInputStream(file);
response.setDateHeader("Last-Modified", file.lastModified());
}
//返回数据
if (PropertyUtil.isDirectRead()) {
out = response.getWriter();
out.print(configInfoBase.getContent());
out.flush();
out.close();
} else {
fis.getChannel()
.transferTo(0L, fis.getChannel().size(), Channels.newChannel(response.getOutputStream()));
}
}
//加锁数据 返回数据不存在
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,
requestIp);
response.setStatus(HttpServletResponse.SC_NOT_FOUND);
response.getWriter().println("config data not exist");
return HttpServletResponse.SC_NOT_FOUND + "";
}
//在尝试了10次加锁后失败，返回资源冲突
else {
PULL_LOG.info("[client-get] clientIp={}, {}, get data during dump", clientIp, groupKey);
response.setStatus(HttpServletResponse.SC_CONFLICT);
response.getWriter().println("requested file is being modified, please try later.");
return HttpServletResponse.SC_CONFLICT + "";
}
return HttpServletResponse.SC_OK + "";
}doGetConfig 方法对 CacheItem 加读锁成功后从持久化数据层或者文件中读取配置内容，doGetConfig 获取配置内容的逻辑比较简单
2 长连接监听数据变更
根据客户端传入的 probeModify 监听的数据列表，先判断客户端是否支持长轮询，如果客户端支持长轮询，则开启长轮询连接，如果客户端不支持，则检测被监听数据 probeModify 中的数据在服务端是否存在变更，如果有直接返回最新的数据。
服务端长轮询是使用的一个延时线程 ClientLongPolling 实现的，用以阻塞客户端的连接，并且将线程 ClientLongPolling 加入到 Queue<ClientLongPolling> allSubs 保存了起来，线程默认延时时间为客户端传入的 Long-Pulling-Timeout 减去0.5秒，因此延时时间一般是29.5秒，在29.5秒后延时线程会直接返回 NULL，由客户端发起下一次长轮询请求，直接返回 NULL 的原因是因为如果在这29.5秒中如果被监听的 probeModify 数据发生了变化，会在发布配置时创建的 DataChangeTask 线程中会从 Queue<ClientLongPolling> allSubs 延时线程列表中找到响应的 ClientLongPolling 线程，将线程取消，同时将最新的数据通过 ClientLongPolling 中保存的 AsyncContext 对象将数据推送给客户端，因为如果 ClientLongPolling 线程在29.5秒后执行了，说明在这期间没有数据变更，因此直接返回NULL。
HTTP Long-Polling 长轮询监听接口为 listener 方法
@PostMapping("/listener")
@Secured(action = ActionTypes.READ, parser = ConfigResourceParser.class)
public void listener(HttpServletRequest request, HttpServletResponse response)
throws ServletException, IOException {
//设置异步参数
request.setAttribute("org.apache.catalina.ASYNC_SUPPORTED", true);
//获取监听的dataId
String probeModify = request.getParameter("Listening-Configs");
if (StringUtils.isBlank(probeModify)) {
throw new IllegalArgumentException("invalid probeModify");
}
probeModify = URLDecoder.decode(probeModify, Constants.ENCODE);
Map<String, String> clientMd5Map;
try {
clientMd5Map = MD5Util.getClientMd5Map(probeModify);
} catch (Throwable e) {
throw new IllegalArgumentException("invalid probeModify");
}
// 开始长轮询监听
inner.doPollingConfig(request, response, clientMd5Map, probeModify.length());
}org.apache.catalina.ASYNC_SUPPORTED 是 Servlet3.0 的新特性，支持异步处理请求
public String doPollingConfig(HttpServletRequest request, HttpServletResponse response,
Map<String, String> clientMd5Map, int probeRequestSize) throws IOException {
// 如果客户端支持长轮询，将开启长轮询监听数据变更
if (LongPollingService.isSupportLongPolling(request)) {
longPollingService.addLongPollingClient(request, response, clientMd5Map, probeRequestSize);
return HttpServletResponse.SC_OK + "";
}
// 如果客户端不支持长轮询，则直接查找probeModify中修改的数据并返回结果
List<String> changedGroups = MD5Util.compareMd5(request, response, clientMd5Map);
String oldResult = MD5Util.compareMd5OldResult(changedGroups);
String newResult = MD5Util.compareMd5ResultString(changedGroups);
String version = request.getHeader(Constants.CLIENT_VERSION_HEADER);
if (version == null) {
version = "2.0.0";
}
int versionNum = Protocol.getVersionNumber(version);
// 2.0.4之前的版本将新老MD5值放入header中
if (versionNum < START_LONG_POLLING_VERSION_NUM) {
response.addHeader(Constants.PROBE_MODIFY_RESPONSE, oldResult);
response.addHeader(Constants.PROBE_MODIFY_RESPONSE_NEW, newResult);
} else {
request.setAttribute("content", newResult);
}
Loggers.AUTH.info("new content:" + newResult);
// 禁用缓存.
response.setHeader("Pragma", "no-cache");
response.setDateHeader("Expires", 0);
response.setHeader("Cache-Control", "no-cache,no-store");
response.setStatus(HttpServletResponse.SC_OK);
return HttpServletResponse.SC_OK + "";
}doPollingConfig 先判断客户端是否支持长轮询如果支持则使用长轮询监听，否则直接返回有变更数据的 md5 值
public void addLongPollingClient(HttpServletRequest req, HttpServletResponse rsp, Map<String, String> clientMd5Map,
int probeRequestSize) {
String str = req.getHeader(LongPollingService.LONG_POLLING_HEADER);
String noHangUpFlag = req.getHeader(LongPollingService.LONG_POLLING_NO_HANG_UP_HEADER);
String appName = req.getHeader(RequestUtil.CLIENT_APPNAME_HEADER);
String tag = req.getHeader("Vipserver-Tag");
int delayTime = SwitchService.getSwitchInteger(SwitchService.FIXED_DELAY_TIME, 500);
// 获取长轮询超时时间，为避免客户端超时这里的超时时间减去了500毫秒
//timeout 一般为 30 - 0.5 = 29.5秒
long timeout = Math.max(10000, Long.parseLong(str) - delayTime);
if (isFixedPolling()) {
timeout = Math.max(10000, getFixedPollingInterval());
} else {
long start = System.currentTimeMillis();
//查找有变化的数据
List<String> changedGroups = MD5Util.compareMd5(req, rsp, clientMd5Map);
//数据发生了变化
if (changedGroups.size() > 0) {
//直接返回变化的数据
generateResponse(req, rsp, changedGroups);
LogUtil.CLIENT_LOG.info("{}|{}|{}|{}|{}|{}|{}", System.currentTimeMillis() - start, "instant",
RequestUtil.getRemoteIp(req), "polling", clientMd5Map.size(), probeRequestSize,
changedGroups.size());
return;
}
//如果监听的数据没有变化并且header中有Long-Pulling-Timeout-No-Hangup标示则直接结束
else if (noHangUpFlag != null && noHangUpFlag.equalsIgnoreCase(TRUE_STR)) {
LogUtil.CLIENT_LOG.info("{}|{}|{}|{}|{}|{}|{}", System.currentTimeMillis() - start, "nohangup",
RequestUtil.getRemoteIp(req), "polling", clientMd5Map.size(), probeRequestSize,
changedGroups.size());
return;
}
}
/**
* 注意！！！
* 当数据没有变化时，代码会执行到这里，开始使用长轮询阻塞请求
*/
String ip = RequestUtil.getRemoteIp(req);
// 开启异步支持
final AsyncContext asyncContext = req.startAsync();
// Servlet的异步超时时间不正确，超时时间由自己来控制
asyncContext.setTimeout(0L);
// 创建ClientLongPolling线程并交给ConfigExecutor执行
ConfigExecutor.executeLongPolling(
new ClientLongPolling(asyncContext, clientMd5Map, ip, probeRequestSize, timeout, appName, tag));
}
//通过Response返回变化的数据
void generateResponse(HttpServletRequest request, HttpServletResponse response, List<String> changedGroups) {
if (null == changedGroups) {
return;
}
try {
final String respString = MD5Util.compareMd5ResultString(changedGroups);
// 禁用缓存.
response.setHeader("Pragma", "no-cache");
response.setDateHeader("Expires", 0);
response.setHeader("Cache-Control", "no-cache,no-store");
response.setStatus(HttpServletResponse.SC_OK);
response.getWriter().println(respString);
} catch (Exception ex) {
PULL_LOG.error(ex.toString(), ex);
}
}addLongPollingClient 首先会检测一次数据是否有变化，如果有则通过 generateResponse 方法直接返回响应结果，否则创建 ClientLongPolling 线程开启长轮询连接，长轮询连接使用一个延时线程实现，延时时间从客户端的 header 中获取，默认为30s，实际上是29.5秒防止客户端超时。
/**
* allSubs存储了与client端的所有长链接列表
*/
final Queue<ClientLongPolling> allSubs;
class ClientLongPolling implements Runnable {
@Override
public void run() {
//run 方法创建了一个延时线程，延时时间为长轮询的超时时间 30 - 0.5 = 29.5秒
//也就是说在29.5秒之后会执行下面的run方法
asyncTimeoutFuture = ConfigExecutor.scheduleLongPolling(new Runnable() {
@Override
public void run() {
try {
/**
* 29.5秒之后开始执行
*/
getRetainIps().put(ClientLongPolling.this.ip, System.currentTimeMillis());
// 删除订阅关系
allSubs.remove(ClientLongPolling.this);
//不会走这个分支
if (isFixedPolling()) {
LogUtil.CLIENT_LOG
.info("{}|{}|{}|{}|{}|{}", (System.currentTimeMillis() - createTime), "fix",
RequestUtil.getRemoteIp((HttpServletRequest) asyncContext.getRequest()),
"polling", clientMd5Map.size(), probeRequestSize);
List<String> changedGroups = MD5Util
.compareMd5((HttpServletRequest) asyncContext.getRequest(),
(HttpServletResponse) asyncContext.getResponse(), clientMd5Map);
if (changedGroups.size() > 0) {
sendResponse(changedGroups);
} else {
sendResponse(null);
}
}
//因为过了29.5秒到了长轮询的超时时间，说明在29.5秒内没有数据发生过变化
//因此发送空数据给client，由client开启下一次长轮询
else {
LogUtil.CLIENT_LOG
.info("{}|{}|{}|{}|{}|{}", (System.currentTimeMillis() - createTime), "timeout",
RequestUtil.getRemoteIp((HttpServletRequest) asyncContext.getRequest()),
"polling", clientMd5Map.size(), probeRequestSize);
//发送空的数据给client
sendResponse(null);
}
} catch (Throwable t) {
LogUtil.DEFAULT_LOG.error("long polling error:" + t.getMessage(), t.getCause());
}
}
}, timeoutTime, TimeUnit.MILLISECONDS);
//注意！！！
//在创建完延时线程后，就将当前对象加入allSubs队列中了，allSubs存储了与client端的所有长链接列表
allSubs.add(this);
}
}ClientLongPolling 会创建一个29.5秒的延时线程，并将当前长轮询对象加入到 allSubs 队列中，在29.5之内如果监听的数据发生了变化会由发布配置的 DataChangeTask 线程将变更数据发送给 Client 同时取消 asyncTimeoutFuture 这个延时线程，如果在29.5内监听的数据没有发送变化则发送空数据给 Client，由 Client 开启下一次长轮询。
总结
- Nacos 客户端SDK在获取配置时会优先从本地文件中读取配置，也就是说如果不想从 Nacos 服务端中获取数据，可以在本地新建文件，这样就可以单独更改集群中某个机器的配置
- Nacos 客户端SDK在从 Nacos 服务器获取配置失败时，会从快照数据中读取配置，而快照数据是存储的上一次从服务器拉取的数据，当 Nacos 服务器挂掉后，从本地快照依然可以获取数据
- Nacos 服务端支持短连接获取配置和长轮询监听配置方式，长轮询监听是基于 Servlet3.0 的异步特性实现的，由客户端发起长轮询请求，服务端使用延时线程阻塞请求，阻塞超时时间为30秒，在30秒内监听数据发生变化服务端将最新数据推送给客户端，在30秒外服务端返回NULL给客户端，由客户端发起下一次长轮询请求，继续监听。