elastic-job的原理简介和使用

古城微笑少年丶 2022-05-22 09:51 133阅读 0赞

elastic-job是当当开源的一款非常好用的作业框架，在这之前，我们开发定时任务一般都是使用quartz或者spring-task（ScheduledExecutorService），无论是使用quartz还是spring-task，我们都会至少遇到两个痛点：  
1.不敢轻易跟着应用服务多节点部署，可能会重复多次执行而引发系统逻辑的错误。  
2.quartz的集群仅仅只是用来HA，节点数量的增加并不能给我们的每次执行效率带来提升，即不能实现水平扩展。

本篇博文将会自顶向下地介绍elastic-job，让大家认识了解并且快速搭建起环境。

## elastic-job产品线说明 ##

elastic-job在2.x之后，出了两个产品线：Elastic-Job-Lite和Elastic-Job-Cloud。我们一般使用Elastic-Job-Lite就能够满足需求，本文也是以Elastic-Job-Lite为主。1.x系列对应的就只有Elastic-Job-Lite，并且在2.x里修改了一些核心类名，差别虽大，原理类似，建议使用2.x系列。写此博文，最新release版本为2.0.5。

![20170311022123242][]

## elastic-job-lite原理 ##

举个典型的job场景，比如余额宝里的昨日收益，系统需要job在每天某个时间点开始，给所有余额宝用户计算收益。如果用户数量不多，我们可以轻易使用quartz来完成，我们让计息job在某个时间点开始执行，循环遍历所有用户计算利息，这没问题。可是，如果用户体量特别大，我们可能会面临着在第二天之前处理不完这么多用户。另外，我们部署job的时候也得注意，我们可能会把job直接放在我们的webapp里，webapp通常是多节点部署的，这样，我们的job也就是多节点，多个job同时执行，很容易造成重复执行，比如用户重复计息，为了避免这种情况，我们可能会对job的执行加锁，保证始终只有一个节点能执行，或者干脆让job从webapp里剥离出来，独自部署一个节点。  
elastic-job就可以帮助我们解决上面的问题，elastic底层的任务调度还是使用的quartz，通过zookeeper来动态给job节点分片。  
我们来看：  
很大体量的用户需要在特定的时间段内计息完成  
我们肯定是希望我们的任务可以通过集群达到水平扩展，集群里的每个节点都处理部分用户，不管用户数量有多庞大，我们只要增加机器就可以了，比如单台机器特定时间能处理n个用户，2台机器处理2n个用户，3台3n，4台4n...，再多的用户也不怕了。  
使用elastic-job开发的作业都是zookeeper的客户端，比如我希望3台机器跑job，我们将任务分成3片，框架通过zk的协调，最终会让3台机器分别分配到0,1,2的任务片，比如server0-->0，server1-->1，server2-->2，当server0执行时，可以只查询id%3==0的用户，server1执行时，只查询id%3==1的用户，server2执行时，只查询id%3==2的用户。  
任务部署多节点引发重复执行  
在上面的基础上，我们再增加server3，此时，server3分不到任务分片，因为只有3片，已经分完了。没有分到任务分片的作业程序将不执行。  
如果此时server2挂了，那么server2的分片项会分配给server3，server3有了分片，就会替代server2执行。  
如果此时server3也挂了，只剩下server0和server1了，框架也会自动把server3的分片随机分配给server0或者server1，可能会这样，server0-->0，server1-->1,2。  
这种特性称之为弹性扩容，即elastic-job名称的由来。

## 代码演示 ##

我们搭建环境通过示例代码来演示上面的例子，elastic-job是不支持单机多实例的，通过zk的协调分片是以ip为单元的。很多同学上来可能就是通过单机多实例来学习，结果导致分片和预期不一致。这里没办法，只能通过多机器或者虚拟机，我们这里使用虚拟机，另外，由于资源有限，我们这里仅仅只模拟两台机器。

节点说明：  
本地宿主机器  
zookeeper、job  
192.168.241.1  
  
虚拟机  
job  
192.168.241.128  
  
环境说明：  
Java  
请使用JDK1.7及其以上版本。  
Zookeeper  
请使用Zookeeper3.4.6及其以上版本  
Elastic-Job-Lite  
2.0.5（2.x系列即可，最好是2.0.4及其以上，因为2.0.4版本有本人提交的少许代码，(\*^\_\_^\*) 嘻嘻……）  
  
需求说明：  
通过两台机器演示动态分片

step1. 引入框架的jar包

\[html\]  [view plain][]   [copy][view plain]

1.    
2.  **<****dependency****>**  
3.      **<****groupId****>**com.dangdang**</****groupId****>**  
4.      **<****artifactId****>**elastic-job-lite-core**</****artifactId****>**  
5.      **<****version****>**2.0.5**</****version****>**  
6.  **</****dependency****>**  
7.    
8.  **<****dependency****>**  
9.      **<****groupId****>**com.dangdang**</****groupId****>**  
10.     **<****artifactId****>**elastic-job-lite-spring**</****artifactId****>**  
11.     **<****version****>**2.0.5**</****version****>**  
12. **</****dependency****>**

step2. 编写job

\[java\]  [view plain][]   [copy][view plain]

1.  **package** com.fanfan.sample001;  
2.    
3.  **import** com.dangdang.ddframe.job.api.ShardingContext;  
4.  **import** com.dangdang.ddframe.job.api.simple.SimpleJob;  
5.    
6.  **import** java.util.Date;  
7.    
8.  /\*\* 
9.   \* Created by fanfan on 2016/12/20. 
10.  \*/  
11. **public** **class** MySimpleJob **implements** SimpleJob \{  
12.     @Override  
13.     **public** **void** execute(ShardingContext shardingContext) \{  
14.         System.out.println(String.format("------Thread ID: %s, 任务总片数: %s, 当前分片项: %s",  
15.                 Thread.currentThread().getId(), shardingContext.getShardingTotalCount(), shardingContext.getShardingItem()));  
16.         /\*\* 
17.          \* 实际开发中，有了任务总片数和当前分片项，就可以对任务进行分片执行了 
18.          \* 比如 SELECT \* FROM user WHERE status = 0 AND MOD(id, shardingTotalCount) = shardingItem 
19.          \*/  
20.     \}  
21. \}

Step3. Spring配置

\[html\]  [view plain][]   [copy][view plain]

1.  **<?****xml** version="1.0" encoding="UTF-8"**?>**  
2.  **<****beans** xmlns="http://www.springframework.org/schema/beans"  
3.         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"  
4.         xmlns:reg="http://www.dangdang.com/schema/ddframe/reg"  
5.         xmlns:job="http://www.dangdang.com/schema/ddframe/job"  
6.         xsi:schemaLocation="http://www.springframework.org/schema/beans  
7.                          http://www.springframework.org/schema/beans/spring-beans.xsd  
8.                          http://www.dangdang.com/schema/ddframe/reg  
9.                          http://www.dangdang.com/schema/ddframe/reg/reg.xsd  
10.                         http://www.dangdang.com/schema/ddframe/job  
11.                         http://www.dangdang.com/schema/ddframe/job/job.xsd"**>**  
12.       
13.     **<****reg:zookeeper** id="regCenter" server-lists="192.168.241.1:2181" namespace="dd-job"  
14.                    base-sleep-time-milliseconds="1000" max-sleep-time-milliseconds="3000" max-retries="3" **/>**  
15.   
16.       
17.     **<****job:simple** id="mySimpleJob" class="com.fanfan.sample001.MySimpleJob" registry-center-ref="regCenter"  
18.                 sharding-total-count="2" cron="0/2 \* \* \* \* ?" overwrite="true" **/>**  
19.   
20. **</****beans****>**

Case1. 单节点

![20170311024524057][]

![Center][]

![Center 1][]

Case2. 增加一个节点

![20170311024527737][]

![Center 2][]

![Center 3][]

![Center 4][]

![Center 5][]

Case3. 断开一个节点

![Center 6][]

![Center 7][]

![Center 8][]

## 作业类型 ##

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还是很有好用的。

比如拿余额宝计息来说：

\[java\]  [view plain][]   [copy][view plain]

1.  **package** com.fanfan.sample001;  
2.    
3.  **import** com.dangdang.ddframe.job.api.ShardingContext;  
4.  **import** com.dangdang.ddframe.job.api.dataflow.DataflowJob;  
5.    
6.  **import** java.util.ArrayList;  
7.  **import** java.util.List;  
8.    
9.  /\*\* 
10.  \* Created by fanfan on 2016/12/23. 
11.  \*/  
12. **public** **class** MyDataFlowJob **implements** DataflowJob<User> \{  
13.   
14.     /\* 
15.         status 
16.         0：待处理 
17.         1：已处理 
18.      \*/  
19.   
20.     @Override  
21.     **public** List<User> fetchData(ShardingContext shardingContext) \{  
22.         List<User> users = **null**;  
23.         /\*\* 
24.          \* users = SELECT \* FROM user WHERE status = 0 AND MOD(id, shardingTotalCount) = shardingItem Limit 0, 30 
25.          \*/  
26.         **return** users;  
27.     \}  
28.   
29.     @Override  
30.     **public** **void** processData(ShardingContext shardingContext, List<User> data) \{  
31.         **for** (User user: data) \{  
32.             System.out.println(String.format("用户 %s 开始计息", user.getUserId()));  
33.             user.setStatus(1);  
34.             /\*\* 
35.              \* update user 
36.              \*/  
37.         \}  
38.     \}  
39. \}

\[html\]  [view plain][]   [copy][view plain]

1.  **<****job:dataflow** id="myDataFlowJob" class="com.fanfan.sample001.MyDataFlowJob" registry-center-ref="regCenter"  
2.                sharding-total-count="2" cron="0 0 02 \* \* ?" streaming-process="true" overwrite="true" **/>**

## 其它功能 ##

上述介绍的是最精简常用的功能。elastic-job的功能集还不止这些，比如像作业事件追踪、任务监听等，另外，elastic-job-lite-console作为一个独立的运维平台还提供了用来查询和操作任务的web页面。  
这些增强的功能读者可以在github/elastic-job上自行学习，相信有了本篇博文的基础，再阅读那些文档就特别简单了。

[20170311022123242]: /images/20220522/9ae583da015546cfb398a933134d9aee.png
[view plain]: https://blog.csdn.net/fanfan_v5/article/details/61310045#
[20170311024524057]: /images/20220522/d14cf857ab09412388e1b2212d79f2b0.png
[Center]: /images/20220522/2a292b8255a140ddab994e9614f69959.png
[Center 1]: /images/20220522/f558dac09938489a97cf018df162cbc3.png
[20170311024527737]: /images/20220522/fb4393610763421d9279dc04516ed071.png
[Center 2]: /images/20220522/f683d972247942ed838bd703a1b7a738.png
[Center 3]: /images/20220522/7c7ae00084f048e18d9eb36a33ca78c2.png
[Center 4]: /images/20220522/251ee2dd096141d6a63838332b4af764.png
[Center 5]: /images/20220522/17719a5a03c34c16bad5ac1fa27d441f.png
[Center 6]: /images/20220522/647ae119ff384a49b27c19184c010605.png
[Center 7]: /images/20220522/24e7bdd4842244218185664c339493e7.png
[Center 8]: /images/20220522/f7e0330f7f9e449985abe21ba3563c8c.png