本篇文章為大家展示了怎么在java中使用elasticjob定時(shí)任務(wù)框架，內(nèi)容簡(jiǎn)明扼要并且容易理解，絕對(duì)能使你眼前一亮，通過(guò)這篇文章的詳細(xì)介紹希望你能有所收獲。

創(chuàng)新互聯(lián)公司堅(jiān)持“要么做到，要么別承諾”的工作理念，服務(wù)領(lǐng)域包括：成都網(wǎng)站制作、成都網(wǎng)站設(shè)計(jì)、外貿(mào)營(yíng)銷(xiāo)網(wǎng)站建設(shè)、企業(yè)官網(wǎng)、英文網(wǎng)站、手機(jī)端網(wǎng)站、網(wǎng)站推廣等服務(wù)，滿(mǎn)足客戶(hù)于互聯(lián)網(wǎng)時(shí)代的新賓網(wǎng)站設(shè)計(jì)、移動(dòng)媒體設(shè)計(jì)的需求，幫助企業(yè)找到有效的互聯(lián)網(wǎng)解決方案。努力成為您成熟可靠的網(wǎng)絡(luò)建設(shè)合作伙伴！

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

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

Elastic-Job主要功能

• 定時(shí)任務(wù)： 基于成熟的定時(shí)任務(wù)作業(yè)框架Quartz cron表達(dá)式執(zhí)行定時(shí)任務(wù)。

• 作業(yè)注冊(cè)中心： 基于Zookeeper和其客戶(hù)端Curator實(shí)現(xiàn)的全局作業(yè)注冊(cè)控制中心。用于注冊(cè)，控制和協(xié)調(diào)分布式作業(yè)執(zhí)行。

• 作業(yè)分片： 將一個(gè)任務(wù)分片成為多個(gè)小任務(wù)項(xiàng)在多服務(wù)器上同時(shí)執(zhí)行。

• 彈性擴(kuò)容縮容： 運(yùn)行中的作業(yè)服務(wù)器崩潰，或新增加n臺(tái)作業(yè)服務(wù)器，作業(yè)框架將在下次作業(yè)執(zhí)行前重新分片，不影響當(dāng)前作業(yè)執(zhí)行。

• 支持多種作業(yè)執(zhí)行模式： 支持OneOff，Perpetual和SequencePerpetual三種作業(yè)模式。

• 失效轉(zhuǎn)移： 運(yùn)行中的作業(yè)服務(wù)器崩潰不會(huì)導(dǎo)致重新分片，只會(huì)在下次作業(yè)啟動(dòng)時(shí)分片。啟用失效轉(zhuǎn)移功能可以在本次作業(yè)執(zhí)行過(guò)程中，監(jiān)測(cè)其他作業(yè)服務(wù)器空閑，抓取未完成的孤兒分片項(xiàng)執(zhí)行。

• 運(yùn)行時(shí)狀態(tài)收集： 監(jiān)控作業(yè)運(yùn)行時(shí)狀態(tài)，統(tǒng)計(jì)最近一段時(shí)間處理的數(shù)據(jù)成功和失敗數(shù)量，記錄作業(yè)上次運(yùn)行開(kāi)始時(shí)間，結(jié)束時(shí)間和下次運(yùn)行時(shí)間。

• 作業(yè)停止，恢復(fù)和禁用：用于操作作業(yè)啟停，并可以禁止某作業(yè)運(yùn)行（上線(xiàn)時(shí)常用）。

• 被錯(cuò)過(guò)執(zhí)行的作業(yè)重觸發(fā)：自動(dòng)記錄錯(cuò)過(guò)執(zhí)行的作業(yè)，并在上次作業(yè)完成后自動(dòng)觸發(fā)?？蓞⒖糛uartz的misfire。

• 多線(xiàn)程快速處理數(shù)據(jù)：使用多線(xiàn)程處理抓取到的數(shù)據(jù)，提升吞吐量。

• 冪等性：重復(fù)作業(yè)任務(wù)項(xiàng)判定，不重復(fù)執(zhí)行已運(yùn)行的作業(yè)任務(wù)項(xiàng)。由于開(kāi)啟冪等性需要監(jiān)聽(tīng)作業(yè)運(yùn)行狀態(tài)，對(duì)瞬時(shí)反復(fù)運(yùn)行的作業(yè)對(duì)性能有較大影響。

• 容錯(cuò)處理：作業(yè)服務(wù)器與Zookeeper服務(wù)器通信失敗則立即停止作業(yè)運(yùn)行，防止作業(yè)注冊(cè)中心將失效的分片分項(xiàng)配給其他作業(yè)服務(wù)器，而當(dāng)前作業(yè)服務(wù)器仍在執(zhí)行任務(wù)，導(dǎo)致重復(fù)執(zhí)行。

• Spring支持：支持spring容器，自定義命名空間，支持占位符。

• 運(yùn)維平臺(tái)：提供運(yùn)維界面，可以管理作業(yè)和注冊(cè)中心。

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

• elastic-job-core

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

• elastic-job-spring

elastic-job對(duì)spring支持的模塊，包括命名空間，依賴(lài)注入，占位符等。

• elastic-job-console

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

• elastic-job-example

使用例子。

• elastic-job-test

測(cè)試elastic-job使用的公用類(lèi)，使用方無(wú)需關(guān)注。

引入maven依賴(lài)

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

com.dangdang
elastic-job-core
1.0.1



com.dangdang
elastic-job-spring
1.0.1

代碼開(kāi)發(fā)

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

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

• OneOff類(lèi)型作業(yè)

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

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

• Perpetual類(lèi)型作業(yè)

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

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

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

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

public class MyElasticJob extends AbstractPerpetualElasticJob {
@Override
protected List fetchData(JobExecutionMultipleShardingContext context) {
List result = // get data from database by sharding items
return result;
}
@Override
protected boolean processData(JobExecutionMultipleShardingContext context, Foo data) {
// process data
return true;
}
}

• SequencePerpetual類(lèi)型作業(yè)

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

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

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

public class MyElasticJob extends AbstractSequencePerpetualElasticJob {
@Override
protected List fetchData(JobExecutionSingleShardingContext context) {
List 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è)中通過(guò)依賴(lài)注入使用Spring容器管理的數(shù)據(jù)源等對(duì)象?？捎胮laceholder占位符從屬性文件中取值。

Spring命名空間配置

命名空間屬性詳細(xì)說(shuō)明

命名空間屬性詳細(xì)說(shuō)明

基于Spring但不使用命名空間

不使用Spring配置

如果不使用Spring框架，可以用如下方式啟動(dòng)作業(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注冊(cè)中心配置對(duì)象
private ZookeeperConfiguration zkConfig = new ZookeeperConfiguration("localhost:2181", "elastic-job-example", 1000, 3000, 3);
// 定義Zookeeper注冊(cè)中心
private CoordinatorRegistryCenter regCenter = new ZookeeperRegistryCenter(zkConfig);
// 定義作業(yè)1配置對(duì)象
private JobConfiguration jobConfig1 = new JobConfiguration("oneOffElasticDemoJob", OneOffElasticDemoJob.class, 10, "0/5 * * * * ?");
// 定義作業(yè)2配置對(duì)象
private JobConfiguration jobConfig2 = new JobConfiguration("perpetualElasticDemoJob", PerpetualElasticDemoJob.class, 10, "0/5 * * * * ?");
// 定義作業(yè)3配置對(duì)象
private JobConfiguration jobConfig3 = new JobConfiguration("sequencePerpetualElasticDemoJob", SequencePerpetualElasticDemoJob.class, 10, "0/5 * * * * ?");
public static void main(final String[] args) {
new JobDemo().init();
}
private void init() {
// 連接注冊(cè)中心
regCenter.init();
// 啟動(dòng)作業(yè)1
new JobController(regCenter, jobConfig1).init();
// 啟動(dòng)作業(yè)2
new JobController(regCenter, jobConfig2).init();
// 啟動(dòng)作業(yè)3
new JobController(regCenter, jobConfig3).init();
}
}

使用限制

• 作業(yè)一旦啟動(dòng)成功后不能修改作業(yè)名稱(chēng)，如果修改名稱(chēng)則視為新的作業(yè)。

• 同一臺(tái)作業(yè)服務(wù)器只能運(yùn)行一個(gè)相同的作業(yè)實(shí)例，因?yàn)樽鳂I(yè)運(yùn)行時(shí)是按照IP注冊(cè)和管理的。

• 作業(yè)根據(jù)/etc/hosts文件獲取IP地址，如果獲取的IP地址是127.0.0.1而非真實(shí)IP地址，應(yīng)正確配置此文件。

• 一旦有服務(wù)器波動(dòng)，或者修改分片項(xiàng)，將會(huì)觸發(fā)重新分片；觸發(fā)重新分片將會(huì)導(dǎo)致運(yùn)行中的Perpetual以及SequencePerpetual作業(yè)再執(zhí)行完本次作業(yè)后不再繼續(xù)執(zhí)行，等待分片結(jié)束后再恢復(fù)正常。

• 開(kāi)啟monitorExecution才能實(shí)現(xiàn)分布式作業(yè)冪等性（即不會(huì)在多個(gè)作業(yè)服務(wù)器運(yùn)行同一個(gè)分片）的功能，但monitorExecution對(duì)短時(shí)間內(nèi)執(zhí)行的作業(yè)（如每5秒一觸發(fā)）性能影響較大，建議關(guān)閉并自行實(shí)現(xiàn)冪等性。

• elastic-job沒(méi)有自動(dòng)刪除作業(yè)服務(wù)器的功能，因?yàn)闊o(wú)法區(qū)分是服務(wù)器崩潰還是正常下線(xiàn)。所以如果要下線(xiàn)服務(wù)器，需要手工刪除zookeeper中相關(guān)的服務(wù)器節(jié)點(diǎn)。由于直接刪除服務(wù)器節(jié)點(diǎn)風(fēng)險(xiǎn)較大，暫時(shí)不考慮在運(yùn)維平臺(tái)增加此功能

實(shí)現(xiàn)原理

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

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

某作業(yè)服務(wù)器上線(xiàn)時(shí)會(huì)自動(dòng)將服務(wù)器信息注冊(cè)到注冊(cè)中心，下線(xiàn)時(shí)會(huì)自動(dòng)更新服務(wù)器狀態(tài)。

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

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

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

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

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

流程圖

作業(yè)啟動(dòng)

作業(yè)執(zhí)行

運(yùn)維平臺(tái)

elastic-job運(yùn)維平臺(tái)以war包形式提供，可自行部署到tomcat或jetty等支持servlet的web容器中。elastic-job-console.war可以通過(guò)編譯源碼或從maven中央倉(cāng)庫(kù)獲取。

登錄

默認(rèn)用戶(hù)名和密碼是root/root，可以通過(guò)修改conf\auth.properties文件修改默認(rèn)登錄用戶(hù)名和密碼。

主要功能

• 登錄安全控制

• 注冊(cè)中心管理

• 作業(yè)維度狀態(tài)查看

• 服務(wù)器維度狀態(tài)查看

• 快捷修改作業(yè)設(shè)置

• 控制作業(yè)暫停和恢復(fù)運(yùn)行

設(shè)計(jì)理念

運(yùn)維平臺(tái)和elastic-job并無(wú)直接關(guān)系，是通過(guò)讀取作業(yè)注冊(cè)中心數(shù)據(jù)展現(xiàn)作業(yè)狀態(tài)，或更新注冊(cè)中心數(shù)據(jù)修改全局配置。

控制臺(tái)只能控制作業(yè)本身是否運(yùn)行，但不能控制作業(yè)進(jìn)程的啟停，因?yàn)榭刂婆_(tái)和作業(yè)本身服務(wù)器是完全分布式的，控制臺(tái)并不能控制作業(yè)服務(wù)器。

不支持項(xiàng)

添加作業(yè)。因?yàn)樽鳂I(yè)都是在首次運(yùn)行時(shí)自動(dòng)添加，使用運(yùn)維平臺(tái)添加作業(yè)并無(wú)必要。

停止作業(yè)。即使刪除了Zookeeper信息也不能真正停止作業(yè)的運(yùn)行，還會(huì)導(dǎo)致運(yùn)行中的作業(yè)出問(wèn)題。

刪除作業(yè)服務(wù)器。由于直接刪除服務(wù)器節(jié)點(diǎn)風(fēng)險(xiǎn)較大，暫時(shí)不考慮在運(yùn)維平臺(tái)增加此功能。

主要界面

總覽頁(yè)

注冊(cè)中心管理頁(yè)

作業(yè)詳細(xì)信息頁(yè)

服務(wù)區(qū)詳細(xì)信息頁(yè)

上述內(nèi)容就是怎么在java中使用elasticjob定時(shí)任務(wù)框架，你們學(xué)到知識(shí)或技能了嗎？如果還想學(xué)到更多技能或者豐富自己的知識(shí)儲(chǔ)備，歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道。