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
|
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
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
|
<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 配置中心的相关配置信息
|
1
2
3
4
5
|
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中情况。
|
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
|
<?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
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
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
|
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
|
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
|
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
|
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!"
;
}
}
|
本文转自布拉君君 51CTO博客，原文链接：http://blog.51cto.com/5148737/1972498，如需转载请自行联系原作者
