1. 什么是Nacos ?
Nacos主要用做注冊中心和配置中心。Nacos介紹，Nacos用法， Nacos源碼下載 etc.. 請查看Nacos官方文檔， 本文基於nacos版本1.2.0進行分析。
2. Nacos代碼入口
從官方文檔給的JAVA SDK 入手， 這樣可以知道使用流程，也可以通過入口，分析代碼。官方給的代碼如下：
try {
String serverAddr = "{serverAddr}";
String dataId = "{dataId}";
String group = "{group}";
Properties properties = new Properties();
properties.put("serverAddr", serverAddr);
ConfigService configService = NacosFactory.createConfigService(properties);
String content = configService.getConfig(dataId, group, 5000);
System.out.println(content);
} catch (NacosException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
代碼先ConfigService，然后通過configService進行數據庫操作，我們也從configservice作為入口，進行代碼分析。
tips: 找入口最好從使用的demo開始，這樣可以快速找到入口，分析代碼。
3. 代碼分析
ConfigService的實現類只有下面一個，我們就從該類的構造方法開始.
public NacosConfigService(Properties properties) throws NacosException {
String encodeTmp = properties.getProperty(PropertyKeyConst.ENCODE);
if (StringUtils.isBlank(encodeTmp)) {
encode = Constants.ENCODE;
} else {
encode = encodeTmp.trim();
}
initNamespace(properties);
// 用戶登錄信息
agent = new MetricsHttpAgent(new ServerHttpAgent(properties));
// 維護nacos服務列表
agent.start();
// 更新維護配置
worker = new ClientWorker(agent, configFilterChainManager, properties);
}
代碼中主要代碼有三行，HttpAgent實現類有MetricsHttpAgent和ServerHttpAgent。我們一步步點進去看。
點進MetricsHttpAgent，核心調用邏輯是通過構造參數傳入的HttpAgent， 該類是對傳入的HttpAgent調用的方法加了增加了時間檢測。（裝飾器模式， 沒有通過名稱寫出來）
在進入ServerHttpAgent，代碼中實現了定時任務調度，登錄Nacos(客戶端獲取配置、服務注冊列表需要建立連接)，時間是5秒一次。
public ServerHttpAgent(Properties properties) throws NacosException {
...
ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
...
}
});
executorService.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
securityProxy.login(serverListMgr.getServerUrls());
}
}, 0, securityInfoRefreshIntervalMills, TimeUnit.MILLISECONDS);
}
我們進入start方法， 通過代碼，我們知道實際調用的是ServerHttpAgent中的start方法， 該類通過定時任務，維護nacos的列表。進入此方法，看看具體實現邏輯。
public synchronized void start() throws NacosException {
if (isStarted || isFixed) {
return;
}
// runnable接口，維護serverUrlList
GetServerListTask getServersTask = new GetServerListTask(addressServerUrl);
for (int i = 0; i < initServerlistRetryTimes && serverUrls.isEmpty(); ++i) {
// 判斷服務列表是否發生改變，如果發生改變，則更新服務列表
getServersTask.run();
try {
this.wait((i + 1) * 100L);
} catch (Exception e) {
LOGGER.warn("get serverlist fail,url: {}", addressServerUrl);
}
}
if (serverUrls.isEmpty()) {
LOGGER.error("[init-serverlist] fail to get NACOS-server serverlist! env: {}, url: {}", name,
addressServerUrl);
throw new NacosException(NacosException.SERVER_ERROR,
"fail to get NACOS-server serverlist! env:" + name + ", not connnect url:" + addressServerUrl);
}
// 將自己在丟到定時任務里面執行，執行時間為30秒一次
TimerService.scheduleWithFixedDelay(getServersTask, 0L, 30L, TimeUnit.SECONDS);
isStarted = true;
}
接下來進入最核心的代碼，如何對文件配置進行同步，緩存， 本地和遠程差異化比較的。
public ClientWorker(final HttpAgent agent, final ConfigFilterChainManager configFilterChainManager, final Properties properties) {
this.agent = agent;
this.configFilterChainManager = configFilterChainManager;
// Initialize the timeout parameter
init(properties);
executor = Executors.newScheduledThreadPool(1, new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r);
t.setName("com.alibaba.nacos.client.Worker." + agent.getName());
t.setDaemon(true);
return t;
}
});
// 通過thread名稱我們知道，是進行長輪詢的
executorService = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors(), new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread t = new Thread(r);
t.setName("com.alibaba.nacos.client.Worker.longPolling." + agent.getName());
t.setDaemon(true);
return t;
}
});
executor.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
try {
// 檢查配置信息
checkConfigInfo();
} catch (Throwable e) {
LOGGER.error("[" + agent.getName() + "] [sub-check] rotate check error", e);
}
}
}, 1L, 10L, TimeUnit.MILLISECONDS);
}
``
初始化了兩個線程池，通過名稱看出，executorService是用來進行長輪詢的線程池，具體干嘛，我們還不清楚，等等再看。 另外一個線程池，executor是用來檢查配置信息的。我們進入此方法。
```java
public void checkConfigInfo() {
// 分任務
int listenerSize = cacheMap.get().size();
// 向上取整為批數
int longingTaskCount = (int) Math.ceil(listenerSize / ParamUtil.getPerTaskConfigSize());
if (longingTaskCount > currentLongingTaskCount) {
for (int i = (int) currentLongingTaskCount; i < longingTaskCount; i++) {
// 要判斷任務是否在執行 這塊需要好好想想。 任務列表現在是無序的。變化過程可能有問題
// executorService在這里使用到， 用來執行配置檢測的長輪詢使用
executorService.execute(new LongPollingRunnable(i));
}
currentLongingTaskCount = longingTaskCount;
}
}
防止一次需要更新的太多，時間慢，所以將配置檢測的任務分批次執行，分成多個LongPollingRunnable執行，核心代碼在LongPollingRunnable中，我們進入代碼中查看。
class LongPollingRunnable implements Runnable {
private int taskId;
public LongPollingRunnable(int taskId) {
this.taskId = taskId;
}
@Override
public void run() {
List<CacheData> cacheDatas = new ArrayList<CacheData>();
List<String> inInitializingCacheList = new ArrayList<String>();
try {
// check failover config
for (CacheData cacheData : cacheMap.get().values()) {
//根據taskId對cacheMap中的數據進行操作。（查看setTaskId方法， 源碼中關聯相關方法沒有被調用，即這個判斷永遠不成立）
if (cacheData.getTaskId() == taskId) {
cacheDatas.add(cacheData);
try {
// 檢測本地配置是否發生變化，標記該配置是否可以用本地緩存的
checkLocalConfig(cacheData);
if (cacheData.isUseLocalConfigInfo()) {
cacheData.checkListenerMd5();
}
} catch (Exception e) {
LOGGER.error("get local config info error", e);
}
}
}
// check server config
// 把可能改變的配置，發送到服務器端進行比較，返回修改配置的dataid, groupId
List<String> changedGroupKeys = checkUpdateDataIds(cacheDatas, inInitializingCacheList);
LOGGER.info("get changedGroupKeys:" + changedGroupKeys);
for (String groupKey : changedGroupKeys) {
String[] key = GroupKey.parseKey(groupKey);
String dataId = key[0];
String group = key[1];
String tenant = null;
if (key.length == 3) {
tenant = key[2];
}
try {
// 從服務端獲取更新的key的值， 並進行緩存，生成文件快照
String[] ct = getServerConfig(dataId, group, tenant, 3000L);
CacheData cache = cacheMap.get().get(GroupKey.getKeyTenant(dataId, group, tenant));
cache.setContent(ct[0]);
if (null != ct[1]) {
cache.setType(ct[1]);
}
LOGGER.info("[{}] [data-received] dataId={}, group={}, tenant={}, md5={}, content={}, type={}",
agent.getName(), dataId, group, tenant, cache.getMd5(),
ContentUtils.truncateContent(ct[0]), ct[1]);
} catch (NacosException ioe) {
String message = String.format(
"[%s] [get-update] get changed config exception. dataId=%s, group=%s, tenant=%s",
agent.getName(), dataId, group, tenant);
LOGGER.error(message, ioe);
}
}
// 檢查新的值，並設置是否首次更新和初始化狀態 （ cacheData 首次出現在cacheMap中&首次check更新）
for (CacheData cacheData : cacheDatas) {
if (!cacheData.isInitializing() || inInitializingCacheList
.contains(GroupKey.getKeyTenant(cacheData.dataId, cacheData.group, cacheData.tenant))) {
cacheData.checkListenerMd5();
cacheData.setInitializing(false);
}
}
inInitializingCacheList.clear();
// 再次執行此任務
executorService.execute(this);
} catch (Throwable e) {
// If the rotation training task is abnormal, the next execution time of the task will be punished
LOGGER.error("longPolling error : ", e);
executorService.schedule(this, taskPenaltyTime, TimeUnit.MILLISECONDS);
}
}
}
此方法執行了緩存判斷，更新哪些緩存，是否需要更新， 生成緩存文件，更新緩存等，我們在進去一些關鍵方法中查看一下。
方法一 checkUpdateDataIds下的checkUpdateConfigStr
List<String> checkUpdateConfigStr(String probeUpdateString, boolean isInitializingCacheList) throws IOException {
List<String> params = new ArrayList<String>(2);
params.add(Constants.PROBE_MODIFY_REQUEST);
params.add(probeUpdateString);
List<String> headers = new ArrayList<String>(2);
headers.add("Long-Pulling-Timeout");
headers.add("" + timeout); // 默認時間是30000ms = 30s
// told server do not hang me up if new initializing cacheData added in
// 如果是初始化過的配置，沒有發生變化的，則會進行等待。 長輪詢，監聽配置的改變。
if (isInitializingCacheList) {
headers.add("Long-Pulling-Timeout-No-Hangup");
headers.add("true");
}
if (StringUtils.isBlank(probeUpdateString)) {
return Collections.emptyList();
}
try {
// In order to prevent the server from handling the delay of the client's long task,
// increase the client's read timeout to avoid this problem.
long readTimeoutMs = timeout + (long) Math.round(timeout >> 1);
// 發送請求 ， 這段邏輯需要去服務端看。 服務端大體邏輯就是 解析傳入的groupdataId 字符串，通過MD5比較, 找出被修改的配置。 如果沒有，則會懸掛起來。
// 服務器在響應的時候會提前500ms返回，防止請求超時。
HttpResult result = agent.httpPost(Constants.CONFIG_CONTROLLER_PATH + "/listener", headers, params,
agent.getEncode(), readTimeoutMs);
if (HttpURLConnection.HTTP_OK == result.code) {
setHealthServer(true);
// 解析服務段返回的修改的配置的數據信息
return parseUpdateDataIdResponse(result.content);
} else {
setHealthServer(false);
LOGGER.error("[{}] [check-update] get changed dataId error, code: {}", agent.getName(), result.code);
}
} catch (IOException e) {
setHealthServer(false);
LOGGER.error("[" + agent.getName() + "] [check-update] get changed dataId exception", e);
throw e;
}
return Collections.emptyList();
}
方法二 getServerConfig方法, 拿到更改的goupid, dataid ，從服務器端獲取數據，更新快照文件
public String[] getServerConfig(String dataId, String group, String tenant, long readTimeout)
throws NacosException {
String[] ct = new String[2];
if (StringUtils.isBlank(group)) {
group = Constants.DEFAULT_GROUP;
}
HttpResult result = null;
try {
List<String> params = null;
if (StringUtils.isBlank(tenant)) {
params = new ArrayList<String>(Arrays.asList("dataId", dataId, "group", group));
} else {
params = new ArrayList<String>(Arrays.asList("dataId", dataId, "group", group, "tenant", tenant));
}
// 發送請求，獲取數據
result = agent.httpGet(Constants.CONFIG_CONTROLLER_PATH, null, params, agent.getEncode(), readTimeout);
} catch (IOException e) {
String message = String.format(
"[%s] [sub-server] get server config exception, dataId=%s, group=%s, tenant=%s", agent.getName(),
dataId, group, tenant);
LOGGER.error(message, e);
throw new NacosException(NacosException.SERVER_ERROR, e);
}
switch (result.code) {
case HttpURLConnection.HTTP_OK:
// 更新快照文件
LocalConfigInfoProcessor.saveSnapshot(agent.getName(), dataId, group, tenant, result.content);
ct[0] = result.content;
if (result.headers.containsKey(CONFIG_TYPE)) {
ct[1] = result.headers.get(CONFIG_TYPE).get(0);
} else {
ct[1] = ConfigType.TEXT.getType();
}
return ct;
case HttpURLConnection.HTTP_NOT_FOUND:
LocalConfigInfoProcessor.saveSnapshot(agent.getName(), dataId, group, tenant, null);
return ct;
case HttpURLConnection.HTTP_CONFLICT: {
LOGGER.error(
"[{}] [sub-server-error] get server config being modified concurrently, dataId={}, group={}, "
+ "tenant={}", agent.getName(), dataId, group, tenant);
throw new NacosException(NacosException.CONFLICT,
"data being modified, dataId=" + dataId + ",group=" + group + ",tenant=" + tenant);
}
case HttpURLConnection.HTTP_FORBIDDEN: {
LOGGER.error("[{}] [sub-server-error] no right, dataId={}, group={}, tenant={}", agent.getName(), dataId,
group, tenant);
throw new NacosException(result.code, result.content);
}
default: {
LOGGER.error("[{}] [sub-server-error] dataId={}, group={}, tenant={}, code={}", agent.getName(), dataId,
group, tenant, result.code);
throw new NacosException(result.code,
"http error, code=" + result.code + ",dataId=" + dataId + ",group=" + group + ",tenant=" + tenant);
}
}
}
至此，我們分析了代碼的大部分邏輯，通過線程任務，記錄數據狀態，來比較服務器端和本地的配置差一點，從而實現數據修改變化的判斷，我們畫一下大概的架構圖。
4. 流程圖
5. 配置中心客戶端大致思想
- a. 每次網絡請求較慢，本地緩存
- b. 異步比較差異，采用主動輪詢 + 被動通知（長輪詢中）的方式
- c. 防止數據量過大，分批處理
- d. 數據變化監聽