Nacos作為配置中心,必然需要保證服務節點的高可用性,那麼Nacos是如何實現集群的呢? 下麵這個圖,表示Nacos集群的部署圖。 Nacos集群工作原理 Nacos作為配置中心的集群結構中,是一種無中心化節點的設計,由於沒有主從節點,也沒有選舉機制,所以為了能夠實現熱備,就需要增加虛擬IP(VI ...
Nacos作為配置中心,必然需要保證服務節點的高可用性,那麼Nacos是如何實現集群的呢?
下麵這個圖,表示Nacos集群的部署圖。
Nacos集群工作原理
Nacos作為配置中心的集群結構中,是一種無中心化節點的設計,由於沒有主從節點,也沒有選舉機制,所以為了能夠實現熱備,就需要增加虛擬IP(VIP)。
Nacos的數據存儲分為兩部分
- Mysql資料庫存儲,所有Nacos節點共用同一份數據,數據的副本機制由Mysql本身的主從方案來解決,從而保證數據的可靠性。
- 每個節點的本地磁碟,會保存一份全量數據,具體路徑:
/data/program/nacos-1/data/config-data/${GROUP}
.
在Nacos的設計中,Mysql是一個中心數據倉庫,且認為在Mysql中的數據是絕對正確的。 除此之外,Nacos在啟動時會把Mysql中的數據寫一份到本地磁碟。
這麼設計的好處是可以提高性能,當客戶端需要請求某個配置項時,服務端會想Ian從磁碟中讀取對應文件返回,而磁碟的讀取效率要比資料庫效率高。
當配置發生變更時:
- Nacos會把變更的配置保存到資料庫,然後再寫入本地文件。
- 接著發送一個HTTP請求,給到集群中的其他節點,其他節點收到事件後,從Mysql中dump剛剛寫入的數據到本地文件中。
另外,NacosServer啟動後,會同步啟動一個定時任務,每隔6小時,會dump一次全量數據到本地文件
配置變更同步入口
當配置發生修改、刪除、新增操作時,通過發佈一個notifyConfigChange
事件。
@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 {
//省略..
if (StringUtils.isBlank(betaIps)) {
if (StringUtils.isBlank(tag)) {
persistService.insertOrUpdate(srcIp, srcUser, configInfo, time, configAdvanceInfo, true);
ConfigChangePublisher
.notifyConfigChange(new ConfigDataChangeEvent(false, dataId, group, tenant, time.getTime()));
} else {
persistService.insertOrUpdateTag(configInfo, tag, srcIp, srcUser, time, true);
ConfigChangePublisher.notifyConfigChange(
new ConfigDataChangeEvent(false, dataId, group, tenant, tag, time.getTime()));
}
}//省略
return true;
}
AsyncNotifyService
配置數據變更事件,專門有一個監聽器AsyncNotifyService,它會處理數據變更後的同步事件。
@Autowired
public AsyncNotifyService(ServerMemberManager memberManager) {
this.memberManager = memberManager;
// Register ConfigDataChangeEvent to NotifyCenter.
NotifyCenter.registerToPublisher(ConfigDataChangeEvent.class, NotifyCenter.ringBufferSize);
// Register A Subscriber to subscribe ConfigDataChangeEvent.
NotifyCenter.registerSubscriber(new Subscriber() {
@Override
public void onEvent(Event event) {
// Generate ConfigDataChangeEvent concurrently
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;
Collection<Member> ipList = memberManager.allMembers(); //得到集群中的ip列表
// 構建NotifySingleTask,並添加到隊列中。
Queue<NotifySingleTask> queue = new LinkedList<NotifySingleTask>();
for (Member member : ipList) { //遍歷集群中的每個節點
queue.add(new NotifySingleTask(dataId, group, tenant, tag, dumpTs, member.getAddress(),
evt.isBeta));
}
//非同步執行任務 AsyncTask
ConfigExecutor.executeAsyncNotify(new AsyncTask(nacosAsyncRestTemplate, queue));
}
}
@Override
public Class<? extends Event> subscribeType() {
return ConfigDataChangeEvent.class;
}
});
}
AsyncTask
@Override
public void run() {
executeAsyncInvoke();
}
private void executeAsyncInvoke() {
while (!queue.isEmpty()) {//遍歷隊列中的數據,直到數據為空
NotifySingleTask task = queue.poll(); //獲取task
String targetIp = task.getTargetIP(); //獲取目標ip
if (memberManager.hasMember(targetIp)) { //如果集群中的ip列表包含目標ip
// start the health check and there are ips that are not monitored, put them directly in the notification queue, otherwise notify
//判斷目標ip的健康狀態
boolean unHealthNeedDelay = memberManager.isUnHealth(targetIp); //
if (unHealthNeedDelay) { //如果目標服務是非健康,則繼續添加到隊列中,延後再執行。
// target ip is unhealthy, then put it in the notification list
ConfigTraceService.logNotifyEvent(task.getDataId(), task.getGroup(), task.getTenant(), null,
task.getLastModified(), InetUtils.getSelfIP(), ConfigTraceService.NOTIFY_EVENT_UNHEALTH,
0, task.target);
// get delay time and set fail count to the task
asyncTaskExecute(task);
} else {
//構建header
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");
}
AuthHeaderUtil.addIdentityToHeader(header);
//通過restTemplate發起遠程調用,如果調用成功,則執行AsyncNotifyCallBack的回調方法
restTemplate.get(task.url, header, Query.EMPTY, String.class, new AsyncNotifyCallBack(task));
}
}
}
}
目標節點接收請求
數據同步的請求地址為,task.url=http://192.168.8.16:8848/nacos/v1/cs/communication/dataChange?dataId=log.yaml&group=DEFAULT_GROUP
@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");
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;
}
dumpService.dump用來實現配置的更新,代碼如下
當前任務會被添加到DumpTaskMgr中管理。
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);
dumpTaskMgr.addTask(taskKey, new DumpTask(groupKey, tag, lastModified, handleIp, isBeta));
DUMP_LOG.info("[dump-task] add task. groupKey={}, taskKey={}", groupKey, taskKey);
}
TaskManager.addTask, 先調用父類去完成任務添加。
@Override
public void addTask(Object key, AbstractDelayTask newTask) {
super.addTask(key, newTask);
MetricsMonitor.getDumpTaskMonitor().set(tasks.size());
}
在這種場景設計中,一般都會採用生產者消費者模式來完成,因此這裡不難猜測到,任務會被保存到一個隊列中,然後有另外一個線程來執行。
NacosDelayTaskExecuteEngine
TaskManager的父類是NacosDelayTaskExecuteEngine,
這個類中有一個成員屬性protected final ConcurrentHashMap<Object, AbstractDelayTask> tasks;
,專門來保存延期執行的任務類型AbstractDelayTask.
在這個類的構造方法中,初始化了一個延期執行的任務,其中具體的任務是ProcessRunnable.
public NacosDelayTaskExecuteEngine(String name, int initCapacity, Logger logger, long processInterval) {
super(logger);
tasks = new ConcurrentHashMap<Object, AbstractDelayTask>(initCapacity);
processingExecutor = ExecutorFactory.newSingleScheduledExecutorService(new NameThreadFactory(name));
processingExecutor
.scheduleWithFixedDelay(new ProcessRunnable(), processInterval, processInterval, TimeUnit.MILLISECONDS);
}
ProcessRunnable
private class ProcessRunnable implements Runnable {
@Override
public void run() {
try {
processTasks();
} catch (Throwable e) {
getEngineLog().error(e.toString(), e);
}
}
}
processTasks
protected void processTasks() {
//獲取所有的任務
Collection<Object> keys = getAllTaskKeys();
for (Object taskKey : keys) {
AbstractDelayTask task = removeTask(taskKey);
if (null == task) {
continue;
}
//獲取任務處理器,這裡返回的是DumpProcessor
NacosTaskProcessor processor = getProcessor(taskKey);
if (null == processor) {
getEngineLog().error("processor not found for task, so discarded. " + task);
continue;
}
try {
// ReAdd task if process failed
//執行具體任務
if (!processor.process(task)) {
retryFailedTask(taskKey, task);
}
} catch (Throwable e) {
getEngineLog().error("Nacos task execute error : " + e.toString(), e);
retryFailedTask(taskKey, task);
}
}
}
DumpProcessor.process
讀取資料庫的最新數據,然後更新本地緩存和磁碟。
版權聲明:本博客所有文章除特別聲明外,均採用 CC BY-NC-SA 4.0 許可協議。轉載請註明來自
Mic帶你學架構
!
如果本篇文章對您有幫助,還請幫忙點個關註和贊,您的堅持是我不斷創作的動力。歡迎關註「跟著Mic學架構」公眾號公眾號獲取更多技術乾貨!