Elastic-Job - 分佈式定時任務框架

Elastic-Job是ddframe中dd-job的做業模塊中分離出來的分佈式彈性做業框架。去掉了和dd-job中的監控和ddframe接入規範部分。該項目基於成熟的開源產品Quartz和Zookeeper及其客戶端Curator進行二次開發。java

項目開源地址:https://github.com/dangdangdotcom/elastic-jobgit

ddframe其餘模塊也有可獨立開源的部分,以前噹噹曾開源過dd-soa的基石模塊DubboX。github

elastic-job和ddframe關係見下圖web

Elastic-Job主要功能

  • 定時任務: 基於成熟的定時任務做業框架Quartz cron表達式執行定時任務。spring

  • 做業註冊中心: 基於Zookeeper和其客戶端Curator實現的全局做業註冊控制中心。用於註冊,控制和協調分佈式做業執行。api

  • 做業分片: 將一個任務分片成爲多個小任務項在多服務器上同時執行。tomcat

  • 彈性擴容縮容: 運行中的做業服務器崩潰,或新增長n臺做業服務器,做業框架將在下次做業執行前從新分片,不影響當前做業執行。安全

  • 支持多種做業執行模式: 支持OneOff,Perpetual和SequencePerpetual三種做業模式。服務器

  • 失效轉移: 運行中的做業服務器崩潰不會致使從新分片,只會在下次做業啓動時分片。啓用失效轉移功能能夠在本次做業執行過程當中,監測其餘做業服務器空閒,抓取未完成的孤兒分片項執行。多線程

  • 運行時狀態收集: 監控做業運行時狀態,統計最近一段時間處理的數據成功和失敗數量,記錄做業上次運行開始時間,結束時間和下次運行時間。

  • 做業中止,恢復和禁用:用於操做做業啓停,並能夠禁止某做業運行(上線時經常使用)。

  • 被錯過執行的做業重觸發:自動記錄錯過執行的做業,並在上次做業完成後自動觸發。可參考Quartz的misfire。

  • 多線程快速處理數據:使用多線程處理抓取到的數據,提高吞吐量。

  • 冪等性:重複做業任務項斷定,不重複執行已運行的做業任務項。因爲開啓冪等性須要監聽做業運行狀態,對瞬時反覆運行的做業對性能有較大影響。

  • 容錯處理:做業服務器與Zookeeper服務器通訊失敗則當即中止做業運行,防止做業註冊中心將失效的分片分項配給其餘做業服務器,而當前做業服務器仍在執行任務,致使重複執行。

  • Spring支持:支持spring容器,自定義命名空間,支持佔位符。

  • 運維平臺:提供運維界面,能夠管理做業和註冊中心。

目錄結構說明

  • elastic-job-core

    elastic-job核心模塊,只經過Quartz和Curator就可執行分佈式做業。

  • elastic-job-spring

    elastic-job對spring支持的模塊,包括命名空間,依賴注入,佔位符等。

  • elastic-job-console

    elastic-job web控制檯,可將編譯以後的war放入tomcat等servlet容器中使用。

  • elastic-job-example

    使用例子。

  • elastic-job-test

    測試elastic-job使用的公用類,使用方無需關注。

引入maven依賴

    elastic-job已經發布到中央倉庫,能夠在pom.xml文件中直接引入maven座標。
      <!-- 引入elastic-job核心模塊 -->
      <dependency>
          <groupId>com.dangdang</groupId>
          <artifactId>elastic-job-core</artifactId>
          <version>1.0.1</version>
      </dependency>
      <!-- 使用springframework自定義命名空間時引入 -->
      <dependency>
          <groupId>com.dangdang</groupId>
          <artifactId>elastic-job-spring</artifactId>
          <version>1.0.1</version>
      </dependency>

代碼開發

提供3種做業類型,分別是OneOff, Perpetual和SequencePerpetual。須要繼承相應的抽象類。

方法參數shardingContext包含做業配置,分片和運行時信息。可經過getShardingTotalCount(),getShardingItems()等方法分別獲取分片總數,運行在本做業服務器的分片序列號集合等。

  • OneOff類型做業

    OneOff做業類型比較簡單,須要繼承AbstractOneOffElasticJob,該類只提供了一個方法用於覆蓋,此方法將被定時執行。用於執行普通的定時任務,與Quartz原生接口類似,只是增長了彈性擴縮容和分片等功能。

public class MyElasticJob extends AbstractOneOffElasticJob {

    @Override
    protected void process(JobExecutionMultipleShardingContext context) {
        // do something by sharding items
    }
}
  • Perpetual類型做業

    Perpetual做業類型略爲複雜,須要繼承AbstractPerpetualElasticJob並能夠指定返回值泛型,該類提供兩個方法可覆蓋,分別用於抓取和處理數據。能夠獲取數據處理成功失敗次數等輔助監控信息。須要注意fetchData方法的返回值只有爲null或長度爲空時,做業纔會中止執行,不然做業會一直運行下去。這點是參照TbSchedule的設計。Perpetual做業類型更適用於流式不間歇的數據處理。

    做業執行時會將fetchData的數據傳遞給processData處理,其中processData獲得的數據是經過多線程(線程池大小可配)拆分的。建議processData處理數據後,更新其狀態,避免fetchData再次抓取到,從而使得做業永遠不會中止。processData的返回值用於表示數據是否處理成功,拋出異常或者返回false將會在統計信息中納入失敗次數,返回true則納入成功次數。

public class MyElasticJob extends AbstractPerpetualElasticJob<Foo> {

    @Override
    protected List<Foo> fetchData(JobExecutionMultipleShardingContext context) {
        List<Foo> result = // get data from database by sharding items
        return result;
    }
    
    @Override
    protected boolean processData(JobExecutionMultipleShardingContext context, Foo data) {
        // process data
        return true;
    }
}
  • SequencePerpetual類型做業

    SequencePerpetual做業類型和Perpetual做業類型極爲類似,所不一樣的是Perpetual做業類型能夠將獲取到的數據多線程處理,但不會保證多線程處理數據的順序。如:從2個分片共獲取到100條數據,第1個分片40條,第2個分片60條,配置爲兩個線程處理,則第1個線程處理前50條數據,第2個線程處理後50條數據,無視分片項;SequencePerpetual類型做業則根據當前服務器所分配的分片項數量進行多線程處理,每一個分片項使用同一線程處理,防止了同一分片的數據被多線程處理,從而致使的順序問題。如:從2個分片共獲取到100條數據,第1個分片40條,第2個分片60條,則系統自動分配兩個線程處理,第1個線程處理第1個分片的40條數據,第2個線程處理第2個分片的60條數據。因爲Perpetual做業可使用多餘分片項的任意線程數處理,因此性能調優的可能會優於SequencePerpetual做業。

public class MyElasticJob extends AbstractSequencePerpetualElasticJob<Foo> {

    @Override
    protected List<Foo> fetchData(JobExecutionSingleShardingContext context) {
        List<Foo> result = // get data from database by sharding items
        return result;
    }
    
    @Override
    protected boolean processData(JobExecutionSingleShardingContext context, Foo data) {
        // process data
        return true;
    }
}

做業配置

與Spring容器配合使用做業,能夠將做業Bean配置爲Spring Bean, 可在做業中經過依賴注入使用Spring容器管理的數據源等對象。可用placeholder佔位符從屬性文件中取值。

  • Spring命名空間配置

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:reg="http://www.dangdang.com/schema/ddframe/reg" 
    xmlns:job="http://www.dangdang.com/schema/ddframe/job" 
    xsi:schemaLocation="http://www.springframework.org/schema/beans
                        http://www.springframework.org/schema/beans/spring-beans.xsd
                        http://www.dangdang.com/schema/ddframe/reg
                        http://www.dangdang.com/schema/ddframe/reg/reg.xsd
                        http://www.dangdang.com/schema/ddframe/job
                        http://www.dangdang.com/schema/ddframe/job/job.xsd
                        ">
    <!--配置做業註冊中心 -->
    <reg:zookeeper id="regCenter" serverLists=" yourhost:2181" namespace="dd-job" baseSleepTimeMilliseconds="1000" maxSleepTimeMilliseconds="3000" maxRetries="3" />
    <!-- 配置做業A-->
    <job:bean id="oneOffElasticJob" class="xxx.MyOneOffElasticJob" regCenter="regCenter" cron="0/10 * * * * ?"   shardingTotalCount="3" shardingItemParameters="0=A,1=B,2=C" />
    <!-- 配置做業B-->
    <job:bean id="perpetualElasticJob" class="xxx.MyPerpetualElasticJob" regCenter="regCenter" cron="0/10 * * * * ?" shardingTotalCount="3" shardingItemParameters="0=A,1=B,2=C" processCountIntervalSeconds="10" concurrentDataProcessThreadCount="10" />
</beans>


<job:bean />命名空間屬性詳細說明

<reg:zookeeper />命名空間屬性詳細說明


  • 基於Spring但不使用命名空間

    <!-- 配置做業註冊中心 -->
    <bean id="regCenter" class="com.dangdang.ddframe.reg.zookeeper.ZookeeperRegistryCenter" init-method="init">
        <constructor-arg>
            <bean class="com.dangdang.ddframe.reg.zookeeper.ZookeeperConfiguration">
                <property name="serverLists" value="${xxx}" />
                <property name="namespace" value="${xxx}" />
                <property name="baseSleepTimeMilliseconds" value="${xxx}" />
                <property name="maxSleepTimeMilliseconds" value="${xxx}" />
                <property name="maxRetries" value="${xxx}" />
            </bean>
        </constructor-arg>
    </bean>    <!-- 配置做業-->
    <bean id="xxxJob" class="com.dangdang.ddframe.job.spring.schedule.SpringJobController" init-method="init">
        <constructor-arg ref="regCenter" />
        <constructor-arg>
            <bean class="com.dangdang.ddframe.job.api.JobConfiguration">
                <constructor-arg name="jobName" value="xxxJob" />
                <constructor-arg name="jobClass" value="xxxDemoJob" />
                <constructor-arg name="shardingTotalCount" value="10" />
                <constructor-arg name="cron" value="0/10 * * * * ?" />
                <property name="shardingItemParameters" value="${xxx}" />
            </bean>
        </constructor-arg>
    </bean>
  • 不使用Spring配置

若是不使用Spring框架,能夠用以下方式啓動做業。

import com.dangdang.ddframe.job.api.JobConfiguration;
import com.dangdang.ddframe.job.schedule.JobController;
import com.dangdang.ddframe.reg.base.CoordinatorRegistryCenter;
import com.dangdang.ddframe.reg.zookeeper.ZookeeperConfiguration;
import com.dangdang.ddframe.reg.zookeeper.ZookeeperRegistryCenter;
import com.dangdang.example.elasticjob.core.job.OneOffElasticDemoJob;
import com.dangdang.example.elasticjob.core.job.PerpetualElasticDemoJob;
import com.dangdang.example.elasticjob.core.job.SequencePerpetualElasticDemoJob;

public class JobDemo {

    // 定義Zookeeper註冊中心配置對象
    private ZookeeperConfiguration zkConfig = new ZookeeperConfiguration("localhost:2181", "elastic-job-example", 1000, 3000, 3);
    
    // 定義Zookeeper註冊中心
    private CoordinatorRegistryCenter regCenter = new ZookeeperRegistryCenter(zkConfig);
    
    // 定義做業1配置對象
    private JobConfiguration jobConfig1 = new JobConfiguration("oneOffElasticDemoJob", OneOffElasticDemoJob.class, 10, "0/5 * * * * ?");
    
    // 定義做業2配置對象
    private JobConfiguration jobConfig2 = new JobConfiguration("perpetualElasticDemoJob", PerpetualElasticDemoJob.class, 10, "0/5 * * * * ?");
    
    // 定義做業3配置對象
    private JobConfiguration jobConfig3 = new JobConfiguration("sequencePerpetualElasticDemoJob", SequencePerpetualElasticDemoJob.class, 10, "0/5 * * * * ?");
    
    public static void main(final String[] args) {
        new JobDemo().init();
    }
    
    private void init() {
        // 鏈接註冊中心
        regCenter.init();
        // 啓動做業1
        new JobController(regCenter, jobConfig1).init();
        // 啓動做業2
        new JobController(regCenter, jobConfig2).init();
        // 啓動做業3
        new JobController(regCenter, jobConfig3).init();
    }
}

使用限制

  • 做業一旦啓動成功後不能修改做業名稱,若是修更名稱則視爲新的做業。

  • 同一臺做業服務器只能運行一個相同的做業實例,由於做業運行時是按照IP註冊和管理的。

  • 做業根據/etc/hosts文件獲取IP地址,若是獲取的IP地址是127.0.0.1而非真實IP地址,應正確配置此文件。

  • 一旦有服務器波動,或者修改分片項,將會觸發從新分片;觸發從新分片將會致使運行中的Perpetual以及SequencePerpetual做業再執行完本次做業後再也不繼續執行,等待分片結束後再恢復正常。

  • 開啓monitorExecution才能實現分佈式做業冪等性(即不會在多個做業服務器運行同一個分片)的功能,但monitorExecution對短期內執行的做業(如每5秒一觸發)性能影響較大,建議關閉並自行實現冪等性。

  • elastic-job沒有自動刪除做業服務器的功能,由於沒法區分是服務器崩潰仍是正常下線。因此若是要下線服務器,須要手工刪除zookeeper中相關的服務器節點。因爲直接刪除服務器節點風險較大,暫時不考慮在運維平臺增長此功能

實現原理

  • 彈性分佈式實現

    1. 第一臺服務器上線觸發主服務器選舉。主服務器一旦下線,則從新觸發選舉,選舉過程當中阻塞,只有主服務器選舉完成,纔會執行其餘任務。

    2. 某做業服務器上線時會自動將服務器信息註冊到註冊中心,下線時會自動更新服務器狀態。

    3. 主節點選舉,服務器上下線,分片總數變動均更新從新分片標記。

    4. 定時任務觸發時,如需從新分片,則經過主服務器分片,分片過程當中阻塞,分片結束後纔可執行任務。如分片過程當中主服務器下線,則先選舉主服務器,再分片。

    5. 經過4可知,爲了維持做業運行時的穩定性,運行過程當中只會標記分片狀態,不會從新分片。分片僅可能發生在下次任務觸發前。

    6. 每次分片都會按服務器IP排序,保證分片結果不會產生較大波動。

    7. 實現失效轉移功能,在某臺服務器執行完畢後主動抓取未分配的分片,而且在某臺服務器下線後主動尋找可用的服務器執行任務。

  • 流程圖

    做業啓動


    做業執行


運維平臺

    elastic-job運維平臺以war包形式提供,可自行部署到tomcat或jetty等支持servlet的web容器中。elastic-job-console.war能夠經過編譯源碼或從maven中央倉庫獲取。

  • 登陸

    默認用戶名和密碼是root/root,能夠經過修改conf\auth.properties文件修改默認登陸用戶名和密碼。

  • 主要功能

    登陸安全控制

    註冊中心管理

    做業維度狀態查看

    服務器維度狀態查看

    快捷修改做業設置

    控制做業暫停和恢復運行

  • 設計理念

    運維平臺和elastic-job並沒有直接關係,是經過讀取做業註冊中心數據展示做業狀態,或更新註冊中心數據修改全局配置。

    控制檯只能控制做業自己是否運行,但不能控制做業進程的啓停,由於控制檯和做業自己服務器是徹底分佈式的,控制檯並不能控制做業服務器。

  • 不支持項

    添加做業。由於做業都是在首次運行時自動添加,使用運維平臺添加做業並沒有必要。

    中止做業。即便刪除了Zookeeper信息也不能真正中止做業的運行,還會致使運行中的做業出問題。

    刪除做業服務器。因爲直接刪除服務器節點風險較大,暫時不考慮在運維平臺增長此功能。

  • 主要界面

  • 總覽頁


  • 註冊中心管理頁


  • 做業詳細信息頁


  • 服務區詳細信息頁

相關文章
相關標籤/搜索