调度系统架构设计

在对系统架构说明之前，我们先来认识一下调度系统常用的名词

1.名词解释

DAG： 全称Directed Acyclic Graph，简称DAG。工作流中的Task任务以有向无环图的形式组装起来，从入度为零的节点进行拓扑遍历，直到无后继节点为止。举例如下图：

dag示例

系统架构图

2.2 架构说明

• MasterServer

  MasterServer采用分布式无中心设计理念，MasterServer主要负责 DAG 任务切分、任务提交监控，并同时监听其它MasterServer和WorkerServer的健康状态。 MasterServer服务启动时向Zookeeper注册临时节点，通过监听Zookeeper临时节点变化来进行容错处理。

  该服务内主要包含:

  • Distributed Quartz分布式调度组件，主要负责定时任务的启停操作，当quartz调起任务后，Master内部会有线程池具体负责处理任务的后续操作

  • MasterSchedulerThread是一个扫描线程，定时扫描数据库中的 command 表，根据不同的命令类型进行不同的业务操作

  • MasterExecThread主要是负责DAG任务切分、任务提交监控、各种不同命令类型的逻辑处理

  • MasterTaskExecThread主要负责任务的持久化

• WorkerServer

  WorkerServer也采用分布式无中心设计理念，WorkerServer主要负责任务的执行和提供日志服务。WorkerServer服务启动时向Zookeeper注册临时节点，并维持心跳。

  该服务包含：

  • FetchTaskThread主要负责不断从Task Queue中领取任务，并根据不同任务类型调用TaskScheduleThread对应执行器。

  • LoggerServer是一个RPC服务，提供日志分片查看、刷新和下载等功能

• ZooKeeper

  ZooKeeper服务，系统中的MasterServer和WorkerServer节点都通过ZooKeeper来进行集群管理和容错。另外系统还基于ZooKeeper进行事件监听和分布式锁。 我们也曾经基于Redis实现过队列，不过我们希望EasyScheduler依赖到的组件尽量地少，所以最后还是去掉了Redis实现。

• Task Queue

  提供任务队列的操作，目前队列也是基于Zookeeper来实现。由于队列中存的信息较少，不必担心队列里数据过多的情况，实际上我们压测过百万级数据存队列，对系统稳定性和性能没影响。

• Alert

  提供告警相关接口，接口主要包括告警两种类型的告警数据的存储、查询和通知功能。其中通知功能又有邮件通知和SNMP(暂未实现)两种。

• API

  API接口层，主要负责处理前端UI层的请求。该服务统一提供RESTful api向外部提供请求服务。 接口包括工作流的创建、定义、查询、修改、发布、下线、手工启动、停止、暂停、恢复、从该节点开始执行等等。

• UI

  系统的前端页面，提供系统的各种可视化操作界面，详见使用手册部分。

2.3 架构设计思想

一、去中心化vs中心化

中心化思想

中心化的设计理念比较简单，分布式集群中的节点按照角色分工，大体上分为两种角色：

1. EasyScheduler中Scheduler线程分布式锁实现流程图：

   

三、线程不足循环等待问题

• 如果一个DAG中没有子流程，则如果Command中的数据条数大于线程池设置的阈值，则直接流程等待或失败。

• 如果一个大的DAG中嵌套了很多子流程，如下图则会产生“死等”状态：

  

  上图中MainFlowThread等待SubFlowThread1结束，SubFlowThread1等待SubFlowThread2结束， SubFlowThread2等待SubFlowThread3结束，而SubFlowThread3等待线程池有新线程，则整个DAG流程不能结束，从而其中的线程也不能释放。这样就形成的子父流程循环等待的状态。此时除非启动新的Master来增加线程来打破这样的”僵局”，否则调度集群将不能再使用。

对于启动新Master来打破僵局，似乎有点差强人意，于是我们提出了以下三种方案来降低这种风险：

1. 计算所有Master的线程总和，然后对每一个DAG需要计算其需要的线程数，也就是在DAG流程执行之前做预计算。因为是多Master线程池，所以总线程数不太可能实时获取。
2. 对单Master线程池进行判断，如果线程池已经满了，则让线程直接失败。
3. 增加一种资源不足的Command类型，如果线程池不足，则将主流程挂起。这样线程池就有了新的线程，可以让资源不足挂起的流程重新唤醒执行。

注意：Master Scheduler线程在获取Command的时候是FIFO的方式执行的。

于是我们选择了第三种方式来解决线程不足的问题。

四、容错设计

容错分为服务宕机容错和任务重试，服务宕机容错又分为Master容错和Worker容错两种情况

1. 宕机容错

服务容错设计依赖于ZooKeeper的Watcher机制，实现原理如图：

    - 任务的优先级也分为5级，依次为HIGHEST、HIGH、MEDIUM、LOW、LOWEST。如下图
        <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的轻量级性，所以选择了gRPC实现远程访问日志信息。

  

• 我们使用自定义Logback的FileAppender和Filter功能，实现每个任务实例生成一个日志文件。

• FileAppender主要实现如下：

  /**
  * 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： taskThreadName-processDefineId_processInstanceId_taskInstanceId
        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实现如下：

  /**
  *  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;
    }
  }
  

总结

本文从调度出发，初步介绍了大数据分布式工作流调度系统--EasyScheduler的架构原理及实现思路。后续会补充