一款分布式的任务调度与分布式计算框架，轻量级、简单易用

源码资源网 · 发表于 3 天前

一款分布式的任务调度与分布式计算框架，除了具备常规的任务调度功能外，还提供暂停/取消运行中的任务、恢复执行已暂停的任务、任务分片、失败重试、广播任务、任务依赖、工作流任务(DAG)、管理器与执行器分离部署、Web管理后台等能力。

轻量级，简单易用，特别适合长任务的执行。功能强大，稳定可靠，历经生产检验。

整体流程

一款分布式的任务调度与分布式计算框架，轻量级、简单易用

Features

分为管理器(Supervisor)和执行器(Worker)两种角色，Supervisor与Worker可分离部署
Supervisor与Worker通过注册中心相互发现，支持的注册中心有：Database、Redis、Consul、Nacos、Zookeeper、Etcd
Supervisor负责生成任务，把任务派发给Worker执行，支持的任务派发方式有：Redis、Http
需要指定Job的分组(group)，Job的任务只会派发给指定组的Worker执行
提供任务分片的能力，重写拆分方法JobSplitter#split即可拆分为多个任务，实现分布式任务及并行执行
支持暂停和取消运行中的任务，已暂停的任务可恢复继续执行，执行失败的任务支持重试
支持任务保存(Savepoint)其执行状态，让手动或异常暂停的任务能从上一次的执行状态中恢复继续执行
任务在执行时若抛出PauseTaskException，会暂停对应实例下的全部任务(包括分派到其它worker机器的任务)
支持广播任务，广播任务会派发给group下的所有worker执行
支持Job间的父子依赖，多个Job配置好依赖关系后便会按既定的依赖顺序依次执行
支持DAG工作流，可把jobExecutor配置为复杂的DAG表达式，如：A->B,C,(D->E)->F,G->H
支持执行中的Task在Worker(服务)发布完成后自动恢复继续执行(见shutdown_strategy配置)
提供Web管理后台，通过用户界面进行作业配置，任务监控等

Comparison[td]

	Quartz	Elastic-Job	Xxl-Job	Disjob
触发类型	Cron	Cron	Cron、固定频率、父子依赖	Cron、指定时间、固定频率、固定延时、父子依赖
任务编排	无	无	无	DAG表达式
任务分片	无	静态分片	广播任务	广播任务、动态分片
停止与恢复	无	无	终止运行中的任务	暂停执行中的任务、恢复执行已暂停的任务
保存执行快照	无	无	无	有
失败重试	无	失效转移	有	有
后台管理	无	有	有	有
监控告警	无	邮件	邮件	邮件、钉钉、企业微信、飞书、短信等

举个简单的例子：统计在(0，1万亿]区间内质数的个数。如果是单机单线程CPU的话要统计很长时间，这里我们就可以使用本框架提供的分布式计算能力来解决此类问题。

任务分片

先根据当前的机器资源情况来决定拆分任务的数量，比如我们有5台机器及每台2 core CPU(质数统计是CPU密集型)，决定拆分为10个任务。

派发任务

Supervisor使用指定的路由算法把拆分的10个子任务派发给这些Worker机器。

接收任务

Worker接收到子任务后，会提交到框架自定义的线程池中执行。

分布式并行计算

在执行时我们可以使用分批次方式(通过代码循环)来统计，这里我们指定task-1在第一次循环统计(0, 1亿]，第二次循环统计(10亿, 11亿]，以此类推最后一次循环统计(9990亿, 9991亿]。同理其它的task也是按同样的方式分布式并行统计。

P.s. 黎曼猜想中可知质数分布是大体均匀的，判断一个数是否质数有很多方法，如埃氏筛法、欧拉筛法、Miller Rabin素性检验，这里我们可以使用Guava库提供的素性检验。

Savepoint

如果在统计过程中机器宕机后怎么办？难道再从头开始统计吗？No No No！我们可以在每循环10次(或每隔执行超过1分钟)时使用Savepoint保存当前task-1的执行快照。宕机异常后的重新启动任务时会读取这份快照数据，从上一次的状态中接着继续统计。以下是task-1任务保存的快照数据样例

{ "next": 4000000001, // 下一次循环时要统计的区间为(40亿, 41亿] "count": 19819734, // 已经统计到了 19819734 个质数 "finished": false // 当前任务是否已经统计完成：true-是；false-否；}

服务发布

这些Task正在执行中，假如此时需要重新发布Worker(服务)该怎么办？Don't worry！在Worker发布的过程中无需人工干预，等服务发布完成一段时间后Task会自动恢复继续执行。

暂停与恢复

假如我们的这几台机器资源需要临时做其它的事情，想把当前的统计任务暂停一段时间。No problem！框架是支持暂停执行中的任务，只需要在管理后台的任务实例页面，找到该任务点击暂停按钮即可。在暂停时任务会接收到一个中断信号，收到中断信号时同样可以在代码中使用Savepoint保存当前的执行快照。

当其它事情执行完后，我们可以在管理后台的任务实例页面，找到被暂停的这个任务，点击恢复按钮，此时任务会从上一次保存的状态中恢复继续执行。

异常中断

子任务在执行过程中若抛出框架的PauseTaskException，则会暂停对应任务实例下全部的10个子任务(包括派发在不同机器中的任务)。同样如果抛出CancelTaskException则会取消对应任务实例下全部的10个子任务。如果抛出其它类型的异常时，只会取消当前子任务，对应任务实例下其它的子任务不受影响。

任务编排

现在这个质数统计的总任务已经执行完了，共10个子任务，每个子任务都统计出了它的那部分结果。Disjob能自动帮我汇总结果吗？Yes！框架提供了非常强大且方便的表达式来编排任务，如：A->B,C,(D->E)->F,G->H，现在我们就可以创建一个汇总任务，然后再把这两个任务编排在一起。

以下是本例质数统计中的job数据，只列了一些主要字段，其中jobExecutor编排了这两个任务执行器(见项目源码)

{
"group": "app-test",
"jobName": "prime-count-dag",
"jobState": 1, // job状态：0-禁用；1-启用；
"jobType": 2, // job类型：1-常规；2-工作流(DAG)；
"jobExecutor": "cn.ponfee.disjob.test.executor.PrimeCountJobExecutor -> cn.ponfee.disjob.test.executor.PrimeAccumulateJobExecutor",
"jobParam": "{"m":1,"n":10000000000,"blockSize":100000000,"parallel":10}",
"triggerType": 2,
"triggerValue": "2023-09-02 18:00:00"
}