[[378874]]

本文轉載自微信公眾號「Java極客技術」，作者鴨血粉絲 。轉載本文請聯系Java極客技術公眾號。  

一、介紹

在前幾篇文章中，我們詳細的介紹了 Quartz 的架構原理以及應用實踐，雖然 Quartz 也可以通過集群方式來保證服務高可用，但是它也有一個的弊端，那就是服務節點數量的增加，并不能提升任務的執行效率，即不能實現水平擴展!

之所以產生這樣的結果，是因為 Quartz 在分布式集群環境下是通過數據庫鎖方式來實現有且只有一個有效的服務節點來運行服務，從而保證服務在集群環境下定時任務不會被重復調用!

如果需要運行的定時任務很少的話，使用 Quartz 不會有太大的問題，但是如果 現在有這么一個需求，例如理財產品，每天6點系統需要計算每個賬戶昨天的收益，假如這個理財產品，有幾個億的用戶，如果都在一個服務實例上跑，可能第二天都無法處理完這項任務!

類似這樣場景還有很多很多，很顯然 Quartz 很難滿足我們這種大批量、任務執行周期長的任務調度!

因此短板，當當網基于 Quartz 開發了一套適合在分布式環境下能高效率的使用服務器資源的 Elastic-Job 定時任務框架!

Elastic-Job-Lite最大的亮點就是支持彈性擴容縮容，怎么實現的呢?

比如現在有個任務要執行，如果將任務進行分片成10個，那么可以同時在10個服務實例上并行執行，互相不影響，從而大大的提升了任務執行效率，并且充分的利用服務器資源!

對于上面的理財產品，如果這個任務需要處理1個億用戶，那么我們可以通過水平擴展，比如對任務進行分片為500，讓500個服務實例同時運行，每個服務實例處理20萬條數據，不出意外的話，1 - 2個小時可以全部跑完，如果時間還是很長，還可以繼續水平擴張，添加服務實例來運行!

2015 年，當當網將其開源，瞬間吸引了一大批程序員的關注，同時登頂開源中國第一名!

下面我們就一起來了解一下這款使用非常廣泛的分布式調度框架。

二、項目架構介紹

Elastic-Job 最開始只有一個 elastic-job-core 的項目，定位輕量級、無中心化，最核心的服務就是支持彈性擴容和數據分片!

從 2.X 版本以后，主要分為 Elastic-Job-Lite 和 Elastic-Job-Cloud 兩個子項目。

其中，Elastic-Job-Lite 定位為輕量級 無 中 心 化 解 決 方 案 ， 使 用jar 包 的 形 式 提 供 分 布 式 任 務 的 協 調 服 務 。

而 Elastic-Job-Cloud 使用 Mesos + Docker 的解決方案，額外提供資源治理、應用分發以及進程隔離等服務(跟 Lite 的區別只是部署方式不同，他們使用相同的 API，只要開發一次)。

今天我們主要介紹的是Elastic-Job-Lite，最主要的功能特性如下：

• 分布式調度協調：采用 zookeeper 實現注冊中心，進行統一調度。
• 支持任務分片：將需要執行的任務進行分片，實現并行調度。
• 支持彈性擴容縮容：將任務拆分為 n 個任務項后，各個服務器分別執行各自分配到的任務項。一旦有新的服務器加入集群，或現有服務器下線，elastic-job 將在保留本次任務執行不變的情況下，下次任務開始前觸發任務重分片。

當然，還有失效轉移、錯過執行作業重觸發等等功能，大家可以訪問官網文檔，以獲取更多詳細資料。

應用在各自的節點執行任務，通過 zookeeper 注冊中心協調。節點注冊、節點選舉、任務分片、監聽都在 E-Job 的代碼中完成。下圖是官網提供得架構圖。

啥也不用多說了，下面我們直接通過實踐介紹，更容易了解里面是怎么玩的!

三、應用實踐

3.1、zookeeper 安裝

elastic-job-lite，是直接依賴 zookeeper 的，因此在開發之前我們需要先準備好對應的 zookeeper 環境，關于 zookeeper 的安裝過程，就不多說了，非常簡單，網上都有教程!

3.2、elastic-job-lite-console 安裝

elastic-job-lite-console，主要是一個任務作業可視化界面管理系統。

可以單獨部署，與平臺不關，主要是通過配置注冊中心和數據源來抓取數據。

獲取的方式也很簡單，直接訪問https://github.com/apache/shardingsphere-elasticjob地址，然后切換到2.1.5的版本號，然后執行mvn clean install進行打包，獲取對應的安裝包將其解壓，進行bin文件夾啟動服務即可!

如果你的網速像蝸牛一樣的慢，還有一個辦法就是從這個地址https://gitee.com/elasticjob/elastic-job獲取對應的源碼!

啟動服務后，在瀏覽器訪問http://127.0.0.1:8899，輸入賬戶、密碼(都是root)即可進入控制臺頁面，類似如下界面!

進入之后，將上文所在的 zookeeper 注冊中心進行配置，包括數據庫 mysql 的數據源也可以配置一下!

3.3、創建工程

本文采用springboot來搭建工程為例，創建工程并添加elastic-job-lite依賴!

<!-- 引入elastic-job-lite核心模塊 --> 
<dependency> 
    <groupId>com.dangdang</groupId> 
    <artifactId>elastic-job-lite-core</artifactId> 
    <version>2.1.5</version> 
</dependency> 
 
<!-- 使用springframework自定義命名空間時引入 --> 
<dependency> 
    <groupId>com.dangdang</groupId> 
    <artifactId>elastic-job-lite-spring</artifactId> 
    <version>2.1.5</version> 
</dependency> 

在配置文件application.properties中提前配置好 zookeeper 注冊中心相關信息!

#zookeeper config 
zookeeper.serverList=127.0.0.1:2181 
zookeeper.namespace=example-elastic-job-test 

3.4、新建 ZookeeperConfig 配置類

@Configuration 
@ConditionalOnExpression("'${zookeeper.serverList}'.length() > 0") 
public class ZookeeperConfig { 
 
    /** 
     * zookeeper 配置 
     * @return 
     */ 
    @Bean(initMethod = "init") 
    public ZookeeperRegistryCenter zookeeperRegistryCenter(@Value("${zookeeper.serverList}") String serverList,  
                                                           @Value("${zookeeper.namespace}") String namespace){ 
        return new ZookeeperRegistryCenter(new ZookeeperConfiguration(serverList,namespace)); 
    } 
 
} 

3.5、新建任務處理類

elastic-job支持三種類型的作業任務處理!

• Simple 類型作業：Simple 類型用于一般任務的處理，只需實現SimpleJob接口。該接口僅提供單一方法用于覆蓋，此方法將定時執行，與Quartz原生接口相似。
• Dataflow 類型作業：Dataflow 類型用于處理數據流，需實現DataflowJob接口。該接口提供2個方法可供覆蓋，分別用于抓取(fetchData)和處理(processData)數據。
• Script類型作業：Script 類型作業意為腳本類型作業，支持 shell，python，perl等所有類型腳本。只需通過控制臺或代碼配置 scriptCommandLine 即可，無需編碼。執行腳本路徑可包含參數，參數傳遞完畢后，作業框架會自動追加最后一個參數為作業運行時信息。

3.6、新建 Simple 類型作業

編寫一個SimpleJob接口的實現類MySimpleJob，當前工作主要是打印一條日志。

@Slf4j 
public class MySimpleJob implements SimpleJob { 
 
    @Override 
    public void execute(ShardingContext shardingContext) { 
        log.info(String.format("Thread ID: %s, 作業分片總數: %s, " + 
                        "當前分片項: %s.當前參數: %s," + 
                        "作業名稱: %s.作業自定義參數: %s" 
                , 
                Thread.currentThread().getId(), 
                shardingContext.getShardingTotalCount(), 
                shardingContext.getShardingItem(), 
                shardingContext.getShardingParameter(), 
                shardingContext.getJobName(), 
                shardingContext.getJobParameter() 
        )); 
    } 
} 

創建一個MyElasticJobListener任務監聽器，用于監聽MySimpleJob的任務執行情況。

@Slf4j 
public class MyElasticJobListener implements ElasticJobListener { 
 
    private long beginTime = 0; 
 
    @Override 
    public void beforeJobExecuted(ShardingContexts shardingContexts) { 
        beginTime = System.currentTimeMillis(); 
        log.info("===>{} MyElasticJobListener BEGIN TIME: {} <===",shardingContexts.getJobName(),  DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss")); 
    } 
 
    @Override 
    public void afterJobExecuted(ShardingContexts shardingContexts) { 
        long endTime = System.currentTimeMillis(); 
        log.info("===>{} MyElasticJobListener END TIME: {},TOTAL CAST: {} <===",shardingContexts.getJobName(), DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss"), endTime - beginTime); 
    } 
 
} 

創建一個MySimpleJobConfig類，將MySimpleJob其注入到zookeeper。

@Configuration 
public class MySimpleJobConfig { 
 
    /** 
     * 任務名稱 
     */ 
    @Value("${simpleJob.mySimpleJob.name}") 
    private String mySimpleJobName; 
 
    /** 
     * cron表達式 
     */ 
    @Value("${simpleJob.mySimpleJob.cron}") 
    private String mySimpleJobCron; 
 
    /** 
     * 作業分片總數 
     */ 
    @Value("${simpleJob.mySimpleJob.shardingTotalCount}") 
    private int mySimpleJobShardingTotalCount; 
 
    /** 
     * 作業分片參數 
     */ 
    @Value("${simpleJob.mySimpleJob.shardingItemParameters}") 
    private String mySimpleJobShardingItemParameters; 
 
    /** 
     * 自定義參數 
     */ 
    @Value("${simpleJob.mySimpleJob.jobParameters}") 
    private String mySimpleJobParameters; 
 
    @Autowired 
    private ZookeeperRegistryCenter registryCenter; 
 
    @Bean 
    public MySimpleJob mySimpleJob() { 
        return new MySimpleJob(); 
    } 
 
    @Bean(initMethod = "init") 
    public JobScheduler simpleJobScheduler(final MySimpleJob mySimpleJob) { 
  //配置任務監聽器 
   MyElasticJobListener elasticJobListener = new MyElasticJobListener(); 
        return new SpringJobScheduler(mySimpleJob, registryCenter, getLiteJobConfiguration(), elasticJobListener); 
    } 
 
    private LiteJobConfiguration getLiteJobConfiguration() { 
        // 定義作業核心配置 
        JobCoreConfiguration simpleCoreConfig = JobCoreConfiguration.newBuilder(mySimpleJobName, mySimpleJobCron, mySimpleJobShardingTotalCount). 
                shardingItemParameters(mySimpleJobShardingItemParameters).jobParameter(mySimpleJobParameters).build(); 
        // 定義SIMPLE類型配置 
        SimpleJobConfiguration simpleJobConfig = new SimpleJobConfiguration(simpleCoreConfig, MySimpleJob.class.getCanonicalName()); 
        // 定義Lite作業根配置 
        LiteJobConfiguration simpleJobRootConfig = LiteJobConfiguration.newBuilder(simpleJobConfig).overwrite(true).build(); 
        return simpleJobRootConfig; 
 
    } 
} 

在配置文件application.properties中配置好對應的mySimpleJob參數!

#elastic job 
#simpleJob類型的job 
simpleJob.mySimpleJob.name=mySimpleJob 
simpleJob.mySimpleJob.cron=0/15 * * * * ? 
simpleJob.mySimpleJob.shardingTotalCount=3 
simpleJob.mySimpleJob.shardingItemParameters=0=a,1=b,2=c 
simpleJob.mySimpleJob.jobParameters=helloWorld 

運行程序，看看效果如何?

在上圖demo中，配置的分片數為3，這個時候會有3個線程進行同時執行任務，因為都是在一臺機器上執行的，這個任務被執行來3次，下面修改一下端口配置，創建三個相同的服務實例，在看看效果如下：

很清晰的看到任務被執行一次!

3.7、新建 DataFlowJob 類型作業

DataFlowJob 類型的任務配置和SimpleJob類似，操作也很簡單!

創建一個DataflowJob類型的實現類MyDataFlowJob。

@Slf4j 
public class MyDataFlowJob implements DataflowJob<String> { 
 
    private boolean flag = false; 
 
    @Override 
    public List<String> fetchData(ShardingContext shardingContext) { 
        log.info("開始獲取數據"); 
        if (flag) { 
            return null; 
        } 
        return Arrays.asList("qingshan", "jack", "seven"); 
    } 
 
    @Override 
    public void processData(ShardingContext shardingContext, List<String> data) { 
        for (String val : data) { 
            // 處理完數據要移除掉，不然就會一直跑,處理可以在上面的方法里執行。這里采用 flag 
            log.info("開始處理數據：" + val); 
        } 
        flag = true; 
    } 
} 

接著創建MyDataFlowJob的配置類，將其注入到zookeeper注冊中心。

Configuration 
public class MyDataFlowJobConfig { 
 
    /** 
     * 任務名稱 
     */ 
    @Value("${dataflowJob.myDataflowJob.name}") 
    private String jobName; 
 
    /** 
     * cron表達式 
     */ 
    @Value("${dataflowJob.myDataflowJob.cron}") 
    private String jobCron; 
 
    /** 
     * 作業分片總數 
     */ 
    @Value("${dataflowJob.myDataflowJob.shardingTotalCount}") 
    private int jobShardingTotalCount; 
 
    /** 
     * 作業分片參數 
     */ 
    @Value("${dataflowJob.myDataflowJob.shardingItemParameters}") 
    private String jobShardingItemParameters; 
 
    /** 
     * 自定義參數 
     */ 
    @Value("${dataflowJob.myDataflowJob.jobParameters}") 
    private String jobParameters; 
 
    @Autowired 
    private ZookeeperRegistryCenter registryCenter; 
 
 
    @Bean 
    public MyDataFlowJob myDataFlowJob() { 
        return new MyDataFlowJob(); 
    } 
 
    @Bean(initMethod = "init") 
    public JobScheduler dataFlowJobScheduler(final MyDataFlowJob myDataFlowJob) { 
        MyElasticJobListener elasticJobListener = new MyElasticJobListener(); 
        return new SpringJobScheduler(myDataFlowJob, registryCenter, getLiteJobConfiguration(), elasticJobListener); 
    } 
 
    private LiteJobConfiguration getLiteJobConfiguration() { 
        // 定義作業核心配置 
        JobCoreConfiguration dataflowCoreConfig = JobCoreConfiguration.newBuilder(jobName, jobCron, jobShardingTotalCount). 
                shardingItemParameters(jobShardingItemParameters).jobParameter(jobParameters).build(); 
        // 定義DATAFLOW類型配置 
        DataflowJobConfiguration dataflowJobConfig = new DataflowJobConfiguration(dataflowCoreConfig, MyDataFlowJob.class.getCanonicalName(), false); 
        // 定義Lite作業根配置 
        LiteJobConfiguration dataflowJobRootConfig = LiteJobConfiguration.newBuilder(dataflowJobConfig).overwrite(true).build(); 
        return dataflowJobRootConfig; 
 
    } 
} 

最后，在配置文件application.properties中配置好對應的myDataflowJob參數!

#dataflow類型的job 
dataflowJob.myDataflowJob.name=myDataflowJob 
dataflowJob.myDataflowJob.cron=0/15 * * * * ? 
dataflowJob.myDataflowJob.shardingTotalCount=1 
dataflowJob.myDataflowJob.shardingItemParameters=0=a,1=b,2=c 
dataflowJob.myDataflowJob.jobParameters=myDataflowJobParamter 

運行程序，看看效果如何?

需要注意的地方是，如果配置的是流式處理類型，它會不停的拉取數據、處理數據，在拉取的時候，如果返回為空，就不會處理數據!

如果配置的是非流式處理類型，和上面介紹的simpleJob類型，處理一樣!

3.8、新建 ScriptJob 類型作業

ScriptJob 類型的任務配置和上面類似，主要是用于定時執行某個腳本，一般用的比較少!

因為目標是腳本，沒有執行的任務，所以無需編寫任務作業類型!

只需要編寫一個ScriptJob類型的配置類即可，命令是echo 'Hello World !內容!

@Configuration 
public class MyScriptJobConfig { 
 
    /** 
     * 任務名稱 
     */ 
    @Value("${scriptJob.myScriptJob.name}") 
    private String jobName; 
 
    /** 
     * cron表達式 
     */ 
    @Value("${scriptJob.myScriptJob.cron}") 
    private String jobCron; 
 
    /** 
     * 作業分片總數 
     */ 
    @Value("${scriptJob.myScriptJob.shardingTotalCount}") 
    private int jobShardingTotalCount; 
 
    /** 
     * 作業分片參數 
     */ 
    @Value("${scriptJob.myScriptJob.shardingItemParameters}") 
    private String jobShardingItemParameters; 
 
    /** 
     * 自定義參數 
     */ 
    @Value("${scriptJob.myScriptJob.jobParameters}") 
    private String jobParameters; 
 
    @Autowired 
    private ZookeeperRegistryCenter registryCenter; 
 
 
    @Bean(initMethod = "init") 
    public JobScheduler scriptJobScheduler() { 
        MyElasticJobListener elasticJobListener = new MyElasticJobListener(); 
        return new JobScheduler(registryCenter, getLiteJobConfiguration(), elasticJobListener); 
    } 
 
    private LiteJobConfiguration getLiteJobConfiguration() { 
        // 定義作業核心配置 
        JobCoreConfiguration scriptCoreConfig = JobCoreConfiguration.newBuilder(jobName, jobCron, jobShardingTotalCount). 
                shardingItemParameters(jobShardingItemParameters).jobParameter(jobParameters).build(); 
        // 定義SCRIPT類型配置 
        ScriptJobConfiguration scriptJobConfig = new ScriptJobConfiguration(scriptCoreConfig, "echo 'Hello World !'"); 
        // 定義Lite作業根配置 
        LiteJobConfiguration scriptJobRootConfig = LiteJobConfiguration.newBuilder(scriptJobConfig).overwrite(true).build(); 
        return scriptJobRootConfig; 
 
    } 
} 

在配置文件application.properties中配置好對應的myScriptJob參數!

#script類型的job 
scriptJob.myScriptJob.name=myScriptJob 
scriptJob.myScriptJob.cron=0/15 * * * * ? 
scriptJob.myScriptJob.shardingTotalCount=3 
scriptJob.myScriptJob.shardingItemParameters=0=a,1=b,2=c 
scriptJob.myScriptJob.jobParameters=myScriptJobParamter 

運行程序，看看效果如何?

3.9、將任務狀態持久化到數據庫

可能有的人會發出疑問，elastic-job是如何存儲數據的，用ZooInspector客戶端鏈接zookeeper注冊中心，你發現對應的任務配置被存儲到相應的樹根上!

而具體作業任務執行軌跡和狀態結果是不會存儲到zookeeper，需要我們在項目中通過數據源方式進行持久化!

將任務狀態持久化到數據庫配置過程也很簡單，只需要在對應的配置類上注入數據源即可，以MySimpleJobConfig為例，代碼如下：

@Configuration 
public class MySimpleJobConfig { 
 
    /** 
     * 任務名稱 
     */ 
    @Value("${simpleJob.mySimpleJob.name}") 
    private String mySimpleJobName; 
 
    /** 
     * cron表達式 
     */ 
    @Value("${simpleJob.mySimpleJob.cron}") 
    private String mySimpleJobCron; 
 
    /** 
     * 作業分片總數 
     */ 
    @Value("${simpleJob.mySimpleJob.shardingTotalCount}") 
    private int mySimpleJobShardingTotalCount; 
 
    /** 
     * 作業分片參數 
     */ 
    @Value("${simpleJob.mySimpleJob.shardingItemParameters}") 
    private String mySimpleJobShardingItemParameters; 
 
    /** 
     * 自定義參數 
     */ 
    @Value("${simpleJob.mySimpleJob.jobParameters}") 
    private String mySimpleJobParameters; 
 
    @Autowired 
    private ZookeeperRegistryCenter registryCenter; 
 
    @Autowired 
    private DataSource dataSource;; 
 
 
    @Bean 
    public MySimpleJob stockJob() { 
        return new MySimpleJob(); 
    } 
 
    @Bean(initMethod = "init") 
    public JobScheduler simpleJobScheduler(final MySimpleJob mySimpleJob) { 
        //添加事件數據源配置 
        JobEventConfiguration jobEventConfig = new JobEventRdbConfiguration(dataSource); 
        MyElasticJobListener elasticJobListener = new MyElasticJobListener(); 
        return new SpringJobScheduler(mySimpleJob, registryCenter, getLiteJobConfiguration(), jobEventConfig, elasticJobListener); 
    } 
 
    private LiteJobConfiguration getLiteJobConfiguration() { 
        // 定義作業核心配置 
        JobCoreConfiguration simpleCoreConfig = JobCoreConfiguration.newBuilder(mySimpleJobName, mySimpleJobCron, mySimpleJobShardingTotalCount). 
                shardingItemParameters(mySimpleJobShardingItemParameters).jobParameter(mySimpleJobParameters).build(); 
        // 定義SIMPLE類型配置 
        SimpleJobConfiguration simpleJobConfig = new SimpleJobConfiguration(simpleCoreConfig, MySimpleJob.class.getCanonicalName()); 
        // 定義Lite作業根配置 
        LiteJobConfiguration simpleJobRootConfig = LiteJobConfiguration.newBuilder(simpleJobConfig).overwrite(true).build(); 
        return simpleJobRootConfig; 
 
    } 
} 

同時，需要在配置文件application.properties中配置好對應的datasource參數!

spring.datasource.url=jdbc:mysql://127.0.0.1:3306/example-elastic-job-test 
spring.datasource.username=root 
spring.datasource.password=root 
spring.datasource.driver-class-name=com.mysql.jdbc.Driver 

運行程序，然后在elastic-job-lite-console控制臺配置對應的數據源!

最后，點擊【作業軌跡】即可查看對應作業執行情況!

四、小結

本文主要圍繞elasticjob的使用進行簡單介紹，希望大家有所收獲!

在分布式環境環境下，elastic-job-lite支持的彈性擴容、任務分片是最大的亮點，在實際使用的時候，任務分片總數盡可能大于服務實例個數，并且是倍數關系，這樣任務在分片的時候，會更加均勻!

如果想深入的了解elasticjob，大家可以訪問官方文檔，獲取更加詳細的使用教程!

五、參考1、elasticjob - 官方文檔

2、博客園 - 吳振照 - 任務調度之 Elastic Job