nacos集群内置数据源,nacos配置中心原理
目录
Nacos集群工作原理配置更改同步条目AsyncNotifyServiceAsyncTask目标节点接收请求Nacos延迟TaskexecuteenProcess RunnableProcessTasks dump producer。进程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的数据转储到本地文件中。此外,在启动NacosServer之后,将同步启动一个计划任务,每隔6个小时,将会有一个完整的数据转储到本地文件。
00-1010修改、删除或添加配置时,发布notifyConfigChange事件。
@ post mapping @ Secured(action=action types。WRITE,parser=configresourceparser . class)publish config(http servlet request请求,HttpServletResponse响应,@RequestParam(value=dataId )字符串dataId,@RequestParam(value=group )字符串组,@RequestParam(value=tenant ,required=false,defaultValue=StringUtils。空)字符串租户、@RequestParam(value=content )字符串内容、@RequestParam(value=tag ,required=false)字符串标记、@RequestParam(value=appName ,required=false)字符串appName、@RequestParam(value=src_user ,required=false)字符串srcUser、@ request param(value= config _ tags ,required=false)字符串configTags、@RequestParam(value=desc ,required=false)字符串desc、@RequestParam(value=use ,required=false.if(stringitils . is blank(beta IPS)){ if(stringitils . is blank(tag)){ persistservice . insert update(src IP,src user,config info,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,它会处理数据变更后的同步事件。
@Autowiredpublic 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
@Overridepublic 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, 先调用父类去完成任务添加。
@Overridepublic 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
读取数据库的最新数据,然后更新本地缓存和磁盘
以上就是Nacos配置中心集群原理及源码分析的详细内容,更多关于Nacos配置中心集群原理的资料请关注盛行IT其它相关文章!
郑重声明:本文由网友发布,不代表盛行IT的观点,版权归原作者所有,仅为传播更多信息之目的,如有侵权请联系,我们将第一时间修改或删除,多谢。