Elastic-job解決的問題

? ? ? ? Elastic-job解決以下問題，我們開發定時任務一般都是使用quartz或者spring-task（ScheduledExecutorService），無論是使用quartz還是spring-task，我們都會至少遇到兩個痛點：

不敢輕易跟著應用服務多節點部署，可能會重復多次執行而引發系統邏輯的錯誤。

quartz的集群僅僅只是用來HA，節點數量的增加并不能給我們的每次執行效率帶來提升，即不能實現水平擴展。

Elastic-job的主要包括以下功能。

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

作業注冊中心：基于Zookeeper和其客戶端Curator實現的全局作業注冊控制中心。用于注冊，控制和協調分布式作業執行。

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

彈性擴容縮容：運行中的作業服務器崩潰，或新增加n臺作業服務器，作業框架將在下次作業執行前重新分片，不影響當前作業執行。

支持多種作業執行模式：支持OneOff（類似Quartz原生作業），Perpetual（類似TbSchedule作業）和SequencePerpetual（類似TbSchedule和kafka的合并，但處理時維持分片順序）三種作業模式。

失效轉移：運行中的作業服務器崩潰不會導致重新分片，只會在下次作業啟動時分片。啟用失效轉移功能可以在本次作業執行過程中，監測其他作業服務器空閑，抓取未完成的孤兒分片項執行。

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

作業停止，恢復和禁用：用于操作作業啟停，并可以禁止某作業運行（上線時常用）。

被錯過執行的作業重觸發：自動記錄錯過執行的作業，并在上次作業完成后自動觸發。可參考Quartz的misfire。

多線程快速處理數據：使用多線程處理抓取到的數據，提升吞吐量。

冪等性：重復作業任務項判定，不重復執行已運行的作業任務項。由于開啟冪等性需要監聽作業運行狀態，對瞬時反復運行的作業對性能有較大影響。

容錯處理：作業服務器與Zookeeper服務器通信失敗則立即停止作業運行，防止作業注冊中心將失效的分片分項配給其他作業服務器，而當前作業服務器仍在執行任務，導致重復執行。

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

運維平臺：提供運維界面，可以管理作業和注冊中心。

運用實例?

? ? ?1.引入依賴?? (基于gradle)

compile('com.dangdang:elastic-job-core:1.0.6') compile('com.dangdang:elastic-job-spring:1.0.6')

? ? ?2.xml文件的配置

<?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/beanshttp://www.springframework.org/schema/beans/spring-beans.xsdhttp://www.dangdang.com/schema/ddframe/reghttp://www.dangdang.com/schema/ddframe/reg/reg.xsdhttp://www.dangdang.com/schema/ddframe/jobhttp://www.dangdang.com/schema/ddframe/job/job.xsd"><!--配置作業注冊中心 serverLists:注冊中心地址(zookeeper或者其他的)maxRetries:最大重試次數--><reg:zookeeper id="regCenter" serverLists="${acw.zookeeper.address}" namespace="duiba-job"baseSleepTimeMilliseconds="1000" maxSleepTimeMilliseconds="3000" maxRetries="3" overwrite="true"/><!-- 配置定時任務shardingTotalCount:分幾個片段執行--><job:bean id="aMonthlyScheduleTaskJob" class="cn.com.duiba.manager.web.biz.job.impl.AMonthlyScheduleTaskJob"regCenter="regCenter" cron="0 0 1 1 * ?" shardingTotalCount="1" overwrite="true"/>

? ? 3.JAVA代碼

import cn.com.duiba.manager.web.common.tools.Environment; import com.dangdang.ddframe.job.api.JobExecutionMultipleShardingContext; import com.dangdang.ddframe.job.plugin.job.type.simple.AbstractSimpleElasticJob; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Value;public abstract class AbstractManagerJob extends AbstractSimpleElasticJob {private static final Logger LOG = LoggerFactory.getLogger(AbstractManagerJob.class);@Value("${duiba.job.skip:false}")private boolean skipJob;public void process(final JobExecutionMultipleShardingContext shardingContext) {if (Environment.isPrepub() || skipJob) {return;}try {doProcess(shardingContext);} catch (Exception e) {LOG.error("", e);}}public abstract void doProcess(final JobExecutionMultipleShardingContext shardingContext); } import cn.com.duiba.manager.web.biz.job.AbstractManagerJob; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component;import cn.com.duiba.developer.center.api.remoteservice.RemoteAppService;import com.dangdang.ddframe.job.api.JobExecutionMultipleShardingContext;@Component public class AMonthlyScheduleTaskJob extends AbstractManagerJob {@Autowiredprivate RemoteAppService remoteAppServiceNew;@Overridepublic void doProcess(JobExecutionMultipleShardingContext shardingContext) {aMonthlySchedule();}private void aMonthlySchedule() {remoteAppServiceNew.updatBudgetRemain();} }

?

總結

以上是生活随笔為你收集整理的当当(Elastic-job)分布式定时任务的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。