作业的必要性以及存在的问题

1. 为什么需要作业？

作业即定时任务。一般来说，系统可使用消息传递代替部分使用作业的场景。两者确有相似之处。可互相替换的场景，如队列表。将待处理的数据放入队列表，然后使用频率极短的定时任务拉取队列表的数据并处理。这种情况使用消息中间件的推送模式可更好的处理实时性数据。而且基于数据库的消息存储吞吐量远远小于基于文件的顺序追加消息存储。

a) 时间驱动 OR 事件驱动：内部系统一般可以通过事件来驱动，但涉及到外部系统，则只能使用时间驱动。如：抓取第三方支付的对账文件，上游业务系统的对账文件。隔天抓取，由于是外部系统，不能像内部系统一样发送事件触发事件。

b) 批量处理 OR 逐条处理：批量处理堆积的数据更加高效，在不需要实时性的情况下比消息中间件更有优势。而且有的业务逻辑只能批量处理，如：电商公司与快递公司结算，一个月结算一次，并且根据送货的数量有提成。比如，当月送货超过1000则额外给快递公司多1%的快递费。

c) 非实时性 OR 实时性：虽然消息中间件可以做到实时处理数据，但有的情况并不需要。如：VIP用户降级，如果超过1年无购买行为，则自动降级。这类需求没有强烈的时间要求，不需要按照时间精确的降级VIP用户。

d) 系统内部 OR 系统解耦：作业一般封装在系统内部，而消息中间件可用于系统间解耦。

2. 当前常见的作业系统存在哪些问题？

a) Quartz：Java事实上的定时任务标准。但Quartz关注点在于定时任务而非数据，并无一套根据数据处理而定制化的流程。虽然Quartz可以基于数据库实现作业的高可用，但缺少分布式并行调度的功能。

b) TBSchedule：阿里早期开源的分布式任务调度系统。代码略陈旧，使用timer而非线程池执行任务调度。众所周知，timer在处理异常状况时是有缺陷的，定时任务用Timer实现有可能出现异常，因为它是基于绝对时间而不是相对时间进行调度的。当环境的系统时间被修改后，原来的定时任务可能就不跑了。另外需要注意一点，捕获并处理定时任务的异常。如果在TimerTask里抛出了异常，那么Timer认为定时任务被取消并终止执行线程。而且TBSchedule作业类型较为单一，只能是获取/处理数据一种模式。还有就是文档缺失比较严重。

c) Crontab：Linux系统级的定时任务执行器。缺乏分布式和集中管理功能。

综上所述，当前存在的作业系统缺少分布式、并行调度、弹性扩容缩容、集中管理、定制化流程型任务等功能，所以需要一个新的作业系统完善这些功能。

解决思路

修改开源产品和基于开源产品重新搭建，是两种可行性较高的方案。

修改开源产品可控性较低，且一般都是针对于某些特定功能，所以我们采用将成熟的开源产品作为积木。

elastic-job-lite主要的设计理念是无中心化的分布式定时调度框架，思路来源于Quartz的基于数据库的高可用方案。但数据库没有分布式协调功能，所以在高可用方案的基础上增加了弹性扩容和数据分片的思路，以便于更大限度的利用分布式服务器的资源。

elastic-job-lite

1. 主要功能

a) 分布式：重写Quartz基于数据库的分布式功能，改用Zookeeper实现注册中心。

b) 并行调度：采用任务分片方式实现。将一个任务拆分为n个独立的任务项，由分布式的服务器并行执行各自分配到的分片项。

c) 弹性扩容缩容：将任务拆分为n个任务项后，各个服务器分别执行各自分配到的任务项。一旦有新的服务器加入集群，或现有服务器下线，elastic-job将在保留本次任务执行不变的情况下，下次任务开始前触发任务重分片。

d) 集中管理：采用基于Zookeeper的注册中心，集中管理和协调分布式作业的状态，分配和监听。外部系统可直接根据Zookeeper的数据管理和监控elastic-job。

e) 定制化流程型任务：作业可分为简单和数据流处理两种模式，数据流又分为高吞吐处理模式和顺序性处理模式，其中高吞吐处理模式可以开启足够多的线程快速的处理数据，而顺序性处理模式将每个分片项分配到一个独立线程，用于保证同一分片的顺序性，这点类似于kafka的分区顺序性。

2. 其他功能

a) 失效转移：弹性扩容缩容在下次作业运行前重分片，但本次作业执行的过程中，下线的服务器所分配的作业将不会重新被分配。失效转移功能可以在本次作业运行中用空闲服务器抓取孤儿作业分片执行。同样失效转移功能也会牺牲部分性能。

b) Spring命名空间支持：elastic-job可以不依赖于spring直接运行，但是也提供了自定义的命名空间方便与spring集成。

c) 运维平台：提供web控制台用于管理作业。

3. 非功能需求

a) 稳定性：在服务器无波动的情况下，并不会重新分片；即使服务器有波动，下次分片的结果也会根据服务器IP和作业名称哈希值算出稳定的分片顺序，尽量不做大的变动。

b) 高性能：同一服务器的批量数据处理采用自动切割并多线程并行处理。

c) 灵活性：所有在功能和性能之间的权衡，都可通过配置开启/关闭。如：elastic-job会将作业运行状态的必要信息更新到注册中心。如果作业执行频度很高，会造成大量Zookeeper写操作，而分布式Zookeeper同步数据可能引起网络风暴。因此为了考虑性能问题，可以牺牲一些功能，而换取性能的提升。

d) 幂等性：elastic-job可牺牲部分性能用以保证同一分片项不会同时在两个服务器上运行。

e) 容错性：作业服务器和Zookeeper断开连接则立即停止作业运行，用于防止分片已经重新分配，而脑裂的服务器仍在继续执行，导致重复执行。

4.开发指南

1. 环境要求

a. Java

请使用JDK1.7及其以上版本。

b. Zookeeper

请使用Zookeeper 3.4.6及其以上版本。

c. Maven

请使用Maven 3.0.4及其以上版本。

2. 作业开发

Elastic-Job-Lite和Elastic-Job-Cloud提供统一作业接口，开发者仅需对业务作业进行一次开发，之后可根据不同的配置以及部署至不同的Lite或Cloud环境。

I Elastic-Job提供Simple、Dataflow和Script 3种作业类型。 方法参数shardingContext包含作业配置、片和运行时信息。可通过getShardingTotalCount(), getShardingItem()等方法分别获取分片总数，运行在本作业服务器的分片序列号等。

a. Simple类型作业

意为简单实现，未经任何封装的类型。需实现SimpleJob接口。该接口仅提供单一方法用于覆盖，此方法将定时执行。与Quartz原生接口相似，但提供了弹性扩缩容和分片等功能。

b. Dataflow类型作业

流式处理

可通过DataflowJobConfiguration配置是否流式处理。

流式处理数据只有fetchData方法的返回值为null或集合长度为空时，作业才停止抓取，否则作业将一直运行下去； 非流式处理数据则只会在每次作业执行过程中执行一次fetchData方法和processData方法，随即完成本次作业。

如果采用流式作业处理方式，建议processData处理数据后更新其状态，避免fetchData再次抓取到，从而使得作业永不停止。 适用于不间歇的数据处理。

3. 作业配置

Elastic-Job配置分为3个层级，分别是Core, Type和Root。每个层级使用相似于装饰者模式的方式装配。

Core对应JobCoreConfiguration，用于提供作业核心配置信息，如：作业名称、分片总数、CRON表达式等。

Type对应JobTypeConfiguration，有3个子类分别对应SIMPLE, DATAFLOW和SCRIPT类型作业，提供3种作业需要的不同配置，如：DATAFLOW类型是否流式处理或SCRIPT类型的命令行等。

Root对应JobRootConfiguration，有2个子类分别对应Lite和Cloud部署类型，提供不同部署类型所需的配置，如：Lite类型的是否需要覆盖本地配置或Cloud占用CPU或Memory数量等。

Spring命名空间配置

与Spring容器配合使用作业，可将作业Bean配置为Spring Bean，并在作业中通过依赖注入使用Spring容器管理的数据源等对象。可用placeholder占位符从属性文件中取值。Lite可考虑使用Spring命名空间方式简化配置。

详细配置信息

4. 作业启动

a. Java启动方式

b. Spring启动方式

将配置Spring命名空间的xml通过Spring启动，作业将自动加载。

5.原理

a.第一台服务器上线触发主服务器选举。主服务器一旦下线，则重新触发选举，选举过程中阻塞，只有主服务器选举完成，才会执行其他任务。

b. 某作业服务器上线时会自动将服务器信息注册到注册中心，下线时会自动更新服务器状态。

c.主节点选举，服务器上下线，分片总数变更均更新重新分片标记。

d.定时任务触发时，如需重新分片，则通过主服务器分片，分片过程中阻塞，分片结束后才可执行任务。如分片过程中主服务器下线，则先选举主服务器，再分片。

e.通过上一项说明可知，为了维持作业运行时的稳定性，运行过程中只会标记分片状态，不会重新分片。分片仅可能发生在下次任务触发前。

f.每次分片都会按服务器IP排序，保证分片结果不会产生较大波动。

g.实现失效转移功能，在某台服务器执行完毕后主动抓取未分配的分片，并且在某台服务器下线后主动寻找可用的服务器执行任务。

作业启动流程

作业执行流程

1.elastic-job详解（一）：数据分片

数据分片的目的在于把一个任务分散到不同的机器上运行，既可以解决单机计算能力上限的问题，也能降低部分任务失败对整体系统的影响。elastic-job并不直接提供数据处理的功能，框架只会将分片项分配至各个运行中的作业服务器（其实是Job实例，部署在一台机器上的多个Job实例也能分片），开发者需要自行处理分片项与真实数据的对应关系。框架也预置了一些分片策略：平均分配算法策略，作业名哈希值奇偶数算法策略，轮转分片策略。同时也提供了自定义分片策略的接口。

分片原理

elastic-job的分片是通过zookeeper来实现的。分片的分片由主节点分配，如下三种情况都会触发主节点上的分片算法执行：

新的Job实例加入集群

现有的Job实例下线（如果下线的是leader节点，那么先选举然后触发分片算法的执行）

主节点选举

上述三种情况，会让zookeeper上leader节点的sharding节点上多出来一个necessary的临时节点，主节点每次执行Job前，都会去看一下这个节点，这个节点就是一个是否执行分片的一个标记 如果有则执行分片算法。

分片的执行结果会存储在zookeeper上，如下图，5个分片，每个分片应该由哪个Job实例来运行都已经分配好。分配的过程就是上面触发分片算法之后的操作。分配完成之后，各个Job实例就会在下次执行的时候使用上这个分配结果。

每个job实例任务触发前都会获取本任务在本实例上的分片情况（直接和上图zookeeper上instance节点比对某一个分片是否该有这个Job实例执行），然后封装成shardingContext，传递给调用任务的实际执行方法：

分片算法

所有的分片策略都继承JobShardingStrategy接口。根据当前注册到ZK的实例列表和在客户端配置的分片数量来进行数据分片。最终将每个Job实例应该获得的分片数字返回出去。 方法签名如下：

分片函数的触发，只会在leader选举的时候触发，也就是说只会在刚启动和leader节点离开的时候触发，并且是在leader节点上触发，而其他节点不会触发。

1. 基于平均分配算法的分片策略

基于平均分配算法的分片策略对应的类是：AverageAllocationJobShardingStrategy。它是默认的分片策略。它的分片效果如下：

如果有3个Job实例, 分成9片, 则每个Job实例分到的分片是: 1=[0,1,2], 2=[3,4,5], 3=[6,7,8].

如果有3个Job实例, 分成8片, 则每个Job实例分到的分片是: 1=[0,1,6], 2=[2,3,7], 3=[4,5].

如果有3个Job实例, 分成10片, 则个Job实例分到的分片是: 1=[0,1,2,9], 2=[3,4,5], 3=[6,7,8].

2. 作业名的哈希值奇偶数决定IP升降序算法的分片策略

这个策略的对应的类是：OdevitySortByNameJobShardingStrategy，它内部其实也是使用AverageAllocationJobShardingStrategy实现，只是在传入的节点实例顺序不一样，也就是上面接口参数的List。AverageAllocationJobShardingStrategy的缺点是一旦分片数小于Job实例数，作业将永远分配至IP地址靠前的Job实例上，导致IP地址靠后的Job实例空闲。而OdevitySortByNameJobShardingStrategy则可以根据作业名称重新分配Job实例负载。如：如果有3个Job实例，分成2片，作业名称的哈希值为奇数，则每个Job实例分到的分片是：1=[0], 2=[1], 3=[] 如果有3个Job实例，分成2片，作业名称的哈希值为偶数，则每个Job实例分到的分片是：3=[0], 2=[1], 1=[]

3. 根据作业名的哈希值对Job实例列表进行轮转的分片策略

这个策略的对应的类是：RotateServerByNameJobShardingStrategy，和上面介绍的策略一样，内部同样是用AverageAllocationJobShardingStrategy实现，也是在传入的List列表顺序上做文章。

4. 自定义分片策略

除了可以使用上述分片策略之外，elastic-job还允许自定义分片策略。我们可以自己实现JobShardingStrategy接口，并且配置到分片方法上去，整个过程比较简单，下面仅仅列出通过配置spring来切换自定义的分片算法的例子：

2.elastic-job详解（二）：作业的调度

JobScheduler是elastic-job作业调度的关键类，也是起始类，在包com.dangdang.ddframe.job.lite.api下。调度任务的执行需要包含两大步骤：任务的配置和任务的注册。JobScheduler的构造函数除了任务配置和注册相关信息之外还有事件和监听。后两者是elastic-job的扩展功能。

任务的配置

由于内部使用quartz作为任务调度框架，任务的配置的相关的基础信息也是和quartz一致的。elastic-job的任务配置类在quartz的基础上（执行方法，cron表达式等）额外封装了分片策略，监控作业相关参数以及与注册中心的时间误差秒数等配置项。

任务的注册

elastic-job是通过zookeeper进行任务协调和故障转移的，任务的注册也就是把任务注册到zookeeper里面去。任务的注册包含在任务的启动过程中。根节点是项目的名称，下面一级是任务的名称。任务一旦创建则不能修改任务的名称，如果修改名称将视为新的任务，创建新的节点。任务名称节点下又包含5个数据子节点，分别是config, instances, leader, servers和sharding。如下图：

1. config节点：

任务的配置信息，包含执行类，cron表达式，分片算法类，分片数量，分片参数等等。config节点的数据是通过ConfigService持久化到zookeeper中去的。默认状态下，如果你修改了Job的配置比如cron表达式，分片数量等是不会更新到zookeeper上去的，除非你把参数overwrite修改成true。

2. instances节点：

同一个Job下的elastic-job的部署实例。一台机器上可以启动多个Job实例，也就是Jar包。instances的命名是IP+@-@+PID。

3. leader节点：

任务实例的主节点信息，通过zookeeper的主节点选举，选出来的主节点信息。下面的子节点分为election，sharding和failover三个子节点。分别用于主节点选举，分片和失效转移处理。election下面的instance节点显式了当前主节点的实例ID：jobInstanceId。latch节点也是一个永久节点用于选举时候的实现分布式锁。sharding节点下面有一个临时节点，necessary，是否需要重新分片的标记。如果分片总数变化，或任务实例节点上下线或启用/禁用，以及主节点选举，都会触发设置重分片标记，主节点会进行分片计算。

4. servers节点：

任务实例的信息，主要是IP地址，任务实例的IP地址。如果多个任务实例在同一台机器上运行则只会出现一个IP子节点。可在IP地址节点写入DISABLED表示该任务实例禁用。 在新的cloud native架构下，servers节点大幅弱化，仅包含控制服务器是否可以禁用这一功能。为了更加纯粹的实现job核心，servers功能未来可能删除，控制服务器是否禁用的能力应该下放至自动化部署系统。

5. sharding节点：

任务的分片信息，子节点是分片项序号，从零开始，至分片总数减一。分片个个数是在任务配置中设置的。分片项序号的子节点存储详细信息。每个分片项下的子节点用于控制和记录分片运行状态。最主要的子节点就是instance。举例来说，上图有三个分片，每个分片下面有个instance的节点，也就说明了这个分片在哪个instance上运行。如上文所说如果分片总数变化，或任务实例节点上下线或启用/禁用，以及主节点选举，都会触发设置重分片标记，主节点会进行分片计算。分片计算的结果也就体现在这instance上。

上文介绍的节点信息并不全面，还有一些会在特定的情况下出现的临时节点，更多详细的介绍可以请参看官方文档：http://dangdangdotcom.github.io/elastic-job/elastic-job-lite/03-design/lite-design/

3.elastic-job详解（三）：Job的手动触发功能

elastic-job的任务都是使用quartz来触发的，quartz表达式一般都是定期执行。但有时候一些周期较长的任务，比如一天一次，几小时一次的任务，我们需要等待很久才能触发一次。如果我们需要测试一些功能，或者说重跑一些任务的话，手动触发任务功能就变得尤为重要，也是任务框架一般都必须要实现的特性。现在我们就来聊一聊elastic-job的手动触发任务的功能。

触发功能的使用

elastic-job有个管理界面console，这里不在赘述怎么样部署和打开管理界面。在这个管理界面上，有个触发按钮，点击这个按钮就能触发这个任务。

任务定时周期是5s，当点击触发的时候，任务立即出发，如下图。

原理分析

elastic-job依靠zookeeper传递消息和quartz本身的触发功能来实现远程触发的功能。

当点击“触发”按钮时，管理页面会从zookeeper中找到当前Job下所有的任务实例，在实例节点上写入数据“TRIGGER”。如下图：

每个节点实例启动的时候，elastic-job默认会将任务触发监听器JobTriggerStatusJobListener启动，使用curator来监控instances节点的数据变化，当出现变化则触发JobTriggerStatusJobListener的dataChanged方法。从而最终调用quartz的triggerJob方法，触发任务。

4.elastic-job详解（四）：失效转移

elastic-job中最关键的特性之一就是失效转移。配置了失效转移之后，如果在任务执行过程中有一个执行实例挂了，那么之前被分配到这个实例的任务（或者分片）会在下次任务执行之前被重新分配到其他正常节点实例上执行。

简单的HA

当某一个任务实例节点宕机（离开与zookeeper的连接），会触发elastic-job主节点的重新分片逻辑。elastic-job启动任务节点以后生成的zookeeper中的instance节点是一个临时节点EPHEMERAL。为什么要用EPHEMERAL节点，就是为了能在任务实例出现问题与zookeeper断开以后，能触发zookeeper的节点移除的事件，从而重新调整分区或者运行节点。既然是EPHEMERAL节点，就可以在zookeeper中配置sessionTimeoutMs参数。在使用spring的elastic-job配置中在如下地方配置：

如果在sessionTimeoutMs的时间段之内触发任务，则异常分片的任务会丢失。举个例子：假如sessionTimeoutMs被设置成1分钟，而本身的任务是30秒执行一次，有三个任务实例在三台机器各自执行分片1,2,3。当分片3所在的机器出现问题，和zk断开了，那么zk节点失效至少要到1分钟以后。期间30秒执行一次的任务分片3，至少会少执行一次。1分钟过后，zk节点失效，触发ListenServersChangedJobListener类的dataChanged方法，在这里方法中判断instance节点变化，然后通过方法shardingService.setReshardingFlag设置重新分片标志位，下次执行任务的时候，leader节点重新分配分片，分片3就会转移到其他好的机器上。

复杂的失效转移

elastic-job的任务配置有个failover，如果开启设置为true的时候，会启动真正的失效转移：，elastic-job的任务又两个配置failover（默认值为false）和monitorExecution（默认值是true）。只有对monitorExecution为true的情况下才可以开启失效转移。

所谓失效转移，就是在执行任务的过程中遇见异常的情况，这个分片任务可以在其他节点再次执行。这个和上面的HA不同，对于HA，上面如果任务终止，那么不会在其他任务实例上再次重新执行。

Job的失效转移监听来源于FailoverListenerManager中JobCrashedJobListener的dataChanged方法。FailoverListenerManager监听的是zk的instance节点删除事件。如果任务配置了failover等于true，其中某个instance与zk失去联系或被删除，并且失效的节点又不是本身，就会触发失效转移逻辑。

首先，在某个任务实例失效时，elastic-job会在leader节点下面创建failover节点以及items节点。items节点下会有失效任务实例的原本应该做的分片好。比如，失效的任务实例原来负责分片1和2。那么items节点下就会有名字叫1的子节点，就代表分片1需要转移到其他节点上去运行。如下图：

然后，由于每个存活着的任务实例都会收到zk节点丢失的事件，哪个分片失效也已经在leader节点的failover子节点下。所以这些或者的任务实例就会争抢这个分片任务来执行。为了保证不重复执行，elastic-job使用了curator的LeaderLatch类来进行选举执行。在获得执行权后，就会在sharding节点的分片上添加failover节点，并写上任务实例，表示这个故障任务迁移到某一个任务实例上去完成。如下图中的sharding节点上的分片1：

执行完成后，会把相应的节点和数据删除，避免下一次重复执行。

5.elastic-job详解（五）：自定义任务参数

在elastic-job详解（三）：Job的手动触发功能一文中讲到了如何手动触发一个Job，但是我们手动触发的时候常常需要输入一些参数。举个栗子：我们有个对账文件，每天凌晨统计一次，统计上一天的数据。但我们发现几天前的某一天的数据有问题，需要重跑统计。这就需要统计程序能执行指定某一天的数据。这个功能就可以使用自定义任务参数来轻松实现。自定义参数，可通过传递该参数为作业调度的业务方法传参，用于实现带参数的作业。例：每次获取的数据量、作业实例从数据库读取的主键、执行某一天的任务等。

1. 支持自定义参数的代码

要使用自定义参数，就必须要在你实现的任务中使用这个参数，下面我们来看一下具体实现的代码：

真正和业务处理方法doWork接收一个日期参数，如果自定义参数没有则按照常规使用当前时间来进行统计，如果通过手动触发给定了参数，则使用给定的参数来做任务。自定义参数是根据shardingContext.getJobParameter()来获取。

2. 触发自定义参数

代码层面支持之后，我们来看一下怎么触发。首先找到console管理界面，连上正在执行的任务，点击“修改”按钮。

进入任务修改界面，在自定义参数一栏添加参数，然后点击更新。

最后，在图1中点击触发按钮。带参数的任务就会被触发。千万不要忘记执行成功之后再次修改回来，避免每次执行都会使用这个自定义参数。

elastic-job内部是在zookeeper中修改任务config节点中”jobParameter”的数据。

扩展阅读