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的树形节点结构图

image.png


引入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

IF

EXISTS `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

     */

    @Override

    public 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) {

        // 配置分布式协调服务(注册中心)Zookeeper

        ZookeeperConfiguration 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 {
    // url
    private static String url = "jdbc:mysql://127.0.0.1:3306/Job?characterEncoding=utf8&useSSL=false";
    // user
    private static String user = "root";
    // password
    private static String password = "root";
    // 驱动程序类
    private static String driver = "com.mysql.jdbc.Driver";
 
    static {
        try {
            Class.forName(driver);
        } catch (ClassNotFoundException e) {
// TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
 
    public static Connection getConnection() {
        try {
            return DriverManager.getConnection(url, user, password);
        } catch (SQLException e) {
// TODO Auto-generated catch block
            e.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 block
                e.printStackTrace();
            } finally {
                if (ps != null) {
                    try {
                        ps.close();
                    } catch (SQLException e) {
// TODO Auto-generated catch block
                        e.printStackTrace();
                    } finally {
                        if (con != null) {
                            try {
                                con.close();
                            } catch (SQLException e) {
// TODO Auto-generated catch block
                                e.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 block
            e.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 block
            e.printStackTrace();
        } finally {
            close(rs, ps, con);
        }
        return null;
    }
}


测试

可先启动⼀个进程,然后再启动⼀个进程(两个进程模拟分布式环境下,通⼀个定时任务部署了两份在工作)

两个进程逐个启动,观察现象

关闭其中执行的进程,观察现象

Leader节点选举机制

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

多个实例同时创建/leader节点

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

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

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

image.png

5、任务分片

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


image.png

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

分片代码

image.png

image.png

6、弹性扩容 

image.png

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

注意:

1)分片项也是⼀个JOB配置,修改配置,重新分片,在下⼀次定时运行之前会重新调用分片算法,那么这个分片算法的结果就是:哪台机器运行哪一个一片,这个结果存储到zk中的,主节点会把分片给分好放到注册中心去,然后执行节点从注册中心获取信息(执行节点在定时任务开启的时候获取相应的分片)。

2)如果所有的节点挂掉值剩下一个节点,所有分片都会指向剩下的一个节点,这也是ElasticJob的高可用。

————————————————

版权声明:本文为CSDN博主「悠然予夏」的原创文章,遵循CC 4.0 BY-SA版权协议,转载请附上原文出处链接及本声明。

原文链接:https://blog.csdn.net/weixin_52851967/article/details/125910448