Disjob—分布式任务调度框架

2023年 10月 15日 28.2k 0

简介

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

框架流程图

应用场景举例

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

  • 拆分任务
  • 先根据当前的机器资源情况来决定拆分任务的数量,比如我们有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-否;
    }
    
  • 暂停与恢复
  • 假如我们的这几台机器资源需要临时做其它的事情,想把当前的统计任务暂停一段时间。No problem!框架是支持暂停执行中的任务,只需要在管理后台的调度实例页面,找到该任务点击暂停按钮即可。在暂停时任务会接收到一个中断信号,收到中断信号时同样可以在代码中使用Savepoint保存当前的执行快照。

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

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

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

    以下是本例中质数统计的job数据,只列了部分主要字段,其中job_handler为这两个任务处理器的编排(代码在项目源码中)

    {
      "jobGroup": "default",
      "jobName": "prime-count-dag",
      "jobState": 1, // job状态:0-禁用;1-启用;
      "jobType": 2,  // job类型:1-普通(Normal);2-工作流(DAG);
      "jobHandler": "cn.ponfee.disjob.test.handler.PrimeCountJobHandler -> cn.ponfee.disjob.test.handler.PrimeAccumulateJobHandler",
      "jobParam": "{"m":1,"n":10000000000,"blockSize":100000000,"parallel":10}",
      "triggerType": 2,
      "triggerValue": "2023-09-02 18:00:00"
    }
    

    本例中的质数统计流程图如下

    范例流程图

    项目地址

    gitee : gitee.com/dromara/dis…

    github: github.com/dromara/dis…

    相关文章

    JavaScript2024新功能:Object.groupBy、正则表达式v标志
    PHP trim 函数对多字节字符的使用和限制
    新函数 json_validate() 、randomizer 类扩展…20 个PHP 8.3 新特性全面解析
    使用HTMX为WordPress增效:如何在不使用复杂框架的情况下增强平台功能
    为React 19做准备:WordPress 6.6用户指南
    如何删除WordPress中的所有评论

    发布评论