分布式任务调度：海量任务处理elastic-Job解决方案(一)_elastic job能常驻百万任务

开局一个github，装备全靠捡：https://github.com/apache/shardingsphere-elasticjob

一、ElasticJob概述

1.1、什么是ElasticJob

ElasticJob 是面向海量任务处理的的分布式调度解决方案，由两个相互独立的子项目 ElasticJob-Lite 和 ElasticJob-Cloud 组成。它的各个产品使用统一的作业 API，开发者仅需一次开发，即可随意部署。具有弹性调度、资源管控、以及作业治理的功能。

1.2、ElasticJob-Lite和ElasticJob-Cloud

（1）ElasticJob-Lite

定位为轻量级无中心化解决方案，使用 jar 的形式提供分布式任务的协调服务

（2）ElasticJob-Cloud

采用自研 Mesos Framework 的解决方案，额外提供资源治理、应用分发以及进程隔离等功能

1.3、ElasticJob功能

• 弹性调度

  • 支持任务在分布式场景下的分片和高可用
  • 能够水平扩展任务的吞吐量和执行效率
  • 任务处理能力随资源配备弹性伸缩

• 资源分配

  • 在适合的时间将适合的资源分配给任务并使其生效
  • 相同任务聚合至相同的执行器统一处理
  • 动态调配追加资源至新分配的任务

• 作业治理

  • 失效转移
  • 错过作业重新执行
  • 自诊断修复

• 作业依赖(TODO)

  • 基于有向无环图（DAG）的作业间依赖
  • 基于有向无环图（DAG）的作业分片间依赖

• 作业开放生态

  • 可扩展的作业类型统一接口
  • 丰富的作业类型库，如数据流、脚本、HTTP、文件、大数据等
  • 易于对接业务作业，能够与 Spring 依赖注入无缝整合

• 可视化管控端

  • 作业管控端
  • 作业执行历史数据追踪
  • 注册中心管理

1.4、使用案例场景

大多数情况下，定时任务我们一般使用quartz开源框架就能满足应用场景。但如果考虑到健壮性等其它一些因素，就需要自己下点工夫，比如：要避免单点故障，至少得部署2个节点吧，但是部署多个节点，又有其它问题，有些数据在某一个时刻只能处理一次，比如 i = i+1 这些无法保证幂等的操作，run多次跟run一次，完全是不同的效果。

 

对于上面的问题，可自行设计过一个基于zk分布式锁的解决方案：

1、每类定时job，可以分配一个独立的标识（比如：xxx_job)

2、这类job的实例，部署在多个节点上时，每个节点启动前，向zk申请一个分布式锁（在xxx_job节点下）

3、拿到锁的实例，才允许启动定时任务(通过代码控制quartz的schedule)，没拿到锁的，处于standby状态，一直监听锁的变化

4、如果某个节点挂了，分布式锁自动释放，其它节点这时会抢到锁，按上面的逻辑，就会从standby状态，转为激活状态，小三正式上位，继续执行定时job。

 

这个方案，基本上解决了HA和业务正确性的问题，但是美中不足的地方有2点：

1、无法充分利用机器性能，处于standby的节点，实际上只是一个备胎，平时啥也不干

2、性能不方便扩展，比如：某个job一次要处理上千万的数据，仅1个激活节点，要处理很久

 

ElasticJob可解决上述问题。    ElasticJob相当于quartz+zk的加强版，它允许对定时任务分片，可以集群部署(每个job的"分片"会分散到各个节点上)，如果某个节点挂了，该节点上的分片，会调度到其它节点上

二、ElasticJob功能详解

2.1、调度模型

（1）进程内调度：ElasticJob-Lite

作业能够透明化的与业务应用系统相结合。 它能够方便的与 Spring 、Dubbo 等 Java 框架配合使用，在作业中可自由使用 Spring 注入的 Bean，如数据源连接池、Dubbo 远程服务等，更加方便的贴合业务开发

（2）进程级调度：ElasticJob-Cloud（也具有进程内调度功能）

能够对作业服务器的资源进行控制，因此其作业类型可划分为常驻任务和瞬时任务。 常驻任务类似于 ElasticJob-Lite，是进程内调度；瞬时任务则完全不同，它充分的利用了资源分配的削峰填谷能力，是进程级的调度，每次任务的会启动全新的进程处理。

2.2、弹性调度

弹性调度是 ElasticJob 最重要的功能，能够让任务通过分片进行水平扩展的任务处理

（1）分片

分片项的概念，使得任务可以在分布式的环境下运行，每台任务服务器只运行分配给该服务器的分片。 随着服务器的增加或宕机，ElasticJob 会近乎实时的感知服务器数量的变更，从而重新为分布式的任务服务器分配更加合理的任务分片项，使得任务可以随着资源的增加而提升效率。

任务的分布式执行，需要将一个任务拆分为多个独立的任务项，然后由分布式的服务器分别执行某一个或几个分片项

举例：如果作业分为 4 片，用两台服务器执行，则每个服务器分到 2 片，分别负责作业的 50% 的负载

（2）分片项

ElasticJob 并不直接提供数据处理的功能，而是将分片项分配至各个运行中的作业服务器，开发者需要自行处理分片项与业务的对应关系。 分片项为数字，始于 0 而终于分片总数减 1

（3）资源最大限度利用

当新增加作业服务器时，ElasticJob 会通过注册中心的临时节点的变化感知到新服务器的存在，并在下次任务调度的时候重新分片，新的服务器会承载一部分作业分片

（4）高可用

当作业服务器在运行中宕机时，注册中心同样会通过临时节点感知，并将在下次运行时将分片转移至仍存活的服务器，以达到作业高可用的效果。 本次由于服务器宕机而未执行完的作业，则可以通过失效转移的方式继续执行

（5）ElasticJob-Lite 实现原理

无作业调度中心节点。而是基于部署作业框架的程序在到达相应时间点时各自触发调度。

注册中心：仅用于作业注册和监控信息存储。

主作业节点：仅用于处理分片和清理等功能。

• 第一台服务器上线触发主服务器选举。主服务器一旦下线，则重新触发选举，选举过程中阻塞，只有主服务器选举完成，才会执行其他任务。
• 某作业服务器上线时会自动将服务器信息注册到注册中心，下线时会自动更新服务器状态。
• 节点选举，服务器上下线，分片总数变更均更新重新分片标记。
• 定时任务触发时，如需重新分片，则通过主服务器分片，分片过程中阻塞，分片结束后才可执行任务。如分片过程中主服务器下线，则先选举主服务器，再分片。
• 每次分片都会按服务器IP排序，保证分片结果不会产生较大波动。
• 实现失效转移功能，在某台服务器执行完毕后主动抓取未分配的分片，并且在某台服务器下线后主动寻找可用的服务器执行任务。

................未完待续

2.3、资源分配

ElasticJob-Cloud 作业运行模式：瞬时作业、常驻作业

调度器：ElasticJob-Cloud 基于 Mesos 的 Framework 开发，用于资源调度和应用分发，需要独立启动并提供服务。

作业应用：指作业打包部署后的应用，描述了作业启动需要用到的 CPU、内存、启动脚本及应用下载路径等基本信息。 每个作业应用可以包含一个或多个作业。

作业：实际运行的具体任务，和 ElasticJob-Lite 共用同样的作业生态。 在注册作业之前必须先注册作业应用。

资源：指作业启动或运行需要用到的 CPU、内存。

2.4、失效转移

ElasticJob 不会在本次执行过程中进行重新分片，而是等待下次调度之前才开启重新分片流程。 当作业执行过程中服务器宕机，失效转移允许将该次未完成的任务在另一作业节点上补偿执行。

失效转移：失效转移是当前执行作业的临时补偿执行机制，在下次作业运行时，会通过重分片对当前作业分配进行调整。

实时执行：当其他服务器感知到有失效转移的作业需要处理时，且该作业服务器已经完成了本次任务，则会实时的拉取待失效转移的分片项，并开始补偿执行。

异步执行：作业服务在本次任务执行结束后，会向注册中心问询待执行的失效转移分片项，如果有，则开始补偿执行。

适用场景：不建议短间隔作业开启失效转移。

2.5、错过任务执行

错过任务重执行功能可以使逾期未执行的作业在之前作业执行完成之后立即执行

适用场景：不建议短间隔作业开启失效转移。适合运行耗时较长且间隔较长的作业场景

2.6、作业开放生态

作业接口：ElasticJob 的作业可划分为基于 class 类型和基于 type 类型两种。Class 类型的作业由开发者直接使用，需要由开发者实现该作业接口实现业务逻辑。典型代表：Simple 类型、Dataflow 类型。 Type 类型的作业只需提供类型名称即可，开发者无需实现该作业接口，而是通过外置配置的方式使用。典型代表：Script 类型、HTTP 类型

执行器接口：用于执行用户定义的作业接口，通过 Java 的 SPI 机制织入 ElasticJob生态。

三、elastic-Job和xxl-Job的区别

xxl-job是基于数据库的，任务量大时会增加数据库压力有较大瓶颈；而elastic-job是基于zookeeper的，可处理海量任务

接入的企业：

• elastic-Job：36氪，当当网，国美，金柚网，联想，唯品会，亚信，平安，猪八戒
• xxl-Job：大众点评，运满满，优信二手车，拍拍贷

（1）多节点部署时任务不能重复执行

X-Job　：　使用Quartz基于数据库的分布式功能
E-Job　 ：　将任务拆分为n个任务项后，各个服务器分别执行各自分配到的任务项。一旦有新的服务器加入集群，或现有服务器下线，elastic-job将在保留本次任务执行不变的情况下，下次任务开始前触发任务重分片。

（2）日志可追溯

X-Job　：　支持，有日志查询界面
E-Job　：　可通过事件订阅的方式处理调度过程的重要事件，用于查询、统计和监控

（3）监控告警

X-Job　：　调度失败时，将会触发失败报警，如发送报警邮件。

任务调度失败时邮件通知的邮箱地址，支持配置多邮箱地址，配置多个邮箱地址时用逗号分隔

E-Job　：　通过事件订阅方式可自行实现

作业运行状态监控、监听作业服务器存活、监听近期数据处理成功、数据流类型作业（可通过监听近期数据处理成功数判断作业流量是否正常,如果小于作业正常处理的阀值，可选择报警。）、监听近期数据处理失败（可通过监听近期数据处理失败数判断作业处理结果，如果大于0，可选择报警。）

（4）支持并行调度

X-Job　：　调度系统多线程（默认10个线程）触发调度运行，确保调度精确执行，不被堵塞。
E-Job　：　采用任务分片方式实现。将一个任务拆分为n个独立的任务项，由分布式的服务器并行执行各自分配到的分片项。

（5）高可用策略

X-Job　：　“调度中心”通过DB锁保证集群分布式调度的一致性, 一次任务调度只会触发一次执行；
E-Job　：　调度器的高可用是通过运行几个指向同一个ZooKeeper集群的Elastic-Job-Cloud-Scheduler实例来实现的。ZooKeeper用于在当前主Elastic-Job-Cloud-Scheduler实例失败的情况下执行领导者选举。通过至少两个调度器实例来构成集群，集群中只有一个调度器实例提供服务，其他实例处于”待命”状态。当该实例失败时，集群会选举剩余实例中的一个来继续提供服务

（6）失败处理策略

X-Job　：　调度失败时的处理策略，策略包括：失败告警（默认）、失败重试；
E-Job　：　弹性扩容缩容在下次作业运行前重分片，但本次作业执行的过程中，下线的服务器所分配的作业将不会重新被分配。失效转移功能可以在本次作业运行中用空闲服务器抓取孤儿作业分片执行。同样失效转移功能也会牺牲部分性能

（7）动态分片策略

X-Job　：　分片广播任务以执行器为维度进行分片，支持动态扩容执行器集群从而动态增加分片数量，协同进行业务处理；在进行大数据量业务操作时可显著提升任务处理能力和速度。

执行器集群部署时，任务路由策略选择”分片广播”情况下，一次任务调度将会广播触发对应集群中所有执行器执行一次任务，同时传递分片参数；可根据分片参数开发分片任务；

E-Job　：　支持多种分片策略，可自定义分片策略

默认包含三种分片策略： 基于平均分配算法的分片策略、 作业名的哈希值奇偶数决定IP升降序算法的分片策略、根据作业名的哈希值对Job实例列表进行轮转的分片策略，支持自定义分片策略

elastic-job的分片是通过zookeeper来实现的。分片的分片由主节点分配，如下三种情况都会触发主节点上的分片算法执行：
a、新的Job实例加入集群
b、现有的Job实例下线（如果下线的是leader节点，那么先选举然后触发分片算法的执行）
c、主节点选举”

 

 

 

官方文档：https://shardingsphere.apache.org/elasticjob/current/cn/overview/

官网：http://shardingsphere.apache.org/elasticjob/index_zh.html