江苏龙网

  1. 主页 > 生活百科 > >

Nacos配置中心集群原理及源码分析( 二 )

Nacos

目标节点接收请求数据同步的请求地址为，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 = https://www.isolves.com/it/wl/js/2022-03-30/"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());}在这种场景设计中，一般都会采用生产者消费者模式来完成，因此这里不难猜测到，任务会被保存到一个队列中，然后有另外一个线程来执行 。
NacosDelayTaskExecuteEngineTaskManager的父类是
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);}ProcessRunnableprivate class ProcessRunnable implements Runnable {@Overridepublic void run() {try {processTasks();} catch (Throwable e) {getEngineLog().error(e.toString(), e);}}}processTasksprotected void processTasks() {//获取所有的任务Collection<Object> keys = getAllTaskKeys();for (Object taskKey : keys) {AbstractDelayTask task = removeTask(taskKey);if (null == task) {continue;}//获取任务处理器，这里返回的是DumpProcessorNacosTaskProcessor 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读取数据库的最新数据，然后更新本地缓存和磁盘 。

原文
https://www.cnblogs.com/mic112/p/16071963.html




推荐阅读


上一篇：处暑吃饺子吗 中元节吃饺子吗

下一篇：做月饼没有线水用碱水可以吗 做月饼用的碱水怎么配