背景

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

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

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

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

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

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

接下來(lái)談?wù)勎覍?duì)動(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)起來(lái)難度比較大，涉及到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腳本來(lái)做這樣的事情，一般情況下重啟來(lái)發(fā)布新的任務(wù)會(huì)比較常見，如果各位一定要實(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á)式來(lái)指定執(zhí)行時(shí)間，不同的是每個(gè)任務(wù)都有自己的參數(shù)，業(yè)務(wù)邏輯都是固定的定時(shí)發(fā)布。
接下來(lái)我給大家介紹下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)添加比較簡(jiǎn)單，只需要接收任務(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ù)也存儲(chǔ)到了注冊(cè)中心，但是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è)問題最簡(jiǎn)單的方式就是將任務(wù)的節(jié)點(diǎn)都集中管理起來(lái)，無(wú)論動(dòng)態(tài)任務(wù)在哪個(gè)節(jié)點(diǎn)上進(jìn)行注冊(cè)，都需要將這個(gè)請(qǐng)求轉(zhuǎn)發(fā)到其他的節(jié)點(diǎn)上進(jìn)行初始化操作，這樣就可以保證多節(jié)點(diǎn)分片的任務(wù)正常執(zhí)行。

還有一種對(duì)使用者更友好的辦法是對(duì)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ā)請(qǐng)求到其他節(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)添加的邏輯，大家引入依賴即可使用。

使用方式比較簡(jiǎn)單，只需要在啟動(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來(lái)動(dòng)態(tài)的注冊(cè)任務(wù)，API列表如下：

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

完整字段請(qǐng)參考：

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請(qǐng)求，參數(shù)只要任務(wù)名稱即可，比如：/job/remove?jobName=任務(wù)名?？梢杂糜谌蝿?wù)完成之后清空注冊(cè)中心的任務(wù)信息。

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

最新評(píng)論