快捷導(dǎo)航

java定時任務(wù)框架elasticjob詳解

更新時間：2019年06月27日 12:05:50 作者：shijian4810

這篇文章主要介紹了java定時任務(wù)框架elasticjob詳解，Elastic-Job是ddframe中dd-job的作業(yè)模塊中分離出來的分布式彈性作業(yè)框架。該項目基于成熟的開源產(chǎn)品Quartz和Zookeeper及其客戶端Curator進(jìn)行二次開發(fā)。,需要的朋友可以參考下

前言

Elastic-Job是ddframe中dd-job的作業(yè)模塊中分離出來的分布式彈性作業(yè)框架。去掉了和dd-job中的監(jiān)控和ddframe接入規(guī)范部分。該項目基于成熟的開源產(chǎn)品Quartz和Zookeeper及其客戶端Curator進(jìn)行二次開發(fā)。 ddframe其他模塊也有可獨(dú)立開源的部分，之前當(dāng)當(dāng)曾開源過dd-soa的基石模塊DubboX。項目開源地址：https://github.com/dangdangdotcom/elastic-job
Elastic-Job是ddframe中dd-job的作業(yè)模塊中分離出來的分布式彈性作業(yè)框架。去掉了和dd-job中的監(jiān)控和ddframe接入規(guī)范部分。該項目基于成熟的開源產(chǎn)品Quartz和Zookeeper及其客戶端Curator進(jìn)行二次開發(fā)。

項目開源地址

ddframe其他模塊也有可獨(dú)立開源的部分，之前當(dāng)當(dāng)曾開源過dd-soa的基石模塊DubboX。

elastic-job和ddframe關(guān)系見下圖

Elastic-Job主要功能

定時任務(wù)：基于成熟的定時任務(wù)作業(yè)框架Quartz cron表達(dá)式執(zhí)行定時任務(wù)。
作業(yè)注冊中心：基于Zookeeper和其客戶端Curator實現(xiàn)的全局作業(yè)注冊控制中心。用于注冊，控制和協(xié)調(diào)分布式作業(yè)執(zhí)行。
作業(yè)分片：將一個任務(wù)分片成為多個小任務(wù)項在多服務(wù)器上同時執(zhí)行。
彈性擴(kuò)容縮容：運(yùn)行中的作業(yè)服務(wù)器崩潰，或新增加n臺作業(yè)服務(wù)器，作業(yè)框架將在下次作業(yè)執(zhí)行前重新分片，不影響當(dāng)前作業(yè)執(zhí)行。
支持多種作業(yè)執(zhí)行模式：支持OneOff，Perpetual和SequencePerpetual三種作業(yè)模式。
失效轉(zhuǎn)移：運(yùn)行中的作業(yè)服務(wù)器崩潰不會導(dǎo)致重新分片，只會在下次作業(yè)啟動時分片。啟用失效轉(zhuǎn)移功能可以在本次作業(yè)執(zhí)行過程中，監(jiān)測其他作業(yè)服務(wù)器空閑，抓取未完成的孤兒分片項執(zhí)行。
運(yùn)行時狀態(tài)收集：監(jiān)控作業(yè)運(yùn)行時狀態(tài)，統(tǒng)計最近一段時間處理的數(shù)據(jù)成功和失敗數(shù)量，記錄作業(yè)上次運(yùn)行開始時間，結(jié)束時間和下次運(yùn)行時間。
作業(yè)停止，恢復(fù)和禁用：用于操作作業(yè)啟停，并可以禁止某作業(yè)運(yùn)行（上線時常用）。
被錯過執(zhí)行的作業(yè)重觸發(fā)：自動記錄錯過執(zhí)行的作業(yè)，并在上次作業(yè)完成后自動觸發(fā)。可參考Quartz的misfire。
多線程快速處理數(shù)據(jù)：使用多線程處理抓取到的數(shù)據(jù)，提升吞吐量。
冪等性：重復(fù)作業(yè)任務(wù)項判定，不重復(fù)執(zhí)行已運(yùn)行的作業(yè)任務(wù)項。由于開啟冪等性需要監(jiān)聽作業(yè)運(yùn)行狀態(tài)，對瞬時反復(fù)運(yùn)行的作業(yè)對性能有較大影響。
容錯處理：作業(yè)服務(wù)器與Zookeeper服務(wù)器通信失敗則立即停止作業(yè)運(yùn)行，防止作業(yè)注冊中心將失效的分片分項配給其他作業(yè)服務(wù)器，而當(dāng)前作業(yè)服務(wù)器仍在執(zhí)行任務(wù)，導(dǎo)致重復(fù)執(zhí)行。
Spring支持：支持spring容器，自定義命名空間，支持占位符。
運(yùn)維平臺：提供運(yùn)維界面，可以管理作業(yè)和注冊中心。

目錄結(jié)構(gòu)說明

elastic-job-core

elastic-job核心模塊，只通過Quartz和Curator就可執(zhí)行分布式作業(yè)。

elastic-job-spring

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

elastic-job-console

elastic-job web控制臺，可將編譯之后的war放入tomcat等servlet容器中使用。

elastic-job-example

使用例子。

elastic-job-test

測試elastic-job使用的公用類，使用方無需關(guān)注。

引入maven依賴

elastic-job已經(jīng)發(fā)布到中央倉庫，可以在pom.xml文件中直接引入maven坐標(biāo)。

<!-- 引入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>

代碼開發(fā)

提供3種作業(yè)類型，分別是OneOff, Perpetual和SequencePerpetual。需要繼承相應(yīng)的抽象類。

方法參數(shù)shardingContext包含作業(yè)配置，分片和運(yùn)行時信息?？赏ㄟ^getShardingTotalCount(),getShardingItems()等方法分別獲取分片總數(shù)，運(yùn)行在本作業(yè)服務(wù)器的分片序列號集合等。

OneOff類型作業(yè)

OneOff作業(yè)類型比較簡單，需要繼承AbstractOneOffElasticJob，該類只提供了一個方法用于覆蓋，此方法將被定時執(zhí)行。用于執(zhí)行普通的定時任務(wù)，與Quartz原生接口相似，只是增加了彈性擴(kuò)縮容和分片等功能。

public class MyElasticJob extends AbstractOneOffElasticJob {
@Override
protected void process(JobExecutionMultipleShardingContext context) {
// do something by sharding items
}
}

Perpetual類型作業(yè)

Perpetual作業(yè)類型略為復(fù)雜，需要繼承AbstractPerpetualElasticJob并可以指定返回值泛型，該類提供兩個方法可覆蓋，分別用于抓取和處理數(shù)據(jù)。可以獲取數(shù)據(jù)處理成功失敗次數(shù)等輔助監(jiān)控信息。

需要注意fetchData方法的返回值只有為null或長度為空時，作業(yè)才會停止執(zhí)行，否則作業(yè)會一直運(yùn)行下去。這點是參照TbSchedule的設(shè)計。Perpetual作業(yè)類型更適用于流式不間歇的數(shù)據(jù)處理。

作業(yè)執(zhí)行時會將fetchData的數(shù)據(jù)傳遞給processData處理，其中processData得到的數(shù)據(jù)是通過多線程（線程池大小可配）拆分的。

建議processData處理數(shù)據(jù)后，更新其狀態(tài)，避免fetchData再次抓取到，從而使得作業(yè)永遠(yuǎn)不會停止。processData的返回值用于表示數(shù)據(jù)是否處理成功，拋出異?；蛘叻祷豧alse將會在統(tǒng)計信息中歸入失敗次數(shù)，返回true則歸入成功次數(shù)。

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類型作業(yè)

SequencePerpetual作業(yè)類型和Perpetual作業(yè)類型極為相似，所不同的是Perpetual作業(yè)類型可以將獲取到的數(shù)據(jù)多線程處理，但不會保證多線程處理數(shù)據(jù)的順序。

如：從2個分片共獲取到100條數(shù)據(jù)，第1個分片40條，第2個分片60條，配置為兩個線程處理，則第1個線程處理前50條數(shù)據(jù)，第2個線程處理后50條數(shù)據(jù)，無視分片項；SequencePerpetual類型作業(yè)則根據(jù)當(dāng)前服務(wù)器所分配的分片項數(shù)量進(jìn)行多線程處理，每個分片項使用同一線程處理，防止了同一分片的數(shù)據(jù)被多線程處理，從而導(dǎo)致的順序問題。

如：從2個分片共獲取到100條數(shù)據(jù)，第1個分片40條，第2個分片60條，則系統(tǒng)自動分配兩個線程處理，第1個線程處理第1個分片的40條數(shù)據(jù)，第2個線程處理第2個分片的60條數(shù)據(jù)。由于Perpetual作業(yè)可以使用多余分片項的任意線程數(shù)處理，所以性能調(diào)優(yōu)的可能會優(yōu)于SequencePerpetual作業(yè)。

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;
}
}

作業(yè)配置

與Spring容器配合使用作業(yè)，可以將作業(yè)Bean配置為Spring Bean，可在作業(yè)中通過依賴注入使用Spring容器管理的數(shù)據(jù)源等對象?？捎胮laceholder占位符從屬性文件中取值。

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
">
<!--配置作業(yè)注冊中心 -->
<reg:zookeeper id="regCenter" serverLists=" yourhost:2181" namespace="dd-job" baseSleepTimeMilliseconds="1000" maxSleepTimeMilliseconds="3000" maxRetries="3" />
<!-- 配置作業(yè)A-->
<job:bean id="oneOffElasticJob" class="xxx.MyOneOffElasticJob" regCenter="regCenter" cron="0/10 * * * * ?" shardingTotalCount="3" shardingItemParameters="0=A,1=B,2=C" />
<!-- 配置作業(yè)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 />命名空間屬性詳細(xì)說明

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

基于Spring但不使用命名空間

<!-- 配置作業(yè)注冊中心 -->
  <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>  <!-- 配置作業(yè)-->
  <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框架，可以用如下方式啟動作業(yè)。

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);
// 定義作業(yè)1配置對象
private JobConfiguration jobConfig1 = new JobConfiguration("oneOffElasticDemoJob", OneOffElasticDemoJob.class, 10, "0/5 * * * * ?");
// 定義作業(yè)2配置對象
private JobConfiguration jobConfig2 = new JobConfiguration("perpetualElasticDemoJob", PerpetualElasticDemoJob.class, 10, "0/5 * * * * ?");
// 定義作業(yè)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();
// 啟動作業(yè)1
new JobController(regCenter, jobConfig1).init();
// 啟動作業(yè)2
new JobController(regCenter, jobConfig2).init();
// 啟動作業(yè)3
new JobController(regCenter, jobConfig3).init();
}
}

使用限制

作業(yè)一旦啟動成功后不能修改作業(yè)名稱，如果修改名稱則視為新的作業(yè)。
同一臺作業(yè)服務(wù)器只能運(yùn)行一個相同的作業(yè)實例，因為作業(yè)運(yùn)行時是按照IP注冊和管理的。
作業(yè)根據(jù)/etc/hosts文件獲取IP地址，如果獲取的IP地址是127.0.0.1而非真實IP地址，應(yīng)正確配置此文件。
一旦有服務(wù)器波動，或者修改分片項，將會觸發(fā)重新分片；觸發(fā)重新分片將會導(dǎo)致運(yùn)行中的Perpetual以及SequencePerpetual作業(yè)再執(zhí)行完本次作業(yè)后不再繼續(xù)執(zhí)行，等待分片結(jié)束后再恢復(fù)正常。
開啟monitorExecution才能實現(xiàn)分布式作業(yè)冪等性（即不會在多個作業(yè)服務(wù)器運(yùn)行同一個分片）的功能，但monitorExecution對短時間內(nèi)執(zhí)行的作業(yè)（如每5秒一觸發(fā)）性能影響較大，建議關(guān)閉并自行實現(xiàn)冪等性。
elastic-job沒有自動刪除作業(yè)服務(wù)器的功能，因為無法區(qū)分是服務(wù)器崩潰還是正常下線。所以如果要下線服務(wù)器，需要手工刪除zookeeper中相關(guān)的服務(wù)器節(jié)點。由于直接刪除服務(wù)器節(jié)點風(fēng)險較大，暫時不考慮在運(yùn)維平臺增加此功能

實現(xiàn)原理

彈性分布式實現(xiàn)

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

某作業(yè)服務(wù)器上線時會自動將服務(wù)器信息注冊到注冊中心，下線時會自動更新服務(wù)器狀態(tài)。

主節(jié)點選舉，服務(wù)器上下線，分片總數(shù)變更均更新重新分片標(biāo)記。

定時任務(wù)觸發(fā)時，如需重新分片，則通過主服務(wù)器分片，分片過程中阻塞，分片結(jié)束后才可執(zhí)行任務(wù)。如分片過程中主服務(wù)器下線，則先選舉主服務(wù)器，再分片。

通過4可知，為了維持作業(yè)運(yùn)行時的穩(wěn)定性，運(yùn)行過程中只會標(biāo)記分片狀態(tài)，不會重新分片。分片僅可能發(fā)生在下次任務(wù)觸發(fā)前。

每次分片都會按服務(wù)器IP排序，保證分片結(jié)果不會產(chǎn)生較大波動。

實現(xiàn)失效轉(zhuǎn)移功能，在某臺服務(wù)器執(zhí)行完畢后主動抓取未分配的分片，并且在某臺服務(wù)器下線后主動尋找可用的服務(wù)器執(zhí)行任務(wù)。