一文读懂，硬核 Apache DolphinScheduler3.0 源码解析
Posted 海豚调度平台
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了一文读懂，硬核 Apache DolphinScheduler3.0 源码解析相关的知识，希望对你有一定的参考价值。
点亮 ⭐️ Star · 照亮开源之路
https://github.com/apache/dolphinscheduler
本文目录
- 1 DolphinScheduler的设计与策略
- 1.1 分布式设计
- 1.1.1 中心化
- 1.1.2 去中心化
- 1.2 DophinScheduler架构设计
- 1.3 容错问题
- 1.3.1 宕机容错
- 1.3.2 失败重试
- 1.4 远程日志访问
- 2 DolphinScheduler源码分析
- 2.1 工程模块介绍与配置文件
- 2.1.1 工程模块介绍
- 2.1.2 配置文件
- 2.2 Api主要任务操作接口
- 2.3 Quaterz架构与运行流程
- 2.3.1 概念与架构
- 2.3.2 初始化与执行流程
- 2.3.3 集群运转
- 2.4 Master启动与执行流程
- 2.4.1 概念与执行逻辑
- 2.4.2 集群与槽（slot）
- 2.4.3 代码执行流程
- 2.5 Work启动与执行流程
- 2.5.1 概念与执行逻辑
- 2.5.2 代码执行流程
- 2.6 rpc交互
- 2.6.1 Master与Worker交互
- 2.6.2 其他服务与Master交互
- 2.7 负载均衡算法
- 2.7.1 加权随机
- 2.7.2 线性负载
- 2.7.3 平滑轮询
- 2.8 日志服务
- 2.9 报警
- 3 后记
- 3.1 Make friends
- 3.2 参考文献
前言
研究Apache Dolphinscheduler也是机缘巧合，平时负责基于xxl-job二次开发出来的调度平台，因为遇到了并发性能瓶颈，到了不得不优化重构的地步，所以搜索市面上应用较广的调度平台以借鉴优化思路。
在阅读完DolphinScheduler代码之后，便生出了将其设计与思考记录下来的念头，这便是此篇文章的来源。因为没有正式生产使用，业务理解不一定透彻，理解可能有偏差，欢迎大家交流讨论。
1 DolphinScheduler的设计与策略
大家能关注DolphinScheduler那么一定对调度系统有了一定的了解，对于调度所涉及的到一些专有名词在这里就不做过多的介绍，重点介绍一下流程定义，流程实例，任务定义，任务实例。（没有作业这个概念确实也很新奇，可能是不想和Quartz的JobDetail重叠）。
- 任务定义：各种类型的任务，是流程定义的关键组成，如sql,shell,spark,mr，python等；
- 任务实例：任务的实例化，标识着具体的任务执行状态；
- 流程定义：一组任务节点通过依赖关系建立的起来的有向无环图（DAG）；
- 流程实例：通过手动或者定时调度生成的流程实例；
- 定时调度：系统采用Quartz 分布式调度器，并同时支持cron表达式可视化的生成；
1.1 分布式设计
分布式系统的架构设计基本分为中心化和去中心化两种，各有优劣，凭借各自的业务选择。
1.1.1 中心化
中心化设计比较简单，集群中的节点安装角色可以分为Master和Slave两种，如下图：
Master: Master的角色主要负责任务分发并监督Slave的健康状态，可以动态的将任务均衡到Slave上，以致Slave节点不至于“忙死”或”闲死”的状态。
中心化设计存在一些问题。
第一点，一旦Master出现了问题，则群龙无首，整个集群就会崩溃。
为了解决这个问题，大多数Master/Slave架构模式都采用了主备Master的设计方案，可以是热备或者冷备，也可以是自动切换或手动切换，而且越来越多的新系统都开始具备自动选举切换Master的能力,以提升系统的可用性。
第二点，如果Scheduler在Master上，虽然可以支持一个DAG中不同的任务运行在不同的机器上，但是会产生Master的过负载。如果Scheduler在Slave上，一个DAG中所有的任务都只能在某一台机器上进行作业提交，在并行任务比较多的时候，Slave的压力可能会比较大。
xxl-job就是采用这种设计方式，但是存在相应的问题。管理器(admin)宕机集群会崩溃，Scheduler在管理器上，管理器负责所有任务的校验和分发，管理器存在过载的风险,需要开发者想方案解决。
1.1.2 去中心化
在去中心化设计里，通常没有Master/Slave的概念，所有的角色都是一样的，地位是平等的，去中心化设计的核心设计在于整个分布式系统中不存在一个区别于其他节点的“管理者”，因此不存在单点故障问题。
但由于不存在“管理者”节点所以每个节点都需要跟其他节点通信才得到必须要的机器信息，而分布式系统通信的不可靠性，则大大增加了上述功能的实现难度。实际上，真正去中心化的分布式系统并不多见。
反而动态中心化分布式系统正在不断涌出。在这种架构下，集群中的管理者是被动态选择出来的，而不是预置的，并且集群在发生故障的时候，集群的节点会自发的举行会议来选举新的管理者去主持工作。
一般都是基于Raft算法实现的选举策略。Raft算法，目前社区也有相应的PR，还没合并。
- PR链接：https://github.com/apache/dolphinscheduler/issues/10874
- 动态展示见链接：http://thesecretlivesofdata.com/
DolphinScheduler的去中心化是Master/Worker注册到注册中心，实现Master集群和Worker集群无中心。
1.2 DophinScheduler架构设计
随手盗用一张官网的系统架构图，可以看到调度系统采用去中心化设计，由UI，API，MasterServer，Zookeeper，WorkServer，Alert等几部分组成。
API: API接口层，主要负责处理前端UI层的请求。该服务统一提供RESTful api向外部提供请求服务。接口包括工作流的创建、定义、查询、修改、发布、下线、手工启动、停止、暂停、恢复、从该节点开始执行等等。
MasterServer: MasterServer采用分布式无中心设计理念，MasterServer集成了Quartz，主要负责 DAG 任务切分、任务提交监控，并同时监听其它MasterServer和WorkerServer的健康状态。MasterServer服务启动时向Zookeeper注册临时节点，通过监听Zookeeper临时节点变化来进行容错处理。WorkServer:WorkerServer也采用分布式无中心设计理念，WorkerServer主要负责任务的执行和提供日志服务。WorkerServer服务启动时向Zookeeper注册临时节点，并维持心跳。
ZooKeeper: ZooKeeper服务，系统中的MasterServer和WorkerServer节点都通过ZooKeeper来进行集群管理和容错。另外系统还基于ZooKeeper进行事件监听和分布式锁。
**Alert:**提供告警相关接口，接口主要包括两种类型的告警数据的存储、查询和通知功能，支持丰富的告警插件自由拓展配置。
1.3 容错问题
容错分为服务宕机容错和任务重试，服务宕机容错又分为Master容错和Worker容错两种情况;
1.3.1 宕机容错
服务容错设计依赖于ZooKeeper的Watcher机制，实现原理如图：
其中Master监控其他Master和Worker的目录，如果监听到remove事件，则会根据具体的业务逻辑进行流程实例容错或者任务实例容错，容错流程图相对官方文档里面的流程图，人性化了些，大家可以参考一下，具体如下所示。
ZooKeeper Master容错完成之后则重新由DolphinScheduler中Scheduler线程调度，遍历 DAG 找到“正在运行”和“提交成功”的任务，对“正在运行”的任务监控其任务实例的状态，对“提交成功”的任务需要判断Task Queue中是否已经存在，如果存在则同样监控任务实例的状态，如果不存在则重新提交任务实例。
Master Scheduler线程一旦发现任务实例为” 需要容错”状态，则接管任务并进行重新提交。注意由于” 网络抖动”可能会使得节点短时间内失去和ZooKeeper的心跳，从而发生节点的remove事件。
对于这种情况，我们使用最简单的方式，那就是节点一旦和ZooKeeper发生超时连接，则直接将Master或Worker服务停掉。
1.3.2 失败重试
这里首先要区分任务失败重试、流程失败恢复、流程失败重跑的概念：
- 任务失败重试是任务级别的，是调度系统自动进行的，比如一个Shell任务设置重试次数为3次，那么在Shell任务运行失败后会自己再最多尝试运行3次。
- 流程失败恢复是流程级别的，是手动进行的，恢复是从只能从失败的节点开始执行或从当前节点开始执行。流程失败重跑也是流程级别的，是手动进行的，重跑是从开始节点进行。
接下来说正题，我们将工作流中的任务节点分了两种类型。
- 一种是业务节点，这种节点都对应一个实际的脚本或者处理语句，比如Shell节点、MR节点、Spark节点、依赖节点等。
- 还有一种是逻辑节点，这种节点不做实际的脚本或语句处理，只是整个流程流转的逻辑处理，比如子流程节等。
每一个业务节点都可以配置失败重试的次数，当该任务节点失败，会自动重试，直到成功或者超过配置的重试次数。逻辑节点不支持失败重试。但是逻辑节点里的任务支持重试。
如果工作流中有任务失败达到最大重试次数，工作流就会失败停止，失败的工作流可以手动进行重跑操作或者流程恢复操作。
1.4 远程日志访问
由于Web(UI)和Worker不一定在同一台机器上，所以查看日志不能像查询本地文件那样。
有两种方案：
- 将日志放到ES搜索引擎上；
- 通过netty通信获取远程日志信息；
介于考虑到尽可能的DolphinScheduler的轻量级性，所以选择了RPC实现远程访问日志信息，具体代码的实践见2.8章节。
2 DolphinScheduler源码分析
上一章的讲解可能初步看起来还不是很清晰，本章的主要目的是从代码层面一一介绍第一张讲解的功能。关于系统的安装在这里并不会涉及，安装运行请大家自行探索。
2.1 工程模块介绍与配置文件
2.1.1 工程模块介绍
- dolphinscheduler-alert 告警模块，提供告警服务;
- dolphinscheduler-api web应用模块，提供 Rest Api 服务，供 UI 进行调用;
- dolphinscheduler-common 通用的常量枚举、工具类、数据结构或者基类 dolphinscheduler-dao 提供数据库访问等操作;
- dolphinscheduler-remote 基于netty的客户端、服务端 ;
- dolphinscheduler-server 日志与心跳服务 ;
- dolphinscheduler-log-server LoggerServer 用于Rest Api通过RPC查看日志;
- dolphinscheduler-master MasterServer服务，主要负责 DAG 的切分和任务状态的监控 ;
- dolphinscheduler-worker WorkerServer服务，主要负责任务的提交、执行和任务状态的更新;
- dolphinscheduler-service service模块，包含Quartz、Zookeeper、日志客户端访问服务，便于server模块和api模块调用 ;
- dolphinscheduler-ui 前端模块;
2.1.2 配置文件
dolphinscheduler-common common.properties
#本地工作目录,用于存放临时文件
data.basedir.path=/tmp/dolphinscheduler
#资源文件存储类型: HDFS,S3,NONE
resource.storage.type=NONE
#资源文件存储路径
resource.upload.path=/dolphinscheduler
#hadoop是否开启kerberos权限
hadoop.security.authentication.startup.state=false
#kerberos配置目录
java.security.krb5.conf.path=/opt/krb5.conf
#kerberos登录用户
login.user.keytab.username=hdfs-mycluster@ESZ.COM
#kerberos登录用户keytab
login.user.keytab.path=/opt/hdfs.headless.keytab
#kerberos过期时间,整数,单位为小时
kerberos.expire.time=2
# 如果存储类型为HDFS,需要配置拥有对应操作权限的用户
hdfs.root.user=hdfs
#请求地址如果resource.storage.type=S3,该值类似为: s3a://dolphinscheduler. 如果resource.storage.type=HDFS, 如果 hadoop 配置了 HA,需要复制core-site.xml 和 hdfs-site.xml 文件到conf目录
fs.defaultFS=hdfs://mycluster:8020
aws.access.key.id=minioadmin
aws.secret.access.key=minioadmin
aws.reginotallow=us-east-1
aws.endpoint=http://localhost:9000
# resourcemanager port, the default value is 8088 if not specified
resource.manager.httpaddress.port=8088
#yarn resourcemanager 地址, 如果resourcemanager开启了HA, 输入HA的IP地址(以逗号分隔),如果resourcemanager为单节点, 该值为空即可
yarn.resourcemanager.ha.rm.ids=192.168.xx.xx,192.168.xx.xx
#如果resourcemanager开启了HA或者没有使用resourcemanager,保持默认值即可. 如果resourcemanager为单节点,你需要将ds1 配置为resourcemanager对应的hostname
yarn.application.status.address=http://ds1:%s/ws/v1/cluster/apps/%s
# job history status url when application number threshold is reached(default 10000, maybe it was set to 1000)
yarn.job.history.status.address=http://ds1:19888/ws/v1/history/mapreduce/jobs/%s
# datasource encryption enable
datasource.encryption.enable=false
# datasource encryption salt
datasource.encryption.salt=!@#$%^&*
# data quality option
data-quality.jar.name=dolphinscheduler-data-quality-dev-SNAPSHOT.jar
#data-quality.error.output.path=/tmp/data-quality-error-data
# Network IP gets priority, default inner outer
# Whether hive SQL is executed in the same session
support.hive.notallow=false
# use sudo or not, if set true, executing user is tenant user and deploy user needs sudo permissions; if set false, executing user is the deploy user and doesnt need sudo permissions
sudo.enable=true
# network interface preferred like eth0, default: empty
#dolphin.scheduler.network.interface.preferred=
# network IP gets priority, default: inner outer
#dolphin.scheduler.network.priority.strategy=default
# system env path
#dolphinscheduler.env.path=dolphinscheduler_env.sh
#是否处于开发模式
development.state=false
# rpc port
alert.rpc.port=50052
# Url endpoint for zeppelin RESTful API
zeppelin.rest.url=http://localhost:8080
dolphinscheduler-api application.yaml
server:
port: 12345
servlet:
session:
timeout: 120m
context-path: /dolphinscheduler/
compression:
enabled: true
mime-types: text/html,text/xml,text/plain,text/css,text/javascript,application/javascript,application/json,application/xml
jetty:
max-http-form-post-size: 5000000
spring:
application:
name: api-server
banner:
charset: UTF-8
jackson:
time-zone: UTC
date-format: "yyyy-MM-dd HH:mm:ss"
servlet:
multipart:
max-file-size: 1024MB
max-request-size: 1024MB
messages:
basename: i18n/messages
datasource:
# driver-class-name: org.postgresql.Driver
# url: jdbc:postgresql://127.0.0.1:5432/dolphinscheduler
driver-class-name: com.mysql.jdbc.Driver
url: jdbc:mysql://127.0.0.1:3306/dolphinscheduler?useUnicode=true&serverTimeznotallow=UTC&characterEncoding=UTF-8&autoRecnotallow=true&useSSL=false&zeroDateTimeBehavior=convertToNull
username: root
password: root
hikari:
connection-test-query: select 1
minimum-idle: 5
auto-commit: true
validation-timeout: 3000
pool-name: DolphinScheduler
maximum-pool-size: 50
connection-timeout: 30000
idle-timeout: 600000
leak-detection-threshold: 0
initialization-fail-timeout: 1
quartz:
auto-startup: false
job-store-type: jdbc
jdbc:
initialize-schema: never
properties:
org.quartz.threadPool:threadPriority: 5
org.quartz.jobStore.isClustered: true
org.quartz.jobStore.class: org.quartz.impl.jdbcjobstore.JobStoreTX
org.quartz.scheduler.instanceId: AUTO
org.quartz.jobStore.tablePrefix: QRTZ_
org.quartz.jobStore.acquireTriggersWithinLock: true
org.quartz.scheduler.instanceName: DolphinScheduler
org.quartz.threadPool.class: org.quartz.simpl.SimpleThreadPool
org.quartz.jobStore.useProperties: false
org.quartz.threadPool.makeThreadsDaemons: true
org.quartz.threadPool.threadCount: 25
org.quartz.jobStore.misfireThreshold: 60000
org.quartz.scheduler.makeSchedulerThreadDaemon: true
# org.quartz.jobStore.driverDelegateClass: org.quartz.impl.jdbcjobstore.PostgreSQLDelegate
org.quartz.jobStore.driverDelegateClass: org.quartz.impl.jdbcjobstore.StdJDBCDelegate
org.quartz.jobStore.clusterCheckinInterval: 5000
management:
endpoints:
web:
exposure:
include: *
metrics:
tags:
application: $spring.application.name
registry:
type: zookeeper
zookeeper:
namespace: dolphinscheduler
# connect-string: localhost:2181
connect-string: 10.255.158.70:2181
retry-policy:
base-sleep-time: 60ms
max-sleep: 300ms
max-retries: 5
session-timeout: 30s
connection-timeout: 9s
block-until-connected: 600ms
digest: ~
audit:
enabled: false
metrics:
enabled: true
python-gateway:
# Weather enable python gateway server or not. The default value is true.
enabled: true
# The address of Python gateway server start. Set its value to `0.0.0.0` if your Python API run in different
# between Python gateway server. It could be be specific to other address like `127.0.0.1` or `localhost`
gateway-server-address: 0.0.0.0
# The port of Python gateway server start. Define which port you could connect to Python gateway server from
# Python API side.
gateway-server-port: 25333
# The address of Python callback client.
python-address: 127.0.0.1
# The port of Python callback client.
python-port: 25334
# Close connection of socket server if no other request accept after x milliseconds. Define value is (0 = infinite),
# and socket server would never close even though no requests accept
connect-timeout: 0
# Close each active connection of socket server if python program not active after x milliseconds. Define value is
# (0 = infinite), and socket server would never close even though no requests accept
read-timeout: 0
# Override by profile
---
spring:
config:
activate:
on-profile: mysql
datasource:
driver-class-name: com.mysql.jdbc.Driver
url: jdbc:mysql://127.0.0.1:3306/dolphinscheduler?useUnicode=true&characterEncoding=UTF-8
quartz:
properties:
org.quartz.jobStore.driverDelegateClass: org.quartz.impl.jdbcjobstore.StdJDBCDelegate
dolphinscheduler-master application.yaml
spring:
banner:
charset: UTF-8
application:
name: master-server
jackson:
time-zone: UTC
date-format: "yyyy-MM-dd HH:mm:ss"
cache:
# default enable cache, you can disable by `type: none`
type: none
cache-names:
- tenant
- user
- processDefinition
- processTaskRelation
- taskDefinition
caffeine:
spec: maximumSize=100,expireAfterWrite=300s,recordStats
datasource:
#driver-class-name: org.postgresql.Driver
#url: jdbc:postgresql://127.0.0.1:5432/dolphinscheduler
driver-class-name: com.mysql.jdbc.Driver
url: jdbc:mysql://127.0.0.1:3306/dolphinscheduler?useUnicode=true&serverTimeznotallow=UTC&characterEncoding=UTF-8&autoRecnotallow=true&useSSL=false&zeroDateTimeBehavior=convertToNull
username: root
password:
hikari:
connection-test-query: select 1
minimum-idle: 5
auto-commit: true
validation-timeout: 3000
pool-name: DolphinScheduler
maximum-pool-size: 50
connection-timeout: 30000
idle-timeout: 600000
leak-detection-threshold: 0
initialization-fail-timeout: 1
quartz:
job-store-type: jdbc
jdbc:
initialize-schema: never
properties:
org.quartz.threadPool:threadPriority: 5
org.quartz.jobStore.isClustered: true
org.quartz.jobStore.class: org.quartz.impl.jdbcjobstore.JobStoreTX
org.quartz.scheduler.instanceId: AUTO
org.quartz.jobStore.tablePrefix: QRTZ_
org.quartz.jobStore.acquireTriggersWithinLock: true
org.quartz.scheduler.instanceName: DolphinScheduler
org.quartz.threadPool.class: org.quartz.simpl.SimpleThreadPool
org.quartz.jobStore.useProperties: false
org.quartz.threadPool.makeThreadsDaemons: true
org.quartz.threadPool.threadCount: 25
org.quartz.jobStore.misfireThreshold: 60000
org.quartz.scheduler.makeSchedulerThreadDaemon: true
# org.quartz.jobStore.driverDelegateClass: org.quartz.impl.jdbcjobstore.PostgreSQLDelegate
org.quartz.jobStore.driverDelegateClass: org.quartz.impl.jdbcjobstore.StdJDBCDelegate
org.quartz.jobStore.clusterCheckinInterval: 5000
registry:
type: zookeeper
zookeeper:
namespace: dolphinscheduler
# connect-string: localhost:2181
connect-string: 10.255.158.70:2181
retry-policy:
base-sleep-time: 60ms
max-sleep: 300ms
max-retries: 5
session-timeout: 30s
connection-timeout: 9s
block-until-connected: 600ms
digest: ~
master:
listen-port: 5678
# master fetch command num
fetch-command-num: 10
# master prepare execute thread number to limit handle commands in parallel
pre-exec-threads: 10
# master execute thread number to limit process instances in parallel
exec-threads: 100
# master dispatch task number per batch
dispatch-task-number: 3
# master host selector to select a suitable worker, default value: LowerWeight. Optional values include random, round_robin, lower_weight
host-selector: lower_weight
# master heartbeat interval, the unit is second
heartbeat-interval: 10
# master commit task retry times
task-commit-retry-times: 5
# master commit task interval, the unit is millisecond
task-commit-interval: 1000
state-wheel-interval: 5
# master max cpuload avg, only higher than the system cpu load average, master server can schedule. default value -1: the number of cpu cores * 2
max-cpu-load-avg: -1
# master reserved memory, only lower than system available memory, master server can schedule. default value 0.3, the unit is G
reserved-memory: 0.3
# failover interval, the unit is minute
failover-interval: 10
# kill yarn jon when failover taskInstance, default true
kill-yarn-job-when-task-failover: true
server:
port: 5679
management:
endpoints:
web:
exposure:
include: *
metrics:
tags:
application: $spring.application.name
metrics:
enabled: true
# Override by profile
---
spring:
config:
activate:
on-profile: mysql
datasource:
driver-class-name: com.mysql.jdbc.Driver
url: jdbc:mysql://127.0.0.1:3306/dolphinscheduler?useUnicode=true&serverTimeznotallow=UTC&characterEncoding=UTF-8&autoRecnotallow=true&useSSL=false&zeroDateTimeBehavior=convertToNull
quartz:
properties:
org.quartz.jobStore.driverDelegateClass: org.quartz.impl.jdbcjobstore.StdJDBCDelegate
dolphinscheduler-worker application.yaml
spring:
banner:
charset: UTF-8
application:
name: worker-server
jackson:
time-zone: UTC
date-format: "yyyy-MM-dd HH:mm:ss"
datasource:
#driver-class-name: org.postgresql.Driver
#url: jdbc:postgresql://127.0.0.1:5432/dolphinscheduler
driver-class-name: com.mysql.jdbc.Driver
url: jdbc:mysql://127.0.0.1:3306/dolphinscheduler?useUnicode=true&serverTimeznotallow=UTC&characterEncoding=UTF-8&autoRecnotallow=true&useSSL=false&zeroDateTimeBehavior=convertToNull
username: root
#password: root
password:
hikari:
connection-test-query: select 1
minimum-idle: 5
auto-commit: true
validation-timeout: 3000
pool-name: DolphinScheduler
maximum-pool-size: 50
connection-timeout: 30000
idle-timeout: 600000
leak-detection-threshold: 0
initialization-fail-timeout: 1
registry:
type: zookeeper
zookeeper:
namespace: dolphinscheduler
# connect-string: localhost:2181
connect-string: 10.255.158.70:2181
retry-policy:
base-sleep-time: 60ms
max-sleep: 300ms
max-retries: 5
session-timeout: 30s
connection-timeout: 9s
block-until-connected: 600ms
digest: ~
worker:
# worker listener port
listen-port: 1234
# worker execute thread number to limit task instances in parallel
exec-threads: 100
# worker heartbeat interval, the unit is second
heartbeat-interval: 10
# worker host weight to dispatch tasks, default value 100
host-weight: 100
# worker tenant auto create
tenant-auto-create: true
# worker max cpuload avg, only higher than the system cpu load average, worker server can be dispatched tasks. default value -1: the number of cpu cores * 2
max-cpu-load-avg: -1
# worker reserved memory, only lower than system available memory, worker server can be dispatched tasks. default value 0.3, the unit is G
reserved-memory: 0.3
# default worker groups separated by comma, like worker.groups=default,test
groups:
- default
# alert server listen host
alert-listen-host: localhost
alert-listen-port: 50052
server:
port: 1235
management:
endpoints:
web:
exposure:
include: *
metrics:
tags:
application: $spring.application.name
metrics:
enabled: true
主要关注数据库，quartz, zookeeper, masker, worker配置。
2.2 API主要任务操作接口
其他业务接口可以不用关注，只需要关注最最主要的流程上线功能接口，此接口可以发散出所有的任务调度相关的代码。
接口：/dolphinscheduler/projects/projectCode/schedules/id/online；此接口会将定义的流程提交到Quartz调度框架；代码如下：
public Map<String, Object> setScheduleState(User loginUser, long projectCode, Integer id, ReleaseState scheduleStatus) Map<String, Object> result = new HashMap<>();
Project project = projectMapper.queryByCode(projectCode); // check project auth boolean hasProjectAndPerm = projectService.hasProjectAndPerm(loginUser, project, result); if (!hasProjectAndPerm) return result;
// check schedule exists Schedule scheduleObj = scheduleMapper.selectById(id);
if (scheduleObj == null) putMsg(result, Status.SCHEDULE\\_CRON\\_NOT_EXISTS, id); return result; // check schedule release state if (scheduleObj.getReleaseState() == scheduleStatus) logger.info("schedule release is already ,neednt to change schedule id: from to ", scheduleObj.getReleaseState(), scheduleObj.getId(), scheduleObj.getReleaseState(), scheduleStatus); putMsg(result, Status.SCHEDULE\\_CRON\\_REALEASE\\_NEED\\_NOT_CHANGE, scheduleStatus); return result; ProcessDefinition processDefinition = processDefinitionMapper.queryByCode(scheduleObj.getProcessDefinitionCode()); if (processDefinition == null || projectCode != processDefinition.getProjectCode()) putMsg(result, Status.PROCESS\\_DEFINE\\_NOT_EXIST, String.valueOf(scheduleObj.getProcessDefinitionCode())); return result; List processTaskRelations = processTaskRelationMapper.queryByProcessCode(projectCode, scheduleObj.getProcessDefinitionCode()); if (processTaskRelations.isEmpty()) putMsg(result, Status.PROCESS\\_DAG\\_IS_EMPTY); return result; if (scheduleStatus == ReleaseState.ONLINE) // check process definition release state if (processDefinition.getReleaseState() != ReleaseState.ONLINE) logger.info("not release process definition id: , name : ", processDefinition.getId(), processDefinition.getName()); putMsg(result, Status.PROCESS\\_DEFINE\\_NOT_RELEASE, processDefinition.getName()); return result; // check sub process definition release state List subProcessDefinitionList = processDefinitionMapper.queryByCodes(subProcessDefineCodes); if (subProcessDefinitionList != null && !subProcessDefinitionList.isEmpty()) for (ProcessDefinition subProcessDefinition : subProcessDefinitionList) /\\\\ \\* if there is no online process, exit directly */ if (subProcessDefinition.getReleaseState() != ReleaseState.ONLINE) logger.info("not release process definition id: , name : ", subProcessDefinition.getId(), subProcessDefinition.getName()); putMsg(result, Status.PROCESS\\_DEFINE\\_NOT_RELEASE, String.valueOf(subProcessDefinition.getId())); return result;
// check master server exists List masterServers = monitorService.getServerListFromRegistry(true);
if (masterServers.isEmpty()) putMsg(result, Status.MASTER\\_NOT\\_EXISTS); return result;
// set status scheduleObj.setReleaseState(scheduleStatus);
scheduleMapper.updateById(scheduleObj);
try switch (scheduleStatus) case ONLINE: logger.info("Call master client set schedule online, project id: , flow id: ,host: ", project.getId(), processDefinition.getId(), masterServers); setSchedule(project.getId(), scheduleObj); break; case OFFLINE: logger.info("Call master client set schedule offline, project id: , flow id: ,host: ", project.getId(), processDefinition.getId(), masterServers); deleteSchedule(project.getId(), id); break; default: putMsg(result, Status.SCHEDULE\\_STATUS\\_UNKNOWN, scheduleStatus.toString()); return result; catch (Exception e) result.put(Constants.MSG, scheduleStatus == ReleaseState.ONLINE ? "set online failure" : "set offline failure"); throw new ServiceException(result.get(Constants.MSG).toString(), e);
putMsg(result, Status.SUCCESS); return result;
public void setSchedule(int projectId, Schedule schedule)
logger.info("set schedule, project id: , scheduleId: ", projectId, schedule.getId());
quartzExecutor.addJob(ProcessScheduleJob.class, projectId, schedule);
public void addJob(Class<? extends Job> clazz, int projectId, final Schedule schedule)
String jobName = this.buildJobName(schedule.getId());
String jobGroupName = this.buildJobGroupName(projectId);
Map<String, Object> jobDataMap = this.buildDataMap(projectId, schedule);
String cronExpression = schedule.getCrontab();
String timezoneId = schedule.getTimezoneId();
/**
* transform from server default timezone to schedule timezone
* e.g. server default timezone is `UTC`
* user set a schedule with startTime `2022-04-28 10:00:00`, timezone is `Asia/Shanghai`,
* api skip to transform it and save into databases directly, startTime `2022-04-28 10:00:00`, timezone is `UTC`, which actually added 8 hours,
* so when add job to quartz, it should recover by transform timezone
*/
Date startDate = DateUtils.transformTimezoneDate(schedule.getStartTime(), timezoneId);
Date endDate = DateUtils.transformTimezoneDate(schedule.getEndTime(), timezoneId);
lock.writeLock().lock();
try
JobKey jobKey = new JobKey(jobName, jobGroupName);
JobDetail jobDetail;
//add a task (if this task already exists, return this task directly)
if (scheduler.checkExists(jobKey))
jobDetail = scheduler.getJobDetail(jobKey);
jobDetail.getJobDataMap().putAll(jobDataMap);
else
jobDetail = newJob(clazz).withIdentity(jobKey).build();
jobDetail.getJobDataMap().putAll(jobDataMap);
scheduler.addJob(jobDetail, false, true);
logger.info("Add job, job name: , group name: ",
jobName, jobGroupName);
TriggerKey triggerKey = new TriggerKey(jobName, jobGroupName);
/*
* Instructs the Scheduler that upon a mis-fire
* situation, the CronTrigger wants to have its
* next-fire-time updated to the next time in the schedule after the
* current time (taking into account any associated Calendar),
* but it does not want to be fired now.
*/
CronTrigger cronTrigger = newTrigger()
.withIdentity(triggerKey)
.startAt(startDate)
.endAt(endDate)
.withSchedule(
cronSchedule(cronExpression)
.withMisfireHandlingInstructionDoNothing()
.inTimeZone(DateUtils.getTimezone(timezoneId))
)
.forJob(jobDetail).build();
if (scheduler.checkExists(triggerKey))
// updateProcessInstance scheduler trigger when scheduler cycle changes
CronTrigger oldCronTrigger = (CronTrigger) scheduler.getTrigger(triggerKey);
String oldCronExpression = oldCronTrigger.getCronExpression();
if (!StringUtils.equalsIgnoreCase(cronExpression, oldCronExpression))
// reschedule job trigger
scheduler.rescheduleJob(triggerKey, cronTrigger);
logger.info("reschedule job trigger, triggerName: , triggerGroupName: , cronExpression: , startDate: , endDate: ",
jobName, jobGroupName, cronExpression, startDate, endDate);
else
scheduler.scheduleJob(cronTrigger);
logger.info("schedule job trigger, triggerName: , triggerGroupName: , cronExpression: , startDate: , endDate: ",
jobName, jobGroupName, cronExpression, startDate, endDate);
catch (Exception e)
throw new ServiceException("add job failed", e);
finally
lock.writeLock().unlock();
2.3 Quaterz架构与运行流程
2.3.1 概念与架构
Quartz 框架主要包括如下几个部分：
- SchedulerFactory：任务调度工厂，主要负责管理任务调度器;
- Scheduler ：任务调度器，主要负责任务调度，以及操作任务的相关接口;
- Job ：任务接口，实现类包含具体任务业务代码;
- JobDetail：用于定义作业的实例;
- Trigger：任务触发器，主要存放 Job 执行的时间策略。例如多久执行一次，什么时候执行，以什么频率执行等等;
- JobBuilder ：用于定义/构建 JobDetail 实例，用于定义作业的实例。
- TriggerBuilder ：用于定义/构建触发器实例;
- Calendar：Trigger 扩展对象，可以排除或者包含某个指定的时间点（如排除法定节假日）;
- JobStore：存储作业和任务调度期间的状态Scheduler的生命期，从 SchedulerFactory 创建它时开始，到 Scheduler 调用Shutdown() 方法时结束；
Scheduler 被创建后，可以增加、删除和列举 Job 和 Trigger，以及执行其它与调度相关的操作（如暂停 Trigger）。但Scheduler 只有在调用 start() 方法后，才会真正地触发 trigger（即执行 job）
2.3.2 初始化与执行流程
Quartz的基本原理就是通过Scheduler来调度被JobDetail和Trigger定义的安装Job接口规范实现的自定义任务业务对象，来完成任务的调度。基本逻辑如下图：
代码时序图如下：
基本内容就是初始化任务调度容器Scheduler，以及容器所需的线程池，数据交互对象JobStore，任务处理线程QuartzSchedulerThread用来处理Job接口的具体业务实现类。
DolphinScheduler的业务类是ProcessScheduleJob，主要功能就是根据调度信息往commond表中写数据。
2.3.3 集群运转
需要注意的事：
- 当Quartz采用集群形式部署的时候，存储介质不能使用内存的形式，也就是不能使用JobStoreRAM。
- Quartz集群对于对于需要被调度的Triggers实例的扫描是使用数据库锁TRIGGER_ACCESS来完成的，保障此扫描过程只能被一个Quartz实例获取到。代码如下：
public List>() public List>() public Boolean validate(Connection conn, List(); for (FiredTriggerRecord ft : acquired) fireInstanceIds.add(ft.getFireInstanceId()); for (OperableTrigger tr : result) if (fireInstanceIds.contains(tr.getFireInstanceId())) return true; return false; catch (SQLException e) throw new JobPersistenceException("error validating trigger acquisition", e); );
3.集群失败实例恢复需要注意的是各个实例恢复各自实例对应的异常实例，因为数据库有调度容器的instanceId信息。代码如下：
protected void clusterRecover(Connection conn, List<SchedulerStateRecord> failedInstances)
throws JobPersistenceException
if (failedInstances.size() > 0)
long recoverIds = System.currentTimeMillis();
logWarnIfNonZero(failedInstances.size(),
"ClusterManager: detected " + failedInstances.size()
+ " failed or restarted instances.");
try
for (SchedulerStateRecord rec : failedInstances)
getLog().info(
"ClusterManager: Scanning for instance \\""
+ rec.getSchedulerInstanceId()
+ "\\"s failed in-progress jobs.");
List<FiredTriggerRecord> firedTriggerRecs = getDelegate()
.selectInstancesFiredTriggerRecords(conn,
rec.getSchedulerInstanceId());
int acquiredCount = 0;
int recoveredCount = 0;
int otherCount = 0;
Set<TriggerKey> triggerKeys = new HashSet<TriggerKey>();
for (FiredTriggerRecord ftRec : firedTriggerRecs)
TriggerKey tKey = ftRec.getTriggerKey();
JobKey jKey = ftRec.getJobKey();
triggerKeys.add(tKey);
// release blocked triggers..
if (ftRec.getFireInstanceState().equals(STATE_BLOCKED))
getDelegate()
.updateTriggerStatesForJobFromOtherState(
conn, jKey,
STATE_WAITING, STATE_BLOCKED);
else if (ftRec.getFireInstanceState().equals(STATE_PAUSED_BLOCKED))
getDelegate()
.updateTriggerStatesForJobFromOtherState(
conn, jKey,
STATE_PAUSED, STATE_PAUSED_BLOCKED);
// release acquired triggers..
if (ftRec.getFireInstanceState().equals(STATE_ACQUIRED))
getDelegate().updateTriggerStateFromOtherState(
conn, tKey, STATE_WAITING,
STATE_ACQUIRED);
acquiredCount++;
else if (ftRec.isJobRequestsRecovery())
// handle jobs marked for recovery that were not fully
// executed..
if (jobExists(conn, jKey))
@SuppressWarnings("deprecation")
SimpleTriggerImpl rcvryTrig = new SimpleTriggerImpl(
"recover_"
+ rec.getSchedulerInstanceId()
+ "_"
+ String.valueOf(recoverIds++),
Scheduler.DEFAULT_RECOVERY_GROUP,
new Date(ftRec.getScheduleTimestamp()));
rcvryTrig.setJobName(jKey.getName());
rcvryTrig.setJobGroup(jKey.getGroup());
rcvryTrig.setMisfireInstruction(SimpleTrigger.MISFIRE_INSTRUCTION_IGNORE_MISFIRE_POLICY);
rcvryTrig.setPriority(ftRec.getPriority());
JobDataMap jd = getDelegate().selectTriggerJobDataMap(conn, tKey.getName(), tKey.getGroup());
jd.put(Scheduler.FAILED_JOB_ORIGINAL_TRIGGER_NAME, tKey.getName());
jd.put(Scheduler.FAILED_JOB_ORIGINAL_TRIGGER_GROUP, tKey.getGroup());
jd.put(Scheduler.FAILED_JOB_ORIGINAL_TRIGGER_FIRETIME_IN_MILLISECONDS, String.valueOf(ftRec.getFireTimestamp()));
jd.put(Scheduler.FAILED_JOB_ORIGINAL_TRIGGER_SCHEDULED_FIRETIME_IN_MILLISECONDS, String.valueOf(ftRec.getScheduleTimestamp()));
rcvryTrig.setJobDataMap(jd);
rcvryTrig.computeFirstFireTime(null);
storeTrigger(conn, rcvryTrig, null, false,
STATE_WAITING, false, true);
recoveredCount++;
else
getLog()
.warn(
"ClusterManager: failed job "
+ jKey
+ " no longer exists, cannot schedule recovery.");
otherCount++;
else
otherCount++;
// free up stateful jobs triggers
if (ftRec.isJobDisallowsConcurrentExecution())
getDelegate()
.updateTriggerStatesForJobFromOtherState(
conn, jKey,
STATE_WAITING, STATE_BLOCKED);
getDelegate()
.updateTriggerStatesForJobFromOtherState(
conn, jKey,
STATE_PAUSED, STATE_PAUSED_BLOCKED);
getDelegate().deleteFiredTriggers(conn,
rec.getSchedulerInstanceId());
// Check if any of the fired triggers we just deleted were the last fired trigger
// records of a COMPLETE trigger.
int completeCount = 0;
for (TriggerKey triggerKey : triggerKeys)
if (getDelegate().selectTriggerState(conn, triggerKey).
equals(STATE_COMPLETE))
List<FiredTriggerRecord> firedTriggers =
getDelegate().selectFiredTriggerRecords(conn, triggerKey.getName(), triggerKey.getGroup());
if (firedTriggers.isEmpty())
if (removeTrigger(conn, triggerKey))
completeCount++;
logWarnIfNonZero(acquiredCount,
"ClusterManager: ......Freed " + acquiredCount
+ " acquired trigger(s).");
logWarnIfNonZero(completeCount,
"ClusterManager: ......Deleted " + completeCount
+ " complete triggers(s).");
logWarnIfNonZero(recoveredCount,
"ClusterManager: ......Scheduled " + recoveredCount
+ " recoverable job(s) for recovery.");
logWarnIfNonZero(otherCount,
"ClusterManager: ......Cleaned-up " + otherCount
+ " other failed job(s).");
if (!rec.getSchedulerInstanceId().equals(getInstanceId()))
getDelegate().deleteSchedulerState(conn,
rec.getSchedulerInstanceId());
catch (Throwable e)
throw new JobPersistenceException("Failure recovering jobs: "
+ e.getMessage(), e);
2.4 Master启动与执行流程
2.4.1 概念与执行逻辑
关键概念：
Quartz相关：
- Scheduler（任务调度容器，一般都是StdScheduler实例）。
- ProcessScheduleJob：（实现Quarts调度框架的Job接口的业务类，专门生成DolphinScheduler数据库业务表t\\_ds\\_commond数据）;
DolphinScheduler相关：
- NettyRemotingServer（netty服务端，包含netty服务端serverBootstrap对象与netty服务端业务处理对象serverHandler）， NettyServerHandler：（netty服务端业务处理类：包含各类处理器以及处理器对应的执行线程池）;
- TaskPluginManager（任务插件管理器，不同类型的任务以插件的形式管理，在应用服务启动的时候，通过@AutoService加载实现了TaskChannelFactory接口的工厂信息到数据库，通过工厂对象来加载各类TaskChannel实现类到缓存）；
- MasterRegistryClient（master操作zk的客户端，封装了master对于zk的所有操作，注册，查询，删除等）;
- MasterSchedulerService（扫描服务，包含业务执行线程和work包含
以上是关于一文读懂，硬核 Apache DolphinScheduler3.0 源码解析的主要内容，如果未能解决你的问题，请参考以下文章