主要功能

定時任務： 基於成熟的定時任務作業框架Quartz cron表示式執行定時任務。

作業註冊中心： 基於Zookeeper和其客戶端Curator實現的全域性作業註冊控制中心。用於註冊，控制和協調分散式作業執行。

作業分片： 將一個任務分片成為多個小任務項在多伺服器上同時執行。

彈性擴容縮容： 執行中的作業伺服器崩潰，或新增加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 {

@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 {

@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>

基於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中相關的伺服器節點。由於直接刪除伺服器節點風險較大，暫時不考慮在運維平臺增加此功能

實現原理

彈性分散式實現

第一臺伺服器上線觸發主伺服器選舉。主伺服器一旦下線，則重新觸發選舉，選舉過程中阻塞，只有主伺服器選舉完成，才會執行其他任務。

某作業伺服器上線時會自動將伺服器資訊註冊到註冊中心，下線時會自動更新伺服器狀態。

主節點選舉，伺服器上下線，分片總數變更均更新重新分片標記。

定時任務觸發時，如需重新分片，則通過主伺服器分片，分片過程中阻塞，分片結束後才可執行任務。如分片過程中主伺服器下線，則先選舉主伺服器，再分片。

通過4可知，為了維持作業執行時的穩定性，執行過程中只會標記分片狀態，不會重新分片。分片僅可能發生在下次任務觸發前。

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

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