Apache DolphinScheduler v2.0.1 Master 和 Worker 执行流程分析系列（一）

海豚调度 2022-10-11

1205

点亮 ⭐️ Star · 照亮开源之路

GitHub:https://github.com/apache/dolphinscheduler

这是一系列关于DolphinScheduler v2.0.1的源码分析文章，包括对Master、Worker基本原理、Master提交执行、Worker接受执行、command解耦、command唯一消费实现原理、kill执行分析等在内的深度解析与思考。

前言

以下是基于海豚v2.0.1源码分析得出的图解：master和worker的执行流程。

后面所有的讲解都是围绕这一张图，所以在看讲解时，请参考整个图去理解。讲解思路是化整为零，积点成线，集线成面。

本文适用于对海豚有一定了解和使用经验的同学。以下简单说一些基础认识，名词叫法可能不太相同，但也应该比较好理解。

调度单元

海豚的调度粒度。海豚里有process和task两个定义，一个process有多个task，依赖关系是task级别的，而真正调度是process级别的，也就是每次调度都是调度整个process。

目前海豚是master和worker架构，分工也很明确：

master负责整个process生命周期，包括task的dag图构建，task的dag执行，选择提交worker，操作DB等；

worker负责task执行，是真正的执行器。这里需要注意，worker不操作DB！

Command解耦

海豚还有个应用层API服务，和master主要通过command解耦。比如正常调度(start_process)，重跑(repeat_running)等

但是，终止(stop)、暂停(pause)等还是通过netty直连master。

Process的事件驱动实现原理

再讲整个流程之前，必须先搞明白master如何管理process整个生命周期的所有操作。

process内有会有多个task，task会按照dag从上往下执行，整个执行时间会很长，如果为每个process开启单独线程同步执行，管理process的所有操作，这样会导致并发数受限，处理能力下降，浪费资源。

海豚则引入了事件，用于驱动线程执行。下面讲解原理（上图中事件驱动相关已经框起来了）

WorkflowExecuteThread：Master每处理一个process，都会对应实例化一个线程类。内部重要组成有（黄色框）

StateEventQueue队列：用于接收存放事件

handleEvent()方法：用户处理事件，主要有PROCESS_STATE_CHANGE,TASK_STATE_CHANGE等事件

processInstanceExecMaps：存放master端所有的WorkflowExecuteThread，key为prcocessInstacneId（红色框）

TaskResponseService和StateEventResponseService：主要有两个操作，1.接收master端所有事件；2.内置定时轮询执行线程，将事件分发到相应process对应的WorkflowExecuteThread线程StateEventQueue队列中（蓝色框）

EventExecuteService：扫描、过滤所有processInstanceExecMaps内事件大于0的WorkflowExecuteThread线程，并内置提交WorkflowExecuteThread的线程池，处理事件（绿色框）

根据以上核心组成，来梳理下每个WorkflowExecuteThread线程是如何被事件驱动的。

master为每个process创建WorkflowExecuteThread，并缓存到processInstanceExecMaps中
TaskResponseService和StateEventResponseService将接收到的事件分发到WorkflowExecuteThread的StateEventQueue队列中
EventExecuteService则不断地提交执行事件大于0的WorkflowExecuteThread，执行handleEvent()方法
不断重复2、3，实现WorkflowExecuteThread事件驱动

Bug案例分析一

2.0.1-release分支WorkflowExecuteThread类1150行，调用 this.processStateChangeHandler(stateEvent); 这个操作是直接触发process级别状态变更操作，如果process是finished(success,failed等)，那么WorkflowExecuteThread会调用endProcess正常结束，但是processInstanceExecMaps中依旧存放着WorkflowExecuteThread引用，并且因为该WorkflowExecuteThread不会再有任何事件，EventExecuteService不能提交而永远不能释放，造成内存泄漏。此外，EventExecuteService在监控到process结束后，独有的notifyProcessChanged（通知process）操作也不能被执行。

Bug解决：

修改为 this.stateEvents.add(stateEvent); 添加到事件队列，等待EventExecuteService提交。

二开建议：

所有的状态变更，都要添加到事件队列。

相关issue：

https://github.com/apache/dolphinscheduler/issues/9065

Bug案例分析二

EventExecuteService的作用是扫描，过滤提交执行WorkflowExecuteThread线程，因为是通过定时轮询执行，所以很有可能在相邻两次时间扫描到相同WorkflowExecuteThread，为了避免多次提交，所以每次轮询会把已经提交的WorkflowExecuteThread存到eventHandlerMap中。这里没有问题，但是需要清醒的认识到，一旦放进去，就表示处理中，也就是以后都不会被提交。

接下来，分析Bug的产生。

FutureCallback处理结束后会有两个结果，onSuccess和onFailure

先看onFailure，里面什么也没处理，对，没有任何处理，那么WorkflowExecuteThread将永远被标记为处理中，即使有事件被添加，事件驱动也失效，那么1）process状态将永远不变；2）内存将永远不能释放

再看onSuccess，同理，如果在 eventHandlerMap.remove(workflowExecuteThread.getKey()); 之前发生异常，那么WorkflowExecuteThread将永远被标记为处理中，即使有事件被添加，事件驱动也失效，那么1）process状态将永远不变；2）内存将永远不能释放。