2019-03-29 16:38:35 +08:00
## 系统架构设计
在对调度系统架构说明之前，我们先来认识一下调度系统常用的名词
2019-03-28 19:44:41 +08:00
### 1.名词解释
**DAG： ， ， ， ：
< p align = "center" >
< img src = "https://analysys.github.io/EasyScheduler/zh_CN/images/dag_examples_cn.jpg" alt = "dag示例" width = "60%" / >
< p align = "center" >
< em > dag示例< / em >
< / p >
< / p >
2019-03-28 20:21:04 +08:00
2019-03-28 19:44:41 +08:00
**流程定义**：通过拖拽任务节点并建立任务节点的关联所形成的可视化**DAG**
**流程实例**：流程实例是流程定义的实例化，可以通过手动启动或定时调度生成
**任务实例**：任务实例是流程定义中任务节点的实例化，标识着具体的任务执行状态
**任务类型**： ， ， ： **SUB_PROCESS** 也是一个单独的流程定义，是可以单独启动执行的
2019-03-28 20:21:04 +08:00
**调度方式：** 系统支持基于cron表达式的定时调度和手动调度。命令类型支持： **恢复被容错的工作流** 和 **恢复等待线程** 两种命令类型是由调度内部控制使用，外部无法调用
2019-03-28 19:44:41 +08:00
**定时调度**：系统采用 **quartz** 分布式调度器，
**依赖**：系统不单单支持 **DAG** 简单的前驱和后继节点之间的依赖，同时还提供**任务依赖**节点，支持**流程间的自定义任务依赖**
**优先级** ：支持流程实例和任务实例的优先级，如果流程实例和任务实例的优先级不设置，则默认是先进先出
**邮件告警**：支持 **SQL任务** 查询结果邮件发送，流程实例运行结果邮件告警及容错告警通知
**失败策略**：对于并行运行的任务，如果有任务失败，提供两种失败策略处理方式，**继续**是指不管并行运行任务的状态，直到流程失败结束。**结束**是指一旦发现失败任务， ，
**补数**：补历史数据，支持**区间并行和串行**两种补数方式
### 2.系统架构
#### 2.1 系统架构图
< p align = "center" >
< img src = "https://analysys.github.io/EasyScheduler/zh_CN/images/architecture.jpg" alt = "系统架构图" width = "70%" / >
< p align = "center" >
< em > 系统架构图< / em >
< / p >
< / p >
#### 2.2 架构说明
* **MasterServer**
MasterServer采用分布式无中心设计理念， ，
MasterServer服务启动时向Zookeeper注册临时节点，
##### 该服务内主要包含:
- **Distributed Quartz**分布式调度组件， ， ，
- **MasterSchedulerThread**是一个扫描线程，定时扫描数据库中的 **command** 表，根据不同的**命令类型**进行不同的业务操作
- **MasterExecThread**主要是负责DAG任务切分、任务提交监控、各种不同命令类型的逻辑处理
- **MasterTaskExecThread**主要负责任务的持久化
* **WorkerServer**
WorkerServer也采用分布式无中心设计理念， ，
##### 该服务包含：
- **FetchTaskThread**主要负责不断从**Task Queue**中领取任务，并根据不同任务类型调用**TaskScheduleThread**对应执行器。
- **LoggerServer**是一个RPC服务，
* **ZooKeeper**
ZooKeeper服务，
我们也曾经基于Redis实现过队列， ，
* **Task Queue**
提供任务队列的操作， ， ， ，
* **Alert**
提供告警相关接口，接口主要包括**告警**两种类型的告警数据的存储、查询和通知功能。其中通知功能又有**邮件通知**和**SNMP(暂未实现)**两种。
* **API**
API接口层，
接口包括工作流的创建、定义、查询、修改、发布、下线、手工启动、停止、暂停、恢复、从该节点开始执行等等。
* **UI**
2019-03-29 16:38:35 +08:00
系统的前端页面，提供系统的各种可视化操作界面，详见**[系统使用手册](系统使用手册.md)**部分。
2019-03-28 19:44:41 +08:00
#### 2.3 架构设计思想
##### 一、去中心化vs中心化
2019-03-28 20:13:51 +08:00
###### 中心化思想
2019-03-28 19:44:41 +08:00
中心化的设计理念比较简单，分布式集群中的节点按照角色分工，大体上分为两种角色：
< p align = "center" >
< img src = "https://analysys.github.io/EasyScheduler/zh_CN/images/master_slave.png" alt = "master-slave角色" width = "50%" / >
< / p >
2019-03-28 20:13:51 +08:00
2019-03-28 19:44:41 +08:00
- Master的角色主要负责任务分发并监督Slave的健康状态， ，
- Worker的角色主要负责任务的执行工作并维护和Master的心跳，
中心化思想设计存在的问题：
- 一旦Master出现了问题， ， ， ， ， ，
- 另外一个问题是如果Scheduler在Master上， ， ， ， ，
###### 去中心化
< p align = "center"
< img src = "https://analysys.github.io/EasyScheduler/zh_CN/images/decentralization.png" alt = "去中心化" width = "50%" / >
< / p >
2019-03-28 20:17:30 +08:00
2019-03-28 19:44:41 +08:00
- 在去中心化设计里， ， ， ， ， ，
- 去中心化设计的核心设计在于整个分布式系统中不存在一个区别于其他节点的”管理者”，因此不存在单点故障问题。但由于不存在” 管理者”节点所以每个节点都需要跟其他节点通信才得到必须要的机器信息，而分布式系统通信的不可靠行，则大大增加了上述功能的实现难度。
- 实际上，真正去中心化的分布式系统并不多见。反而动态中心化分布式系统正在不断涌出。在这种架构下，集群中的管理者是被动态选择出来的，而不是预置的，并且集群在发生故障的时候，集群的节点会自发的举行"会议"来选举新的"管理者"去主持工作。最典型的案例就是ZooKeeper及Go语言实现的Etcd。
- EasyScheduler的去中心化是Master/Worker注册到Zookeeper中， ，
##### 二、分布式锁实践
EasyScheduler使用ZooKeeper分布式锁来实现同一时刻只有一台Master执行Scheduler，
1. 获取分布式锁的核心流程算法如下
< p align = "center" >
< img src = "https://analysys.github.io/EasyScheduler/zh_CN/images/distributed_lock.png" alt = "获取分布式锁流程" width = "50%" / >
< / p >
2. EasyScheduler中Scheduler线程分布式锁实现流程图：
< p align = "center" >
< img src = "https://analysys.github.io/EasyScheduler/zh_CN/images/distributed_lock_procss.png" alt = "获取分布式锁流程" width = "50%" / >
< / p >
##### 三、线程不足循环等待问题
- 如果一个DAG中没有子流程， ，
- 如果一个大的DAG中嵌套了很多子流程， ：
< p align = "center" >
< img src = "https://analysys.github.io/EasyScheduler/zh_CN/images/lack_thread.png" alt = "线程不足循环等待问题" width = "50%" / >
< / p >
上图中MainFlowThread等待SubFlowThread1结束， ， ， ， ， ，
对于启动新Master来打破僵局， ， ：
1. 计算所有Master的线程总和， ， ，
2. 对单Master线程池进行判断， ，
3. 增加一种资源不足的Command类型， ， ，
注意：
于是我们选择了第三种方式来解决线程不足的问题。
##### 四、容错设计
容错分为服务宕机容错和任务重试，
2019-03-28 20:17:30 +08:00
###### 1. 宕机容错
2019-03-28 19:44:41 +08:00
服务容错设计依赖于ZooKeeper的Watcher机制， ：
< p align = "center" >
< img src = "https://analysys.github.io/EasyScheduler/zh_CN/images/fault-tolerant.png" alt = "EasyScheduler容错设计" width = "40%" / >
< / p >
其中Master监控其他Master和Worker的目录， ，
- Master容错流程图：
< p align = "center" >
< img src = "https://analysys.github.io/EasyScheduler/zh_CN/images/fault-tolerant_master.png" alt = "Master容错流程图" width = "40%" / >
< / p >
ZooKeeper Master容错完成之后则重新由EasyScheduler中Scheduler线程调度， ， ， ， ，
- Worker容错流程图：
< p align = "center" >
< img src = "https://analysys.github.io/EasyScheduler/zh_CN/images/fault-tolerant_worker.png" alt = "Worker容错流程图" width = "40%" / >
< / p >
2019-03-29 16:38:35 +08:00
2019-03-28 19:44:41 +08:00
Master Scheduler线程一旦发现任务实例为” 需要容错”状态，则接管任务并进行重新提交。
2019-03-29 16:38:35 +08:00
注意：由于” 网络抖动”可能会使得节点短时间内失去和ZooKeeper的心跳， ， ， ，
2019-03-28 19:44:41 +08:00
###### 2.任务失败重试
这里首先要区分任务失败重试、流程失败恢复、流程失败重跑的概念：
- 任务失败重试是任务级别的， ， ，
- 流程失败恢复是流程级别的，是手动进行的，恢复是从只能**从失败的节点开始执行**或**从当前节点开始执行**
- 流程失败重跑也是流程级别的，是手动进行的，重跑是从开始节点进行
接下来说正题，我们将工作流中的任务节点分了两种类型。
- 一种是业务节点， ， ，
- 还有一种是逻辑节点，这种节点不做实际的脚本或语句处理，只是整个流程流转的逻辑处理，比如子流程节等。
每一个**业务节点**都可以配置失败重试的次数，当该任务节点失败，会自动重试，直到成功或者超过配置的重试次数。**逻辑节点**不支持失败重试。但是逻辑节点里的任务支持重试。
如果工作流中有任务失败达到最大重试次数，工作流就会失败停止，失败的工作流可以手动进行重跑操作或者流程恢复操作
##### 五、任务优先级设计
在早期调度设计中，如果没有优先级设计，采用公平调度设计的话，会遇到先行提交的任务可能会和后继提交的任务同时完成的情况，而不能做到设置流程或者任务的优先级，因此我们对此进行了重新设计，目前我们设计如下：
- 按照**不同流程实例优先级**优先于**同一个流程实例优先级**优先于**同一流程内任务优先级**优先于**同一流程内任务**提交顺序依次从高到低进行任务处理。
- 具体实现是根据任务实例的json解析优先级， ， ，
- 其中流程定义的优先级是考虑到有些流程需要先于其他流程进行处理， ， ，
< p align = "center" >
< img src = "https://analysys.github.io/EasyScheduler/zh_CN/images/process_priority.png" alt = "流程优先级配置" width = "40%" / >
< / p >
- 任务的优先级也分为5级，
< p align = "center" >
< img src = "https://analysys.github.io/EasyScheduler/zh_CN/images/task_priority.png" alt = "任务优先级配置" width = "35%" / >
< / p >
##### 六、Logback和gRPC实现日志访问
- 由于Web(UI)和Worker不一定在同一台机器上， ：
- 将日志放到ES搜索引擎上
- 通过gRPC通信获取远程日志信息
- 介于考虑到尽可能的EasyScheduler的轻量级性，
< p align = "center" >
< img src = "https://analysys.github.io/EasyScheduler/zh_CN/images/grpc.png" alt = "grpc远程访问" width = "50%" / >
< / p >
- 我们使用自定义Logback的FileAppender和Filter功能，
- FileAppender主要实现如下：
```java
/**
* task log appender
*/
public class TaskLogAppender extends FileAppender< ILoggingEvent {
...
@Override
protected void append(ILoggingEvent event) {
if (currentlyActiveFile == null){
currentlyActiveFile = getFile();
}
String activeFile = currentlyActiveFile;
// thread name：
String threadName = event.getThreadName();
String[] threadNameArr = threadName.split("-");
// logId = processDefineId_processInstanceId_taskInstanceId
String logId = threadNameArr[1];
...
super.subAppend(event);
}
}
```
以/流程定义id/流程实例id/任务实例id.log的形式生成日志
- 过滤匹配以TaskLogInfo开始的线程名称：
- TaskLogFilter实现如下：
```java
/**
* task log filter
*/
public class TaskLogFilter extends Filter< ILoggingEvent {
@Override
public FilterReply decide(ILoggingEvent event) {
if (event.getThreadName().startsWith("TaskLogInfo-")){
return FilterReply.ACCEPT;
}
return FilterReply.DENY;
}
}
```
### 总结
2019-03-29 16:38:35 +08:00
本文从调度出发，初步介绍了大数据分布式工作流调度系统--EasyScheduler的架构原理及实现思路。未完待续
2019-03-28 19:44:41 +08:00
