您现在的位置是：首页 > 文章详情

分布式任务调度框架 DisJob 加入 Dromara 开源社区

日期：2023-09-04点击：71收藏

作者简介

网名Ponfee，Dromara开源组织成员，dromara/disjob项目作者。在国内多个一线大厂待过，有过后端、全栈、大数据等相关工作经历。

关于Disjob

Disjob是天然为支持分布式长任务执行而设计的，它除了具备常规的任务调度功能外，还提供：任务拆分及分布式并行执行、暂停及取消运行中的任务、恢复执行被暂停的任务、保存任务的执行快照(Checkpoint)、任务编排(DAG)、广播任务等能力。以下是Disjob的整体流程图：

应用场景举例

举个简单的例子：统计在\\(0，1万亿]\*\*区间内质数的个数。如果单机单线程CPU去统计的话不知道要到何年马月，这里我们就可以用Disjob框架提供的分布式并行执行的能力来解决该类问题。

拆分任务

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

分发任务

总任务被拆分成10个子任务后，框架会使用指定的路由算法把子任务分发给这些机器。

接收任务

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

分布式并行执行

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

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

Checkpoint

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

以下是task-1任务保存的快照数据样例

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

暂停与恢复

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

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

任务编排

现在这个质数统计的总任务已经执行完了，共10个子任务，每个子任务都统计出了它的那部分结果。Disjob能自动帮我汇总结果吗？Yes！框架提供了非常强大且方便的表达式来编排任务，如：A->B,C,(D->E)->D,F->G，现在我们可以创建一个汇总任务，然后再把这两个任务编排在一起。以下是本例中质数统计的job表数据，其中job_handler指定了编排的这两个任务处理器(代码在Disjob的开源项目中)。

 INSERT INTO `sched_job` ( `job_id`, `job_group`, `job_name`, `job_handler`, `job_state`, `job_type`, `job_param`, `trigger_type`, `trigger_value`, `next_trigger_time` ) VALUES ( 1003164910267351009, 'default', 'prime-count-dag', 'cn.ponfee.disjob.test.handler.PrimeCountJobHandler -> cn.ponfee.disjob.test.handler.PrimeAccumulateJobHandler', 1, 2, '{\\"m\\":1,\\"n\\":10000000000,\\"blockSize\\":100000000,\\"parallel\\":10}', 2, '2023-09-02 18:00:00', unix_timestamp()*1000 );