一文读懂，硬核 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

1. #本地工作目录,用于存放临时文件
2. data.basedir.path=/tmp/dolphinscheduler
3. #资源文件存储类型: HDFS,S3,NONE
4. resource.storage.type=NONE
5. #资源文件存储路径
6. resource.upload.path=/dolphinscheduler
7. #hadoop是否开启kerberos权限
8. hadoop.security.authentication.startup.state=false
9. #kerberos配置目录
10. java.security.krb5.conf.path=/opt/krb5.conf
11. #kerberos登录用户
12. login.user.keytab.username=hdfs-mycluster@ESZ.COM
14. #kerberos登录用户keytab
15. login.user.keytab.path=/opt/hdfs.headless.keytab
17. #kerberos过期时间,整数,单位为小时
18. kerberos.expire.time=2
19. #        如果存储类型为HDFS,需要配置拥有对应操作权限的用户
20. hdfs.root.user=hdfs
21. #请求地址如果resource.storage.type=S3,该值类似为: s3a://dolphinscheduler. 如果resource.storage.type=HDFS, 如果 hadoop 配置了 HA,需要复制core-site.xml 和 hdfs-site.xml 文件到conf目录
22. fs.defaultFS=hdfs://mycluster:8020
23. aws.access.key.id=minioadmin
24. aws.secret.access.key=minioadmin
25. aws.region=us-east-1
26. aws.endpoint=http://localhost:9000
27. # resourcemanager port, the default value is 8088 if not specified
28. resource.manager.httpaddress.port=8088
29. #yarn resourcemanager 地址, 如果resourcemanager开启了HA, 输入HA的IP地址(以逗号分隔),如果resourcemanager为单节点, 该值为空即可
30. yarn.resourcemanager.ha.rm.ids=192.168.xx.xx,192.168.xx.xx
31. #如果resourcemanager开启了HA或者没有使用resourcemanager,保持默认值即可. 如果resourcemanager为单节点,你需要将ds1 配置为resourcemanager对应的hostname
32. yarn.application.status.address=http://ds1:%s/ws/v1/cluster/apps/%s
33. # job history status url when application number threshold is reached(default 10000, maybe it was set to 1000)
34. yarn.job.history.status.address=http://ds1:19888/ws/v1/history/mapreduce/jobs/%s
36. # datasource encryption enable
37. datasource.encryption.enable=false
39. # datasource encryption salt
40. datasource.encryption.salt=!@#$%^&*
42. # data quality option
43. data-quality.jar.name=dolphinscheduler-data-quality-dev-SNAPSHOT.jar
45. #data-quality.error.output.path=/tmp/data-quality-error-data
47. # Network IP gets priority, default inner outer
49. # Whether hive SQL is executed in the same session
50. support.hive.oneSession=false
52. # 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 doesn't need sudo permissions
53. sudo.enable=true
55. # network interface preferred like eth0, default: empty
56. #dolphin.scheduler.network.interface.preferred=
58. # network IP gets priority, default: inner outer
59. #dolphin.scheduler.network.priority.strategy=default
61. # system env path
62. #dolphinscheduler.env.path=dolphinscheduler_env.sh
64. #是否处于开发模式
65. development.state=false
67. # rpc port
68. alert.rpc.port=50052
70. # Url endpoint for zeppelin RESTful API
71. zeppelin.rest.url=http://localhost:8080

复制代码

dolphinscheduler-api application.yaml

1. server:
2.   port: 12345
3.   servlet:
4.     session:
5.       timeout: 120m
6.     context-path: /dolphinscheduler/
7.   compression:
8.     enabled: true
9.     mime-types: text/html,text/xml,text/plain,text/css,text/javascript,application/javascript,application/json,application/xml
10.   jetty:
11.     max-http-form-post-size: 5000000
13. spring:
14.   application:
15.     name: api-server
16.   banner:
17.     charset: UTF-8
18.   jackson:
19.     time-zone: UTC
20.     date-format: "yyyy-MM-dd HH:mm:ss"
21.   servlet:
22.     multipart:
23.       max-file-size: 1024MB
24.       max-request-size: 1024MB
25.   messages:
26.     basename: i18n/messages
27.   datasource:
28. #    driver-class-name: org.postgresql.Driver
29. #    url: jdbc:postgresql://127.0.0.1:5432/dolphinscheduler
30.     driver-class-name: com.mysql.jdbc.Driver
31.     url: jdbc:mysql://127.0.0.1:3306/dolphinscheduler?useUnicode=true&serverTimezone=UTC&characterEncoding=UTF-8&autoReconnect=true&useSSL=false&zeroDateTimeBehavior=convertToNull
32.     username: root
33.     password: root
34.     hikari:
35.       connection-test-query: select 1
36.       minimum-idle: 5
37.       auto-commit: true
38.       validation-timeout: 3000
39.       pool-name: DolphinScheduler
40.       maximum-pool-size: 50
41.       connection-timeout: 30000
42.       idle-timeout: 600000
43.       leak-detection-threshold: 0
44.       initialization-fail-timeout: 1
45.   quartz:
46.     auto-startup: false
47.     job-store-type: jdbc
48.     jdbc:
49.       initialize-schema: never
50.     properties:
51.       org.quartz.threadPool:threadPriority: 5
52.       org.quartz.jobStore.isClustered: true
53.       org.quartz.jobStore.class: org.quartz.impl.jdbcjobstore.JobStoreTX
54.       org.quartz.scheduler.instanceId: AUTO
55.       org.quartz.jobStore.tablePrefix: QRTZ_
56.       org.quartz.jobStore.acquireTriggersWithinLock: true
57.       org.quartz.scheduler.instanceName: DolphinScheduler
58.       org.quartz.threadPool.class: org.quartz.simpl.SimpleThreadPool
59.       org.quartz.jobStore.useProperties: false
60.       org.quartz.threadPool.makeThreadsDaemons: true
61.       org.quartz.threadPool.threadCount: 25
62.       org.quartz.jobStore.misfireThreshold: 60000
63.       org.quartz.scheduler.makeSchedulerThreadDaemon: true
64. #      org.quartz.jobStore.driverDelegateClass: org.quartz.impl.jdbcjobstore.PostgreSQLDelegate
65.       org.quartz.jobStore.driverDelegateClass: org.quartz.impl.jdbcjobstore.StdJDBCDelegate
66.       org.quartz.jobStore.clusterCheckinInterval: 5000
68. management:
69.   endpoints:
70.     web:
71.       exposure:
72.         include: '*'
73.   metrics:
74.     tags:
75.       application: ${spring.application.name}
77. registry:
78.   type: zookeeper
79.   zookeeper:
80.     namespace: dolphinscheduler
81. #    connect-string: localhost:2181
82.     connect-string: 10.255.158.70:2181
83.     retry-policy:
84.       base-sleep-time: 60ms
85.       max-sleep: 300ms
86.       max-retries: 5
87.     session-timeout: 30s
88.     connection-timeout: 9s
89.     block-until-connected: 600ms
90.     digest: ~
92. audit:
93.   enabled: false
95. metrics:
96.   enabled: true
98. python-gateway:
99.   # Weather enable python gateway server or not. The default value is true.
100.   enabled: true
101.   # The address of Python gateway server start. Set its value to `0.0.0.0` if your Python API run in different
102.   # between Python gateway server. It could be be specific to other address like `127.0.0.1` or `localhost`
103.   gateway-server-address: 0.0.0.0
104.   # The port of Python gateway server start. Define which port you could connect to Python gateway server from
105.   # Python API side.
106.   gateway-server-port: 25333
107.   # The address of Python callback client.
108.   python-address: 127.0.0.1
109.   # The port of Python callback client.
110.   python-port: 25334
111.   # Close connection of socket server if no other request accept after x milliseconds. Define value is (0 = infinite),
112.   # and socket server would never close even though no requests accept
113.   connect-timeout: 0
114.   # Close each active connection of socket server if python program not active after x milliseconds. Define value is
115.   # (0 = infinite), and socket server would never close even though no requests accept
116.   read-timeout: 0
118. # Override by profile
120. ---
121. spring:
122.   config:
123.     activate:
124.       on-profile: mysql
125.   datasource:
126.     driver-class-name: com.mysql.jdbc.Driver
127.     url: jdbc:mysql://127.0.0.1:3306/dolphinscheduler?useUnicode=true&characterEncoding=UTF-8
128.   quartz:
129.     properties:
130.       org.quartz.jobStore.driverDelegateClass: org.quartz.impl.jdbcjobstore.StdJDBCDelegate

复制代码

‍
dolphinscheduler-master application.yaml

1. spring:
2.   banner:
3.     charset: UTF-8
4.   application:
5.     name: master-server
6.   jackson:
7.     time-zone: UTC
8.     date-format: "yyyy-MM-dd HH:mm:ss"
9.   cache:
10.     # default enable cache, you can disable by `type: none`
11.     type: none
12.     cache-names:
13.       - tenant
14.       - user
15.       - processDefinition
16.       - processTaskRelation
17.       - taskDefinition
18.     caffeine:
19.       spec: maximumSize=100,expireAfterWrite=300s,recordStats
20.   datasource:
21.     #driver-class-name: org.postgresql.Driver
22.     #url: jdbc:postgresql://127.0.0.1:5432/dolphinscheduler
23.     driver-class-name: com.mysql.jdbc.Driver
24.     url: jdbc:mysql://127.0.0.1:3306/dolphinscheduler?useUnicode=true&serverTimezone=UTC&characterEncoding=UTF-8&autoReconnect=true&useSSL=false&zeroDateTimeBehavior=convertToNull
25.     username: root
26.     password:
27.     hikari:
28.       connection-test-query: select 1
29.       minimum-idle: 5
30.       auto-commit: true
31.       validation-timeout: 3000
32.       pool-name: DolphinScheduler
33.       maximum-pool-size: 50
34.       connection-timeout: 30000
35.       idle-timeout: 600000
36.       leak-detection-threshold: 0
37.       initialization-fail-timeout: 1
38.   quartz:
39.     job-store-type: jdbc
40.     jdbc:
41.       initialize-schema: never
42.     properties:
43.       org.quartz.threadPool:threadPriority: 5
44.       org.quartz.jobStore.isClustered: true
45.       org.quartz.jobStore.class: org.quartz.impl.jdbcjobstore.JobStoreTX
46.       org.quartz.scheduler.instanceId: AUTO
47.       org.quartz.jobStore.tablePrefix: QRTZ_
48.       org.quartz.jobStore.acquireTriggersWithinLock: true
49.       org.quartz.scheduler.instanceName: DolphinScheduler
50.       org.quartz.threadPool.class: org.quartz.simpl.SimpleThreadPool
51.       org.quartz.jobStore.useProperties: false
52.       org.quartz.threadPool.makeThreadsDaemons: true
53.       org.quartz.threadPool.threadCount: 25
54.       org.quartz.jobStore.misfireThreshold: 60000
55.       org.quartz.scheduler.makeSchedulerThreadDaemon: true
56. #      org.quartz.jobStore.driverDelegateClass: org.quartz.impl.jdbcjobstore.PostgreSQLDelegate
57.       org.quartz.jobStore.driverDelegateClass: org.quartz.impl.jdbcjobstore.StdJDBCDelegate
58.       org.quartz.jobStore.clusterCheckinInterval: 5000
60. registry:
61.   type: zookeeper
62.   zookeeper:
63.     namespace: dolphinscheduler
64. #    connect-string: localhost:2181
65.     connect-string: 10.255.158.70:2181
66.     retry-policy:
67.       base-sleep-time: 60ms
68.       max-sleep: 300ms
69.       max-retries: 5
70.     session-timeout: 30s
71.     connection-timeout: 9s
72.     block-until-connected: 600ms
73.     digest: ~
75. master:
76.   listen-port: 5678
77.   # master fetch command num
78.   fetch-command-num: 10
79.   # master prepare execute thread number to limit handle commands in parallel
80.   pre-exec-threads: 10
81.   # master execute thread number to limit process instances in parallel
82.   exec-threads: 100
83.   # master dispatch task number per batch
84.   dispatch-task-number: 3
85.   # master host selector to select a suitable worker, default value: LowerWeight. Optional values include random, round_robin, lower_weight
86.   host-selector: lower_weight
87.   # master heartbeat interval, the unit is second
88.   heartbeat-interval: 10
89.   # master commit task retry times
90.   task-commit-retry-times: 5
91.   # master commit task interval, the unit is millisecond
92.   task-commit-interval: 1000
93.   state-wheel-interval: 5
94.   # 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
95.   max-cpu-load-avg: -1
96.   # master reserved memory, only lower than system available memory, master server can schedule. default value 0.3, the unit is G
97.   reserved-memory: 0.3
98.   # failover interval, the unit is minute
99.   failover-interval: 10
100.   # kill yarn jon when failover taskInstance, default true
101.   kill-yarn-job-when-task-failover: true
103. server:
104.   port: 5679
106. management:
107.   endpoints:
108.     web:
109.       exposure:
110.         include: '*'
111.   metrics:
112.     tags:
113.       application: ${spring.application.name}
115. metrics:
116.   enabled: true
118. # Override by profile
120. ---
121. spring:
122.   config:
123.     activate:
124.       on-profile: mysql
125.   datasource:
126.     driver-class-name: com.mysql.jdbc.Driver
127.     url: jdbc:mysql://127.0.0.1:3306/dolphinscheduler?useUnicode=true&serverTimezone=UTC&characterEncoding=UTF-8&autoReconnect=true&useSSL=false&zeroDateTimeBehavior=convertToNull
128.   quartz:
129.     properties:
130.       org.quartz.jobStore.driverDelegateClass: org.quartz.impl.jdbcjobstore.StdJDBCDelegate

复制代码

dolphinscheduler-worker application.yaml

1. spring:
2.   banner:
3.     charset: UTF-8
4.   application:
5.     name: worker-server
6.   jackson:
7.     time-zone: UTC
8.     date-format: "yyyy-MM-dd HH:mm:ss"
9.   datasource:
10.     #driver-class-name: org.postgresql.Driver
11.     #url: jdbc:postgresql://127.0.0.1:5432/dolphinscheduler
12.     driver-class-name: com.mysql.jdbc.Driver
13.     url: jdbc:mysql://127.0.0.1:3306/dolphinscheduler?useUnicode=true&serverTimezone=UTC&characterEncoding=UTF-8&autoReconnect=true&useSSL=false&zeroDateTimeBehavior=convertToNull
14.     username: root
15.     #password: root
16.     password:
17.     hikari:
18.       connection-test-query: select 1
19.       minimum-idle: 5
20.       auto-commit: true
21.       validation-timeout: 3000
22.       pool-name: DolphinScheduler
23.       maximum-pool-size: 50
24.       connection-timeout: 30000
25.       idle-timeout: 600000
26.       leak-detection-threshold: 0
27.       initialization-fail-timeout: 1
29. registry:
30.   type: zookeeper
31.   zookeeper:
32.     namespace: dolphinscheduler
33. #    connect-string: localhost:2181
34.     connect-string: 10.255.158.70:2181
35.     retry-policy:
36.       base-sleep-time: 60ms
37.       max-sleep: 300ms
38.       max-retries: 5
39.     session-timeout: 30s
40.     connection-timeout: 9s
41.     block-until-connected: 600ms
42.     digest: ~
44. worker:
45.   # worker listener port
46.   listen-port: 1234
47.   # worker execute thread number to limit task instances in parallel
48.   exec-threads: 100
49.   # worker heartbeat interval, the unit is second
50.   heartbeat-interval: 10
51.   # worker host weight to dispatch tasks, default value 100
52.   host-weight: 100
53.   # worker tenant auto create
54.   tenant-auto-create: true
55.   # 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
56.   max-cpu-load-avg: -1
57.   # worker reserved memory, only lower than system available memory, worker server can be dispatched tasks. default value 0.3, the unit is G
58.   reserved-memory: 0.3
59.   # default worker groups separated by comma, like 'worker.groups=default,test'
60.   groups:
61.     - default
62.   # alert server listen host
63.   alert-listen-host: localhost
64.   alert-listen-port: 50052
66. server:
67.   port: 1235
69. management:
70.   endpoints:
71.     web:
72.       exposure:
73.         include: '*'
74.   metrics:
75.     tags:
76.       application: ${spring.application.name}
78. metrics:
79.   enabled: true

复制代码

主要关注数据库，quartz, zookeeper, masker, worker配置。
2.2 API主要任务操作接口

其他业务接口可以不用关注，只需要关注最最主要的流程上线功能接口，此接口可以发散出所有的任务调度相关的代码。
接口：/dolphinscheduler/projects/{projectCode}/schedules/{id}/online；此接口会将定义的流程提交到Quartz调度框架；代码如下：
public Map setScheduleState(User loginUser,                                                 long projectCode,                                                 Integer id,                                                 ReleaseState scheduleStatus) {         Map 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 {},needn't 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 subProcessDefineCodes = new ArrayList();             processService.recurseFindSubProcess(processDefinition.getCode(), subProcessDefineCodes);             if (!subProcessDefineCodes.isEmpty()) {                 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;     }

1. public void setSchedule(int projectId, Schedule schedule) {
2.         logger.info("set schedule, project id: {}, scheduleId: {}", projectId, schedule.getId());
4.         quartzExecutor.addJob(ProcessScheduleJob.class, projectId, schedule);
5.     }

复制代码

[code]public void addJob(Class