java定时任务框架elasticjob详解
這篇文章主要介紹了java定時任務框架elasticjob詳解，Elastic-Job是ddframe中dd-job的作業模塊中分離出來的分布式彈性作業框架。該項目基于成熟的開源產品Quartz和Zookeeper及其客戶端Curator進行二次開發。,需要的朋友可以參考下
前言
Elastic-Job是ddframe中dd-job的作業模塊中分離出來的分布式彈性作業框架。去掉了和dd-job中的監控和ddframe接入規范部分。該項目基于成熟的開源產品Quartz和Zookeeper及其客戶端Curator進行二次開發。 ddframe其他模塊也有可獨立開源的部分，之前當當曾開源過dd-soa的基石模塊DubboX。 項目開源地址：https://github.com/dangdangdotcom/elastic-job
Elastic-Job是ddframe中dd-job的作業模塊中分離出來的分布式彈性作業框架。去掉了和dd-job中的監控和ddframe接入規范部分。該項目基于成熟的開源產品Quartz和Zookeeper及其客戶端Curator進行二次開發。
項目開源地址
ddframe其他模塊也有可獨立開源的部分，之前當當曾開源過dd-soa的基石模塊DubboX。
elastic-job和ddframe關系見下圖
Elastic-Job主要功能
- 定時任務： 基于成熟的定時任務作業框架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坐標。
| 1 2 3 4 5 6 7 8 9 10 11 12 | <!-- 引入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原生接口相似，只是增加了彈性擴縮容和分片等功能。
| 1 2 3 4 5 6 | 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則歸入成功次數。
| 1 2 3 4 5 6 7 8 9 10 11 12 | 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作業。
| 1 2 3 4 5 6 7 8 9 10 11 12 | 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命名空間配置
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 | <?xml version="1.0" encoding="UTF-8"?> <beans xmlns="Index of /schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:reg="對不起，您要訪問的頁面暫時沒有找到。" xmlns: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但不使用命名空間
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 | <!-- 配置作業注冊中心 --> ??<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框架，可以用如下方式啟動作業。
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 | 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排序，保證分片結果不會產生較大波動。
實現失效轉移功能，在某臺服務器執行完畢后主動抓取未分配的分片，并且在某臺服務器下線后主動尋找可用的服務器執行任務。
流程圖
作業啟動
作業執行
運維平臺
elastic-job運維平臺以war包形式提供，可自行部署到tomcat或jetty等支持servlet的web容器中。elastic-job-console.war可以通過編譯源碼或從maven中央倉庫獲取。
登錄
默認用戶名和密碼是root/root，可以通過修改conf\auth.properties文件修改默認登錄用戶名和密碼。
主要功能
- 登錄安全控制
- 注冊中心管理
- 作業維度狀態查看
- 服務器維度狀態查看
- 快捷修改作業設置
- 控制作業暫停和恢復運行
設計理念
運維平臺和elastic-job并無直接關系，是通過讀取作業注冊中心數據展現作業狀態，或更新注冊中心數據修改全局配置。
控制臺只能控制作業本身是否運行，但不能控制作業進程的啟停，因為控制臺和作業本身服務器是完全分布式的，控制臺并不能控制作業服務器。
不支持項
添加作業。因為作業都是在首次運行時自動添加，使用運維平臺添加作業并無必要。
停止作業。即使刪除了Zookeeper信息也不能真正停止作業的運行，還會導致運行中的作業出問題。
刪除作業服務器。由于直接刪除服務器節點風險較大，暫時不考慮在運維平臺增加此功能。
主要界面
總覽頁
注冊中心管理頁
作業詳細信息頁
服務區詳細信息頁
以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持腳本之家。
來源：https://www.jb51.net/article/164097.htm
總結
以上是生活随笔為你收集整理的java定时任务框架elasticjob详解的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： ac130空中炮艇模拟器Ce一01大黄蜂
- 下一篇： 江西单招大专女生可以当兵吗？