背景

在使用Elastic-Job的過程中，有很多人遇到了這么一個(gè)問題，就是如何動(dòng)態(tài)的去添加任務(wù)？

在官方的文檔中也有對此作出回答，如下：

動(dòng)態(tài)添加作業(yè)這個(gè)概念每個(gè)人理解不盡相同。

elastic-job-lite為jar包，由開發(fā)或運(yùn)維人員負(fù)責(zé)啟動(dòng)。啟動(dòng)時(shí)自動(dòng)向注冊中心注冊作業(yè)信息并進(jìn)行分布式協(xié)調(diào)，因此并不需要手工在注冊中心填寫作業(yè)信息。 但注冊中心與作業(yè)部署機(jī)無從屬關(guān)系，注冊中心并不能控制將單點(diǎn)的作業(yè)分發(fā)至其他作業(yè)機(jī)，也無法將遠(yuǎn)程服務(wù)器未啟動(dòng)的作業(yè)啟動(dòng)。elastic-job-lite并不會包含ssh免密管理等功能。

elastic-job-cloud為mesos框架，由mesos負(fù)責(zé)作業(yè)啟動(dòng)和分發(fā)。 但需要將作業(yè)打包上傳，并調(diào)用elastic-job-cloud提供的REST API寫入注冊中心。 打包上傳屬于部署系統(tǒng)的范疇elastic-job-cloud并未涉及。

綜上所述，elastic-job已做了基本動(dòng)態(tài)添加功能，但無法做到真正意義的完全自動(dòng)化添加。

接下來談?wù)勎覍?dòng)態(tài)任務(wù)的理解，我眼中的動(dòng)態(tài)任務(wù)分為2種：

一種是全新的任務(wù)，包括實(shí)現(xiàn)的邏輯也是全新的，也就是當(dāng)我們的程序打成一個(gè)jar包后，線上已經(jīng)在運(yùn)行了，這個(gè)時(shí)候我加了一個(gè)新的任務(wù)，如何能做到不停服務(wù)，將這個(gè)任務(wù)集成到已有的任務(wù)中去，這個(gè)實(shí)現(xiàn)起來難度比較大，涉及到Java類的熱加載等，不過最近阿里又有一開源大作JarsLink，GitHub地址：https://github.com/alibaba/jarslink，可以支持在運(yùn)行時(shí)動(dòng)態(tài)加載到系統(tǒng)中，實(shí)現(xiàn)不需要重啟和發(fā)布系統(tǒng)新增功能。還有一種實(shí)現(xiàn)思路我們可以利用Groovy腳本來做這樣的事情，一般情況下重啟來發(fā)布新的任務(wù)會比較常見，如果各位一定要實(shí)現(xiàn)動(dòng)態(tài)的任務(wù)可以自己嘗試著去研究下我提供的思路。

另一種就是執(zhí)行的業(yè)務(wù)邏輯不變，只是運(yùn)行的時(shí)間發(fā)生變化。比如文章的定時(shí)發(fā)布，可以設(shè)置文章在某天的某分鐘進(jìn)行自動(dòng)發(fā)布，實(shí)現(xiàn)這個(gè)功能有多種方式，你可以不停的掃描任務(wù)，一到時(shí)間點(diǎn)就自動(dòng)發(fā)布，比較優(yōu)雅的方式就是為每篇文章的自動(dòng)發(fā)布都設(shè)置一個(gè)任務(wù)，通過Cron表達(dá)式來指定執(zhí)行時(shí)間，不同的是每個(gè)任務(wù)都有自己的參數(shù)，業(yè)務(wù)邏輯都是固定的定時(shí)發(fā)布。接下來我給大家介紹下Elastic-Job實(shí)現(xiàn)上面講的第二種動(dòng)態(tài)任務(wù)的方式，也就是任務(wù)的實(shí)現(xiàn)邏輯已經(jīng)是存在的，只是需要發(fā)布成多個(gè)不同時(shí)間去觸發(fā)的任務(wù)。

實(shí)戰(zhàn)

實(shí)現(xiàn)任務(wù)的動(dòng)態(tài)添加比較簡單，只需要接收任務(wù)的信息，然后初始化一下就可以了，在實(shí)現(xiàn)的過程中筆者遇到了一個(gè)麻煩的問題？

在多節(jié)點(diǎn)分片任務(wù)卻只有一個(gè)節(jié)點(diǎn)能執(zhí)行，問題原因在于當(dāng)有任務(wù)A和任務(wù)B，2個(gè)節(jié)點(diǎn)的時(shí)候,我們調(diào)用A節(jié)點(diǎn)的接口進(jìn)行任務(wù)的動(dòng)態(tài)添加，在A節(jié)點(diǎn)中初始化了任務(wù)調(diào)度器，數(shù)據(jù)也存儲到了注冊中心，但是B節(jié)點(diǎn)是不知道有新的任務(wù)添加，默認(rèn)的使用方法是每個(gè)節(jié)點(diǎn)在啟動(dòng)時(shí)去初始化任務(wù)調(diào)度器，而我們的B節(jié)點(diǎn)已經(jīng)啟動(dòng)過了，任務(wù)是新添加的。

解決這個(gè)問題最簡單的方式就是將任務(wù)的節(jié)點(diǎn)都集中管理起來，無論動(dòng)態(tài)任務(wù)在哪個(gè)節(jié)點(diǎn)上進(jìn)行注冊，都需要將這個(gè)請求轉(zhuǎn)發(fā)到其他的節(jié)點(diǎn)上進(jìn)行初始化操作，這樣就可以保證多節(jié)點(diǎn)分片的任務(wù)正常執(zhí)行。

還有一種對使用者更友好的辦法是對Zookeeper中的節(jié)點(diǎn)進(jìn)行監(jiān)聽，當(dāng)有新的節(jié)點(diǎn)創(chuàng)建時(shí)，就自動(dòng)獲取這個(gè)節(jié)點(diǎn)的配置信息，在本地進(jìn)行任務(wù)初始化，通過這樣的方式就可以不用去轉(zhuǎn)發(fā)請求到其他節(jié)點(diǎn)了，只要在任何節(jié)點(diǎn)有添加操作，都能被監(jiān)聽到，并自己去初始化。

監(jiān)控代碼如下：

/** * 開啟任務(wù)監(jiān)聽,當(dāng)有任務(wù)添加時(shí)，監(jiān)聽zk中的數(shù)據(jù)增加，自動(dòng)在其他節(jié)點(diǎn)也初始化該任務(wù) */public void monitorJobRegister() { CuratorFramework client = zookeeperRegistryCenter.getClient(); @SuppressWarnings('resource') PathChildrenCache childrenCache = new PathChildrenCache(client, '/', true); PathChildrenCacheListener childrenCacheListener = new PathChildrenCacheListener() { public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception { ChildData data = event.getData(); switch (event.getType()) { case CHILD_ADDED: String config = new String(client.getData().forPath(data.getPath() + '/config')); Job job = JsonUtils.toBean(Job.class, config); addJob(job); break; default: break; } } }; childrenCache.getListenable().addListener(childrenCacheListener); try { childrenCache.start(StartMode.POST_INITIALIZED_EVENT); } catch (Exception e) { e.printStackTrace(); } }

為了方便大家使用，我將動(dòng)態(tài)添加任務(wù)的功能集成到了我之前的elastic-job-spring-boot-starter（https://github.com/yinjihuan/elastic-job-spring-boot-starter）中集成了動(dòng)態(tài)添加的邏輯，大家引入依賴即可使用。

使用方式比較簡單，只需要在啟動(dòng)類上加一個(gè)ComponentScan注解，讓Spring能夠掃描到elastic-job-spring-boot-starter提供的代碼即可：

@SpringBootApplication@EnableElasticJob//開啟動(dòng)態(tài)任務(wù)添加API@ComponentScan(basePackages = {'com.cxytiandi'})public class JobApplication { public static void main(String[] args) { new SpringApplicationBuilder().sources(JobApplication.class).web(true).run(args); try { new CountDownLatch(1).await(); } catch (InterruptedException e) { } }}

配置好之后，啟動(dòng)項(xiàng)目就可以通過REST API來動(dòng)態(tài)的注冊任務(wù)，API列表如下：

/job添加任務(wù)是POST請求，數(shù)據(jù)格式為JSON體提交，格式如下：{'jobName':'DynamicJob13','cron':'0 33 16 ?','jobType':'SIMPLE','jobClass':'com.cxytiandi.job.demo.DynamicJob','jobParameter':'2222222','shardingTotalCount':1}

完整字段請參考：

https://github.com/yinjihuan/elastic-job-spring-boot-starter/blob/master/spring-boot-elastic-job-starter/src/main/java/com/cxytiandi/elasticjob/dynamic/bean/Job.java

注意：jobClass必須事先存在于服務(wù)中* /job/remove

刪除任務(wù)是GET請求，參數(shù)只要任務(wù)名稱即可，比如：/job/remove?jobName=任務(wù)名。可以用于任務(wù)完成之后清空注冊中心的任務(wù)信息。

以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持好吧啦網(wǎng)。