分布式调度框架Elastic-Job

1、Elastic-Job介绍

Elastic-Job是当当网开源的⼀个分布式调度解决方案，基于Quartz二次开发的，由两个相互独立的子项目Elastic-Job-Lite和Elastic-Job-Cloud组成。我们要学习的是 Elastic-Job-Lite，它定位为轻量级无中心化解决方案，使用Jar包的形式提供分布式任务的协调服务，而Elastic-Job-Cloud子项目需要结合Mesos以及Docker在云环境下使用。

Elastic-Job的github地址：https://github.com/elasticjob

主要功能介绍

分布式调度协调，在分布式环境中，任务能够按指定的调度策略执行，并且能够避免同⼀任务多实例重复执行
丰富的调度策略基于成熟的定时任务作业框架Quartz cron表达式执行定时任务
弹性扩容缩容当集群中增加某⼀个实例，它应当也能够被选举并执行任务；当集群减少⼀个实例时，它所执行的任务能被转移到别的实例来执行。
失效转移某实例在任务执行失败后，会被转移到其他实例执行
错过执行作业重触发若因某种原因导致作业错过执行，自动记录错过执行的作业，并在上次作业完成后自动触发。
支持并行调度、支持任务分片，任务分片是指将一个任务分为多个小任务项在多个实例同时执行。
作业分片一致性，当任务被分片后，保证同一分片在分布式环境中仅一个执行实例。

2、Elastic-Job-Lite应用

jar包（API） + 安装zk软件
Elastic-Job依赖于Zookeeper进行分布式协调，所以需要安装Zookeeper软件（3.4.6版本以上），Zookeeper的本质功能：存储+通知。

安装Zookeeper（此处单例配置）

1）我们使用3.4.10版本，在linux平台解压下载的zookeeper-3.4.10.tar.gz
2）进⼊conf目录，cp zoo_sample.cfg zoo.cfg
3) 进⼊bin目录，启动zk服务
启动 ./zkServer.sh start
停止 ./zkServer.sh stop
查看状态 ./zkServer.sh status

3、Zookeeper的树形节点结构图

引入Jar包

<!-- https://mvnrepository.com/artifact/com.dangdang/elastic-job-lite-core-->
<dependency><groupId>com.dangdang</groupId><artifactId>elastic-job-lite-core</artifactId><version>2.1.5</version>
</dependency>

定时任务实例

需求：每隔两秒钟执行一次定时任务（resume表中未归档的数据归档到resume_bak表中，
每次归档1条记录）
1）resume_bak和resume表结构完全⼀样
2）resume表中数据归档之后不删除，只将state置为"已归档"
数据表结构

-- ----------------------------
-- Table structure for resume
-- ----------------------------
DROP TABLE
IFEXISTS `resume`;
CREATE TABLE `resume` (`id` BIGINT ( 20 ) NOT NULL AUTO_INCREMENT,`name` VARCHAR ( 255 ) DEFAULT NULL,`sex` VARCHAR ( 255 ) DEFAULT NULL,`phone` VARCHAR ( 255 ) DEFAULT NULL,`address` VARCHAR ( 255 ) DEFAULT NULL,`education` VARCHAR ( 255 ) DEFAULT NULL,`state` VARCHAR ( 255 ) DEFAULT NULL,PRIMARY KEY ( `id` )
) ENGINE = INNODB AUTO_INCREMENT = 1001 DEFAULT CHARSET = utf8;SET FOREIGN_KEY_CHECKS = 1;

程序开发

定时任务类

import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import elasticJob.util.JdbcUtil;import java.util.List;
import java.util.Map;/*** ElasticJobLite定时任务逻辑处理类*/
public class ArchJob implements SimpleJob {/*** execute方法中书写我们的业务逻辑（execute方法每次定时任务都会执行一次）** @param shardingContext*/@Overridepublic void execute(ShardingContext shardingContext) {// 从resume表中查询出一条记录（未归档）String selectSql = "select * from resume where state='未归档' limit 1";List<Map<String, Object>> list = JdbcUtil.executeQuery(selectSql);if (list == null || list.size() == 0) {System.out.println("数据已经处理完毕");return;}// “未归档”更改为“已归档”Map<String, Object> stringObjectMap = list.get(0);Long id = (Long) stringObjectMap.get("id");String name = (String) stringObjectMap.get("name");String education = (String) stringObjectMap.get("education");System.out.println("==============id:" + id + "name:" + name + "education:" + education);String updateSql = "update resume set state='已归档' where id=?";JdbcUtil.executeUpdate(updateSql, id);// 归档这条记录，把这条记录插入到resume_bak表中String insertSql = "insert into resume_bak select * from resume where id=?";JdbcUtil.executeUpdate(insertSql, id);}
}

主类

import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.lite.api.JobScheduler;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperConfiguration;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;public class ElasticJobMain {public static void main(String[] args) {// 配置分布式协调服务（注册中心）ZookeeperZookeeperConfiguration zookeeperConfiguration = new ZookeeperConfiguration("8.142.8.105:2181", "data-archive-job");CoordinatorRegistryCenter coordinatorRegistryCenter = new ZookeeperRegistryCenter(zookeeperConfiguration);coordinatorRegistryCenter.init();// 配置任务（时间事件、定时任务作业逻辑、调度器）JobCoreConfiguration jobCoreConfiguration = JobCoreConfiguration.newBuilder("archive-job", "*/2 * * * * ?", 1).build();SimpleJobConfiguration simpleJobConfiguration = new SimpleJobConfiguration(jobCoreConfiguration, ArchJob.class.getName());JobScheduler jobScheduler = new JobScheduler(coordinatorRegistryCenter, LiteJobConfiguration.newBuilder(simpleJobConfiguration).overwrite(true).build());jobScheduler.init();}
}

JdbcUtil工具类

import java.sql.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;public class JdbcUtil {// urlprivate static String url = "jdbc:mysql://127.0.0.1:3306/Job?characterEncoding=utf8&useSSL=false";// userprivate static String user = "root";// passwordprivate static String password = "root";// 驱动程序类private static String driver = "com.mysql.jdbc.Driver";static {try {Class.forName(driver);} catch (ClassNotFoundException e) {
// TODO Auto-generated catch blocke.printStackTrace();}}public static Connection getConnection() {try {return DriverManager.getConnection(url, user, password);} catch (SQLException e) {
// TODO Auto-generated catch blocke.printStackTrace();}return null;}public static void close(ResultSet rs, PreparedStatement ps, Connection con) {if (rs != null) {try {rs.close();} catch (SQLException e) {
// TODO Auto-generated catch blocke.printStackTrace();} finally {if (ps != null) {try {ps.close();} catch (SQLException e) {
// TODO Auto-generated catch blocke.printStackTrace();} finally {if (con != null) {try {con.close();} catch (SQLException e) {
// TODO Auto-generated catch blocke.printStackTrace();}}}}}}}public static void executeUpdate(String sql, Object... obj) {Connection con = getConnection();PreparedStatement ps = null;try {ps = con.prepareStatement(sql);for (int i = 0; i < obj.length; i++) {ps.setObject(i + 1, obj[i]);}ps.executeUpdate();} catch (SQLException e) {
// TODO Auto-generated catch blocke.printStackTrace();} finally {close(null, ps, con);}}public static List<Map<String, Object>> executeQuery(String sql, Object... obj) {Connection con = getConnection();ResultSet rs = null;PreparedStatement ps = null;try {ps = con.prepareStatement(sql);for (int i = 0; i < obj.length; i++) {ps.setObject(i + 1, obj[i]);}rs = ps.executeQuery();List<Map<String, Object>> list = new ArrayList<>();int count = rs.getMetaData().getColumnCount();while (rs.next()) {Map<String, Object> map = new HashMap<String, Object>();for (int i = 0; i < count; i++) {Object ob = rs.getObject(i + 1);String key = rs.getMetaData().getColumnName(i + 1);map.put(key, ob);}list.add(map);}return list;} catch (SQLException e) {
// TODO Auto-generated catch blocke.printStackTrace();} finally {close(rs, ps, con);}return null;}
}

测试

可先启动⼀个进程，然后再启动⼀个进程（两个进程模拟分布式环境下，通⼀个定时任务部署了两份在工作）
两个进程逐个启动，观察现象
关闭其中执行的进程，观察现象

Leader节点选举机制

每个Elastic-Job的任务执行实例App作为Zookeeper的客户端来操作ZooKeeper的znode

多个实例同时创建/leader节点
/leader节点只能创建⼀个，后创建的会失败，创建成功的实例会被选为leader节点，执行任务

4、Elastic-Job-Lite轻量级去中心化的特点

如何理解轻量级和去中心化？

5、任务分片

一个大的非常耗时的作业Job，比如：一次要处理一亿的数据，那这一亿的数据存储在数据库中，如果用一个作业节点处理一亿数据要很久，在互联网领域是不太能接受的，互联网领域更希望机器的增加去横向扩展处理能力。所以，ElasticJob可以把作业分为多个的task（每⼀个task就是⼀个任务分片），每⼀个task交给具体的⼀个机器实例去处理（⼀个机器实例是可以处理多个task的），但是具体每个task执行什么逻辑由我们自己来指定。

Strategy策略定义这些分片项怎么去分配到各个机器上去，默认是平均去分，可以定制，比如某一个机器负载比较高或者预配置比较高，那么就可以写策略。分片和作业本身是通过一个注册中心协调的，因为在分布式环境下，状态数据肯定集中到⼀点，才可以在分布式中沟通。

分片代码

6、弹性扩容

新增加⼀个运行实例app3，它会自动注册到注册中心，注册中心发现新的服务上线，注册中心会通知ElasticJob 进行重新分片，那么总得分片项有多少，那么就可以搞多少个实例机器，完全可以分1000片那么就可以搞1000台机器⼀起执行作业

注意：
1）分片项也是⼀个JOB配置，修改配置，重新分片，在下⼀次定时运行之前会重新调用分片算法，那么这个分片算法的结果就是：哪台机器运行哪一个一片，这个结果存储到zk中的，主节点会把分片给分好放到注册中心去，然后执行节点从注册中心获取信息(执行节点在定时任务开启的时候获取相应的分片)。
2）如果所有的节点挂掉值剩下一个节点，所有分片都会指向剩下的一个节点，这也是ElasticJob的高可用。