[Java知识库]分布式调度Elastic-job

概述

什么是任务调度？

我们可以先思考一下下面业务场景的解决方案：

- 某平台需要在用户会员过期日前三天进行短信提醒

- 某电商平台需要上午10点，下午3点，晚上8点发放一批优惠券

以上场景就是任务调度所需要解决的问题

而在我们在学习spring的过程就会学习使用spring提供的定时任务注解@Scheduled来进行定时任务的执行

大致回忆一下，在业务类中方法贴上此注解

@Scheduled(cron = "0/20 * * * * ? ")  //此为cron表达式，设置定时所需的时间点或周期
 public void doWork(){
 	//doSomething   
 }

然后在启动类上贴上@EnableScheduling注解，可以理解为要贴有此注解才能让定时任务生效

为什么要使用分布式调度？

感觉Spring给我们提供的这个注解可以完成任务调度的功能，好像已经完美解决问题了，为什么还需要分布式呢?

主要有如下这几点原因:

????????1.单机处理极限：原本1分钟内需要处理1万个订单，但是现在需要1分钟内处理10万个订单；原来一个统计需要1小时，现在业务方需要10分钟就统计出来。你也许会说，你也可以多线程、单机多进程处理。的确，多线程并行处理可以提高单位时间的处理效率，但是单机能力毕竟有限（主要是CPU、内存和磁盘），始终会有单机处理不过来的情况。

????????2.高可用：单机版的定式任务调度只能在一台机器上运行，如果程序或者系统出现异常就会导致功能不可用。虽然可以在单机程序实现的足够稳定，但始终有机会遇到非程序引起的故障，而这个对于一个系统的核心功能来说是不可接受的。

????????3.防止重复执行: 在单机模式下，定时任务是没什么问题的。但当我们部署了多台服务，同时又每台服务又有定时任务时，若不进行合理的控制在同一时间，只有一个定时任务启动执行，这时，定时执行的结果就可能存在混乱和错误了

这个时候就需要分布式的任务调度来实现了。

更俗的话就是两种情况：

????????1.例如我们要去对表的数据进行备份的操作，这个表中有1000W条数据

若是我们只是单一去执行这个需求，那么会耗费很多的时间

????????那么这种情况就有可能会去进行一个集群，但是集群后又会出现一个问题，就是在spring定时器情况下，所有的集群做的都是同样的是事情，而我们需要的就是不同的微服务做的是不同的，但是都是为了去备份这一份数据库的数据，所以无法实现

????????2.例如要把mysql数据同步到redis中

????????正常情况下我们也会去进行集群操作，比如我们有两个服务，此时两个服务都有定时任务，这个定时任务就会在两台机器中同时执行，我们可能只想要让其中一个去执行，而不是同时执行，所以使用spring定时器是不行的

第一个例子总结说就是为了将大的拆成小的，分成多份各自执行各自的，但是都是为了完成一个需求

第二个例子总结说就是集群中都是同样功能，但是定时器只需要其中一个进行执行就可以

Elastic-Job介绍

Elastic-Job是一个分布式调度的解决方案，使用Elastic-Job可以快速实现分布式任务调度。

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

功能列表:

- 分布式调度协调

????????在分布式环境中，任务能够按照指定的调度策略执行，并且能够避免同一任务多实例重复执行。

- 丰富的调度策略：

????????基于成熟的定时任务作业框架Quartz cron表达式执行定时任务。

- 弹性拓容缩容

????????当集群中增加一个实例，它应当能够被选举被执行任务；当集群减少一个实例时，他所执行的任务能被转移到别的示例中执行。

- 失效转移

????????某示例在任务执行失败后，会被转移到其他实例执行。

- 错过执行任务重触发

????????若因某种原因导致作业错过执行，自动记录错误执行的作业，并在下次次作业完成后自动触发。

- 支持并行调度

????????支持任务分片，任务分片是指将一个任务分成多个小任务在多个实例同时执行。

- 作业分片一致性

????????当任务被分片后，保证同一分片在分布式环境中仅一个执行实例。

- 支持作业生命周期操作

????????可以动态对任务进行开启及停止操作。

- 丰富的作业类型

????????????????支持Simple、DataFlow、Script三种作业类型

（elasticJob会把定时任务的信息存放到zookeeper中，zookeeper不单单是注册中心，也可以作为一个存数据的容器 ）

系统架构图

?

Elastic-Job快速入门

环境搭建

1.版本要求：

- JDK 要求1.7以上保本

- Maven 要求3.0.4及以上版本

- Zookeeper 要求采取3.4.6以上版本

2.Zookeeper需要进行安装和运行

3.创建maven项目

添加依赖

<dependency>
    <groupId>com.dangdang</groupId>
    <artifactId>elastic-job-lite-core</artifactId>
    <version>2.1.5</version>
</dependency>

4.任务类代码

public class MyElasticJob implements SimpleJob {
    /**
     * Elasticjob会按照你的corn表达式去执行execute方法
     */
    public void execute(ShardingContext shardingContext) {
        System.out.println("定时任务开始====>"+new Date());
    }
}

5.配置类

public class JobDemo {
    public static void main(String[] args) {
        /**
         * 第一个参数 注册中心配置
         * 第二个参数 是任务的配置：比如任务在什么时间节点上哪个类中执行方法
         */
        new JobScheduler(createRegistryCenter(), createJobConfiguration()).init();
    }

    private static CoordinatorRegistryCenter createRegistryCenter() {
        /**
         * CoordinatorRegistryCenter  注册中心对象
         * new ZookeeperConfiguration("localhost:2181", "elastic-job-demo")
         * 一个是本地的注册中心的端口   一个是保存在zookeeper的名
         */
        CoordinatorRegistryCenter regCenter = new ZookeeperRegistryCenter
                (new ZookeeperConfiguration("localhost:2181", "elastic-job-demo"));
        regCenter.init();
        return regCenter;
    }

    private static LiteJobConfiguration createJobConfiguration() {
        /**
         * 第一个参数：给定时任务取得名字，任务名称（唯一
         * 第二个参数: corn表达式
         * 第三个参数: 分片数量
         */
        // 定义作业核心配置
        JobCoreConfiguration simpleCoreConfig = JobCoreConfiguration.newBuilder("demoSimpleJob","0/3 * * * * ?",1).build();
        // 定义SIMPLE类型配置
        //当到达时间点的时候，去执行第二个参数的那个类里的execute方法
        SimpleJobConfiguration simpleJobConfig = new SimpleJobConfiguration(simpleCoreConfig, MyElasticJob.class.getCanonicalName());
        // 定义Lite作业根配置
        LiteJobConfiguration simpleJobRootConfig = LiteJobConfiguration.newBuilder(simpleJobConfig).build();
        return simpleJobRootConfig;
    }
}

6.进行测试

- 运行单个程序，查看是否按照cron表达式的内容进行任务的调度
- 运行多个程序(集群)，查看是否只会有一个实例进行任务调度
- 运行多个程序后，把正在进行任务调度的进程关掉，查看其它进程是否能继续进行任务调度

注意：它有一个失效转移机制，当启动一个挂掉了，他就会转移到另一个节点中进行定时任务

SpringBoot集成Elastic-Job

1.创建项目

2.添加Maven依赖

    <groupId>cn.wolfcode</groupId>
    <artifactId>elastic-job-springboot</artifactId>
    <version>1.0.0</version>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.3.RELEASE</version>
    </parent>
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>com.dangdang</groupId>
            <artifactId>elastic-job-lite-spring</artifactId>
            <version>2.1.5</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
    </dependencies>

3.相关配置

因为配置中心的地址并不是固定的，所以我们应该把这个地址信息配置在配置文件中，所以在配置文件application.yml中添加配置如下:

zookeeper:
  addr: localhost:2181
  namespace: springboot-elastic-demo

4.zookeeper注册中心配置类

@Configuration
public class MyRegisterCenterConfig {

    /**
     * 注册中心配置
     */
    @Bean
    public  CoordinatorRegistryCenter createRegistryCenter(@Value("${zookeeper.addr}") String addr,@Value("${zookeeper.namespace}") String namespace) {
        CoordinatorRegistryCenter regCenter = new ZookeeperRegistryCenter(new ZookeeperConfiguration(addr, namespace));
        regCenter.init();
        return regCenter;
    }
}

5.任务调度类

@Configuration
public class MyJobConfiguration {

    @Autowired
    private CoordinatorRegistryCenter coordinatorRegistryCenter;

        /**
     *  spring中使用这个类，将他交给spring容器管理，在启动时就会按照这个任务类的配置（new SpringJobScheduler）去执行定时任务
     */
    @Bean(initMethod = "init")
    public SpringJobScheduler springJobScheduler(MyElasticJob elasticJob){

        /**
         * 第一个参数   任务类   在指定时间要执行哪一个方法  需要一个elasticJob类型的类
         * 第二个参数   注册中心
         */
        return new SpringJobScheduler(elasticJob,coordinatorRegistryCenter,createJobConfiguration("0/3 * * * * ?",MyElasticJob.class));
    }


    private static LiteJobConfiguration createJobConfiguration(String cron, Class clz) {
        /**
         * 第一个参数：任务名称（唯一
         * 第二个参数: corn表达式
         * 第三个参数: 分片数量
         */
        // 定义作业核心配置
        JobCoreConfiguration simpleCoreConfig = JobCoreConfiguration.newBuilder(clz.getSimpleName(), cron, 1).build();
        // 定义SIMPLE类型配置
        SimpleJobConfiguration simpleJobConfig = new SimpleJobConfiguration(simpleCoreConfig, clz.getCanonicalName());
        // 定义Lite作业根配置
        LiteJobConfiguration simpleJobRootConfig = LiteJobConfiguration.newBuilder(simpleJobConfig).build();
        return simpleJobRootConfig;
    }
}

可以启动第二个服务8082，进行测试，就可以发现我们已经解决了那个集群中只需要一个服务进行定时任务的需求，并且还能验证转移机制

分片概念

????????作业分片是指任务的分布式执行，需要将一个任务拆分为多个独立的任务项，然后由分布式的应用实例分别执行某一个或者几个分布项。

????????例如：Elastic-Job快速入门中文件备份的案例（入门案例链接在最下面），现有两台服务器，每台服务器分别跑一个应用实例。为了快速执行作业，可以讲任务分成4片，那么每个应用实例都执行两片。作业遍历数据逻辑应为：实例1查找text和image类型文件执行备份，实例2查找radio和vedio类型文件执行备份。如果由于服务器拓容应用实例数量增加为4，则作业遍历数据的逻辑应为: 4个实例分别处理text,image,radio,video类型的文件。

可以看到，通过对任务的合理分片化，从而达到任务并行处理的效果.

? ? ? ?通俗讲就是多台机器执行一个任务，想要的效果就是一个大的任务拆分为很多小的任务并在多台机器中执行-----------分片机制

分片项与业务处理解耦

????????Elastic-Job并不直接提供数据处理的功能,框架只会将分片项分配至各个运行中的作业服务器，开发者需要自行处理分片项与真实数据的对应关系

最大限度利用资源

????????将分片项设置大于服务器的数据，最好是大于服务器倍数的数量，作业将会合理利用分布式资源，动态的分配分片项.

????????例如: 3台服务器，分成10片，则分片项结果为服务器A=0,1,2;服务器B=3,4,5;服务器C=6,7,8,9.如果 服务器C奔溃，则分片项分配结果为服务器A=0,1,2,3,4;服务器B=5,6,7,8,9.在不丢失分片项的情况下，最大限度利用现有的资源提高吞吐量.

????????那么就可以对springboot继承Elastic-Job案例进行改造，改成任务分片

Dataflow类型调度任务

????????我们从之前的入门案例和集成案例可以发现，我们的任务类都是实现SimpleJob的，那么还有一种类型就是dataflow。

????????Dataflow类型的定时任务需要实现Dataflowjob接口，该接口提供2个方法供覆盖，分别用于抓取（fetchData）和处理（processData）数据，我们继续对例子进行改造。

????????Dataflow类型用于处理数据流，他和SimpleJob不同，它以数据流的方式执行，调用fetchData抓取数据，知道抓取不到数据才停止作业。

????????simpleJob相当于生活中的搬家时一次性把所有东西都搬过去，Dataflowjob相当于分几次把东西搬走。

????????例如我们需要查询1000W条的数据库数据并备份，而我们若是一次性查出1000W的数据去统一做备份，那么可能效率就会比较低。若是使用dataflowjob 那么我们可以一次查询20W条数据备份完后再去处理后面的20W条数据，依次如此，一点点处理完。

备注：部分摘取于一些博主，侵删。

??????????个人总结，有误请大佬指出。