Elastic job是当当网架构师张亮,曹昊和江树建基于Zookepper、Quartz开发并开源的一个Java分布式定时任务,解决了Quartz不支持分布式的弊端。Elastic job主要的功能有支持弹性扩容,通过Zookepper集中管理和监控job,支持失效转移等,这些都是Quartz等其他定时任务无法比拟的。
目前Elastic job的最新版本已经由原来的elastic-job-core分离除了两个项目,分别为Elastic-Job-Lite和Elastic-Job-Cloud。Elastic-Job是一个分布式调度解决方案,由两个相互独立的子项目Elastic-Job-Lite和Elastic-Job-Cloud组成,Elastic-Job-Lite定位为轻量级无中心化解决方案,使用jar包的形式提供分布式任务的协调服务。 Elastic-Job-Cloud使用Mesos + Docker(TBD)的解决方案,额外提供资源治理、应用分发以及进程隔离等服务,Elastic-Job-Lite和Elastic-Job-Cloud提供同一套API开发作业,开发者仅需一次开发,即可根据需要以Lite或Cloud的方式部署
开头copy 网上一段术语,讲解的很清晰。总结就是:elastic job 是用来实现 分布式的定时任务,比如说定时任务项目,部署在2台,但是只需要执行一次,而又想保持2台服务器代码都一致,这时候elasitc job 可以很完美的解决问题,当然使用的是zookeeper的相关特性。
1.实战
项目目录如下
1.pom.xml
新建maven项目,引入springboot的相关jar包
再引入elastic job 的相关核心包 elastic-job-core,elastic-job-spring
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.xj</groupId> <artifactId>el</artifactId> <version>0.0.1-SNAPSHOT</version> <packaging>jar</packaging> <name>el</name> <url>http://maven.apache.org</url> <properties> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> </properties> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>1.4.0.RELEASE</version> </parent> <dependencies> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>3.8.1</version> <scope>test</scope> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <!-- 引入elastic-job-core核心模块 --> <dependency> <groupId>com.dangdang</groupId> <artifactId>elastic-job-core</artifactId> <version>1.1.1</version> </dependency> <!-- 使用springframework自定义命名空间时引入 --> <dependency> <groupId>com.dangdang</groupId> <artifactId>elastic-job-spring</artifactId> <version>1.1.1</version> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-maven-plugin</artifactId> </plugin> </plugins> </build> </project>
2.reg.properties
存放zookeeper 配置中心的相关配置信息
serverLists=10.3.142.107:2181,10.3.142.107:2182,106.3.14.107:2183 namespace=elastic-job-example baseSleepTimeMilliseconds=1000 maxSleepTimeMilliseconds=3000 maxRetries=3
3.withNamespace.xml
存放作业的相关详情,这里测试存放了 simple 与dataFlow 2中情况。
<?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:context="http://www.springframework.org/schema/context" 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.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.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 "> <context:property-placeholder location="classpath:conf/*.properties" /> <!--配置作业注册中心 --> <reg:zookeeper id="regCenter" server-lists="${serverLists}" namespace="${namespace}" base-sleep-time-milliseconds="${baseSleepTimeMilliseconds}" max-sleep-time-milliseconds="${baseSleepTimeMilliseconds}" max-retries="${maxRetries}" /> <!-- 配置作业--> <job:simple id="mySimpleJob" class="com.xj.el.MySimpleJob" registry-center-ref="regCenter" sharding-total-count="2" cron="0/3 * * * * ?" overwrite="true" /> <job:dataflow id="throughputDataFlow" class="com.xj.el.MyThroughputDataFlowElasticJob" registry-center-ref="regCenter" cron="0/10 * * * * ?" sharding-total-count="3" sharding-item-parameters="0=A,1=B,2=C" process-count-interval-seconds= "10" concurrent-data-process-thread-count="10"/> </beans>
elastic-job提供了三种类型的作业:Simple类型作业、Dataflow类型作业、Script类型作业。这里主要讲解前两者。Script类型作业意为脚本类型作业,支持shell,python,perl等所有类型脚本,使用不多,可以参见github文档。
SimpleJob需要实现SimpleJob接口,意为简单实现,未经过任何封装,与quartz原生接口相似,比如示例代码中所使用的job。
Dataflow类型用于处理数据流,需实现DataflowJob接口。该接口提供2个方法可供覆盖,分别用于抓取(fetchData)和处理(processData)数据。
可通过DataflowJobConfiguration配置是否流式处理。
流式处理数据只有fetchData方法的返回值为null或集合长度为空时,作业才停止抓取,否则作业将一直运行下去; 非流式处理数据则只会在每次作业执行过程中执行一次fetchData方法和processData方法,随即完成本次作业。
实际开发中,Dataflow类型的job还是很有好用的。
4.MySimpleJob.java
package com.xj.el; import com.dangdang.ddframe.job.api.JobExecutionMultipleShardingContext; import com.dangdang.ddframe.job.plugin.job.type.simple.AbstractSimpleElasticJob; public class MySimpleJob extends AbstractSimpleElasticJob{ @Override public void process(JobExecutionMultipleShardingContext shardingContext) { System.out.println(String.format("------Thread ID: %s, 任务总片数: %s, 当前分片项: %s", Thread.currentThread().getId(), shardingContext.getShardingTotalCount(), shardingContext.getShardingItems())); } }
5.MyThroughputDataFlowElasticJob.java
package com.xj.el; import java.sql.Time; import java.text.SimpleDateFormat; import java.util.ArrayList; import java.util.Date; import java.util.List; import com.dangdang.ddframe.job.api.JobExecutionMultipleShardingContext; import com.dangdang.ddframe.job.internal.job.AbstractJobExecutionShardingContext; import com.dangdang.ddframe.job.plugin.job.type.dataflow.AbstractIndividualThroughputDataFlowElasticJob; public class MyThroughputDataFlowElasticJob extends AbstractIndividualThroughputDataFlowElasticJob<TestBean>{ public List<TestBean> fetchData(JobExecutionMultipleShardingContext shardingContext) { List<TestBean> testBeanList = new ArrayList(); //获取testBean的相关数据 return testBeanList; } public boolean processData(JobExecutionMultipleShardingContext shardingContext, TestBean data) { //处理上文testBean的相关数据 return false; } }
6.springboot 测试主类 SpringBootSampleApplication.java
package com.xj.el; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.context.annotation.ImportResource; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; /** * Hello world! * */ @RestController @EnableAutoConfiguration @ImportResource(locations = {"classpath:withNamespace.xml"}) public class SpringBootSampleApplication { public static void main(String[] args) { SpringApplication.run(SpringBootSampleApplication.class, args); } @RequestMapping("/") String home() { return "Hello World!"; } }
原创文章,作者:kepupublish,如若转载,请注明出处:https://blog.ytso.com/tech/opensource/198214.html