1任務(wù)管理類

因為Nacos中有很多地方使用了這個TaskManager,所以我們得先了解一下這個類是干啥用的,方便后面閱讀源碼時候不會吃力;

先說結(jié)論: TaskManager 可以看成是一個待執(zhí)行的任務(wù)集合,用于處理一定要執(zhí)行成功的任務(wù) 單線程的方式處理任務(wù)，保證任務(wù)一定被成功處理; 如果執(zhí)行失敗了,任務(wù)會被重新放入集合中等待下一次被消費;

AbstractTask

AbstractTask是個抽象類,所有的需要被執(zhí)行的任務(wù)都繼續(xù)這個類; 這個類主要提供執(zhí)行任務(wù)所需要的數(shù)據(jù)和方法;例如

/* 一個任務(wù)兩次處理的間隔，單位是毫秒*/ private long taskInterval; /*任務(wù)上次被處理的時間，用毫秒表示*/ private long lastProcessTime; /* TaskManager 判斷當前是否需要處理這個Task，子類可以O(shè)verride這個函數(shù)實現(xiàn)自己的邏輯 */ public boolean shouldProcess() { return (System.currentTimeMillis() - this.lastProcessTime >= this.taskInterval); }

TaskProcessor任務(wù)處理器

TaskProcessor 是任務(wù)處理器接口,它有個方法

boolean process(String taskType, AbstractTask task);

用于執(zhí)行對應(yīng)的AbstractTask任務(wù)類; 不同的任務(wù)類型,可以實現(xiàn)自己的執(zhí)行任務(wù)邏輯;

TaskManager任務(wù)管理類

TaskManager 是個任務(wù)管理類; 它里面有兩個屬性保存了待消費的任務(wù)AbstractTask,和任務(wù)執(zhí)行需要的TaskProcessor;

/**待消費的任務(wù)AbstractTask**/ private final ConcurrentHashMap<String, AbstractTask> tasks = new ConcurrentHashMap<String, AbstractTask>(); /**任務(wù)AbstractTask對應(yīng)的任務(wù)執(zhí)行器TaskProcessor**/ private final ConcurrentHashMap<String, TaskProcessor> taskProcessors =new ConcurrentHashMap<String, TaskProcessor>();

如果taskProcessors中沒有找到對應(yīng)的任務(wù)執(zhí)行器,那么它里面有一個默認執(zhí)行器會執(zhí)行

/**默認執(zhí)行器**/ private TaskProcessor defaultTaskProcessor;

2使用用例

Nacos配置中心模塊很重要一個功能就是,在初始化的時候以及每隔一段時間就會去數(shù)據(jù)庫中把所有數(shù)據(jù)Dump到磁盤中;Dump就是一個任務(wù)類AbstractTask; 我們上面說過 AbstractTask就是一個信息承載對象,主要給TaskProcessor提供執(zhí)行所需要的數(shù)據(jù);我們看看DumpTask;

DumpTask

DumpTask定義了自己的一些屬性; 再看看其他的例如DumpAllTask、DumpAllBetaTask

這兩個任務(wù)類只定義了TASK_ID 既然有DumpTask任務(wù)類,那肯定就有對應(yīng)的任務(wù)處理器類DumpProcessor;

DumpProcessor

DumpProcessor 是DumpTask任務(wù)的執(zhí)行器;執(zhí)行器中的方法

public boolean process(String taskType, AbstractTask task)

代碼太長就不在這里分析了,它里面主要做的操作就是 保存配置文件到本地磁盤中，并緩存md5 詳細可以看文章 【Nacos源碼之配置管理 四】DumpService如何將配置文件全部Dump到磁盤中

對應(yīng)DumpAllTask、DumpAllBetaTask 任務(wù)的任務(wù)執(zhí)行器有DumpAllProcessor、DumpAllBetaProcessor

3DumpAllTask任務(wù)觸發(fā)執(zhí)行的地方

上面是DumpAllTask的定義和DumpAllTaskProcessor執(zhí)行器的定義;定義好了之后是怎么被觸發(fā)的呢？

DumpService初始化Dump配置信息

這個類就是專門Dump配置信息的服務(wù)類;上面提及的DumpAll就是在這里被調(diào)用的;我們來看下他主要方法;

@PostConstruct public void init() { DumpAllProcessor dumpAllProcessor = new DumpAllProcessor(this); /**在new這個TaskManager類的時候,專門執(zhí)行任務(wù)的一個線程就已經(jīng)開始啟動了,這不過這個時候還沒有任務(wù)Task添加進去**/ dumpAllTaskMgr = new TaskManager( "com.alibaba.nacos.server.DumpAllTaskManager"); dumpAllTaskMgr.setDefaultTaskProcessor(dumpAllProcessor); Runnable dumpAll = new Runnable() { @Override public void run() { dumpAllTaskMgr.addTask(DumpAllTask.TASK_ID, new DumpAllTask()); } }; /**每10分鐘執(zhí)行一次DumpAll操作**/ TimerTaskService.scheduleWithFixedDelay(dumpAll, initialDelay, DUMP_ALL_INTERVAL_IN_MINUTE, TimeUnit.MINUTES); }

DumpService在初始化的時候回調(diào)用這個init方法; 1.先new了一個DumpAllProcessor執(zhí)行器; 2.再new 了一個TaskManager任務(wù)管理器;在new這個任務(wù)管理器的時候,就會啟動一個線程專門去執(zhí)行所有待執(zhí)行的任務(wù);只不過這個時候還沒有添加任務(wù); 3.將這個任務(wù)管理器的默認執(zhí)行器設(shè)置為DumpAllProcessor; 4.每十分鐘執(zhí)行一次往TaskManager中添加一個DumpAllTask的任務(wù);一經(jīng)添加就會被TaskManager中的線程 processingThread 執(zhí)行process方法;