Apache DolphinScheduler-1.3.9源码分析（一）

引言

随着大数据的发展，任务调度系统成为了数据处理和管理中至关重要的部分。Apache DolphinScheduler 是一款优秀的开源分布式工作流调度平台，在大数据场景中得到广泛应用。

在本文中，我们将对 Apache DolphinScheduler 1.3.9 版本的源码进行深入分析，介绍 Master 启动以及调度流程。

通过这些分析，开发者可以更好地理解 DolphinScheduler 的工作机制，并在实际使用中更高效地进行二次开发或优化。

Master Server启动

启动流程图

Master调度工作流流程图

MasterServer启动方法

public void run() {
    // init remoting server
    NettyServerConfig serverConfig = new NettyServerConfig();
    serverConfig.setListenPort(masterConfig.getListenPort());
    this.nettyRemotingServer = new NettyRemotingServer(serverConfig);
    this.nettyRemotingServer.registerProcessor(CommandType.TASK_EXECUTE_RESPONSE, new TaskResponseProcessor());
    this.nettyRemotingServer.registerProcessor(CommandType.TASK_EXECUTE_ACK, new TaskAckProcessor());
    this.nettyRemotingServer.registerProcessor(CommandType.TASK_KILL_RESPONSE, new TaskKillResponseProcessor());
    this.nettyRemotingServer.start();

    // self tolerant
    this.zkMasterClient.start();
    this.zkMasterClient.setStoppable(this);

    // scheduler start
    this.masterSchedulerService.start();

    // start QuartzExecutors
    // what system should do if exception
    try {
        logger.info("start Quartz server...");
        QuartzExecutors.getInstance().start();
    } catch (Exception e) {
        try {
            QuartzExecutors.getInstance().shutdown();
        } catch (SchedulerException e1) {
            logger.error("QuartzExecutors shutdown failed : " + e1.getMessage(), e1);
        }
        logger.error("start Quartz failed", e);
    }

    /**
     * register hooks, which are called before the process exits
     */
    Runtime.getRuntime().addShutdownHook(new Thread(() -> {
        if (Stopper.isRunning()) {
            close("shutdownHook");
        }
    }));

}

• nettyServer会注册三种Command

1. TASK_EXECUTE_ACK：Worker在接收到Master执行任务的请求后，会给Master发送一条Ack Command，告诉Master已经开始执行Task了。
2. TASK_EXECUTE_RESPONSE：Worker在执行完Task之后，会给Master发送一条Response Command，告诉Master任务调度/执行结果。
3. TASK_KILL_RESPONSE：Master接收到Task停止的请求会，会给Worker发送TASK_KILL_REQUEST Command，之后Worker会把Task_KILL_RESPONSE Command返回给Master。

• 启动调度和定时器。
• 添加ShutdownHook，关闭资源。

Master 配置文件

master.listen.port=5678

# 限制Process Instance并发调度的线程数
master.exec.threads=100

# 限制每个ProcessInstance可以执行的任务数
master.exec.task.num=20

# 每一批次可以分发的任务数
master.dispatch.task.num=3

# master需要选择一个稳定的worker去执行任务
# 算法有：Random，RoundRobin，LowerWeight。默认是LowerWeight
master.host.selector=LowerWeight

# master需要向Zookeeper发送心跳，单位：秒
master.heartbeat.interval=10

# master提交任务失败，重试次数
master.task.commit.retryTimes=5

# master提交任务失败，重试时间间隔
master.task.commit.interval=1000

# master最大cpu平均负载，只有当系统cpu平均负载还没有达到这个值，master才能调度任务
# 默认值为-1，系统cpu核数 * 2
master.max.cpuload.avg=-1

# master为其他进程保留内存，只有当系统可用内存大于这个值，master才能调度
# 默认值0.3G
master.reserved.memory=0.3

Master Scheduler启动

MasterSchedulerService初始化方法

public void init(){
    // masterConfig.getMasterExecThreads()，master.properties里master.exec.threads=100
    // 该线程池的核心线程数和最大线程数为100
    this.masterExecService = (ThreadPoolExecutor)ThreadUtils.newDaemonFixedThreadExecutor("Master-Exec-Thread", masterConfig.getMasterExecThreads());
    NettyClientConfig clientConfig = new NettyClientConfig();
    this.nettyRemotingClient = new NettyRemotingClient(clientConfig);
}

MasterSchedulerService启动方法

public void run() {
    logger.info("master scheduler started");
    while (Stopper.isRunning()){
        try {
            // 这个方法是用来检查master cpu load和memory，判断master是否还有资源进行调度
            // 如果不能调度，Sleep 1 秒种
            boolean runCheckFlag = OSUtils.checkResource(masterConfig.getMasterMaxCpuloadAvg(), masterConfig.getMasterReservedMemory());
            if(!runCheckFlag) {
                Thread.sleep(Constants.SLEEP_TIME_MILLIS);
                continue;
            }
            if (zkMasterClient.getZkClient().getState() == CuratorFrameworkState.STARTED) {
                // 这里才是真正去执行调度的方法
                scheduleProcess();
            }
        } catch (Exception e) {
            logger.error("master scheduler thread error", e);
        }
    }
}

MasterSchedulerService调度方法

private void scheduleProcess() throws Exception {
    InterProcessMutex mutex = null;
    try {
        // 阻塞式获取分布式锁
        mutex = zkMasterClient.blockAcquireMutex();
        // 获取线程池的活跃线程数
        int activeCount = masterExecService.getActiveCount();
        // make sure to scan and delete command  table in one transaction
        // 获取其中一个command，必须保证操作都在一个事务里
        Command command = processService.findOneCommand();
        if (command != null) {
            logger.info("find one command: id: {}, type: {}", command.getId(),command.getCommandType());

            try{
                // 获取ProcessInstance，
                // 这个方法会根据master.exec.threads配置和活跃线程数来判断是否可以调度processInstance
                ProcessInstance processInstance = processService.handleCommand(logger,
                        getLocalAddress(),
                        this.masterConfig.getMasterExecThreads() - activeCount, command);
                if (processInstance != null) {
                    logger.info("start master exec thread , split DAG ...");
                    masterExecService.execute(
                    new MasterExecThread(
                            processInstance
                            , processService
                            , nettyRemotingClient
                            ));
                }
            }catch (Exception e){
                logger.error("scan command error ", e);
                processService.moveToErrorCommand(command, e.toString());
            }
        } else{
            //indicate that no command ,sleep for 1s
            Thread.sleep(Constants.SLEEP_TIME_MILLIS);
        }
    } finally{
        // 释放分布式锁
        zkMasterClient.releaseMutex(mutex);
    }
}

ProcessService处理Command的方法

public ProcessInstance handleCommand(Logger logger, String host, int validThreadNum, Command command) {
    // 这里是去构造ProcessInstance
    ProcessInstance processInstance = constructProcessInstance(command, host);
    //cannot construct process instance, return null;
    if(processInstance == null){
        logger.error("scan command, command parameter is error: {}", command);
        moveToErrorCommand(command, "process instance is null");
        return null;
    }
    // 这里是检测当前剩余线程数是否大于等于该ProcessDefinition及其所有子Process的数量
    // 如果检测不通过，process instance的状态变为wait thread.并且返回空的process instance
    if(!checkThreadNum(command, validThreadNum)){
        logger.info("there is not enough thread for this command: {}", command);
        return setWaitingThreadProcess(command, processInstance);
    }
    processInstance.setCommandType(command.getCommandType());
    processInstance.addHistoryCmd(command.getCommandType());
    saveProcessInstance(processInstance);
    this.setSubProcessParam(processInstance);
    delCommandByid(command.getId());
    return processInstance;
}

MasterExecThread初始化方法

public MasterExecThread(ProcessInstance processInstance, ProcessService processService, NettyRemotingClient nettyRemotingClient){
    this.processService = processService;

    this.processInstance = processInstance;
    this.masterConfig = SpringApplicationContext.getBean(MasterConfig.class);
    // master.properties文件里的master.task.exec.num
    int masterTaskExecNum = masterConfig.getMasterExecTaskNum();
    this.taskExecService = ThreadUtils.newDaemonFixedThreadExecutor("Master-Task-Exec-Thread",
                                                                    masterTaskExecNum);
    this.nettyRemotingClient = nettyRemotingClient;
}

MasterExecThread启动方法

public void run() {
    // 省略...
    try {
        if (processInstance.isComplementData() &&  Flag.NO == processInstance.getIsSubProcess()){
            // 补数逻辑... 暂不看
            executeComplementProcess();
        }else{
            // 执行task方法
            executeProcess();
        }
    }catch (Exception e){
        logger.error("master exec thread exception", e);
        logger.error("process execute failed, process id:{}", processInstance.getId());
        processInstance.setState(ExecutionStatus.FAILURE);
        processInstance.setEndTime(new Date());
        processService.updateProcessInstance(processInstance);
    }finally {
        taskExecService.shutdown();
    }
}


private void executeProcess() throws Exception {
    // 前置
    prepareProcess();
    // 执行
    runProcess();
    // 后置
    endProcess();
}

private void runProcess(){
    // 从根task开始提交
    submitPostNode(null);
    boolean sendTimeWarning = false;
    while(!processInstance.isProcessInstanceStop() && Stopper.isRunning()){
        // 省略部分代码...
        
        // 根据cpu load avg和Memorry判断是否可以调度
        if(canSubmitTaskToQueue()){
            submitStandByTask();
        }
        try {
            Thread.sleep(Constants.SLEEP_TIME_MILLIS);
        } catch (InterruptedException e) {
            logger.error(e.getMessage(),e);
        }
        updateProcessInstanceState();
    }

    logger.info("process:{} end, state :{}", processInstance.getId(), processInstance.getState());
}

// 获取可以并行的task
/**
* task 1 -> task 2 -> task3
* task 4 -> task 5
* task 6
* task 1,task4,task6可以并行跑
*/
private void submitPostNode(String parentNodeName){
    Set<String> submitTaskNodeList = DagHelper.parsePostNodes(parentNodeName, skipTaskNodeList, dag, completeTaskList);
    List<TaskInstance> taskInstances = new ArrayList<>();
    for(String taskNode : submitTaskNodeList){
        taskInstances.add(createTaskInstance(processInstance, taskNode,
                dag.getNode(taskNode)));
    }
    // if previous node success , post node submit
    for(TaskInstance task : taskInstances){
        if(readyToSubmitTaskQueue.contains(task)){
            continue;
        }
        if(completeTaskList.containsKey(task.getName())){
            logger.info("task {} has already run success", task.getName());
            continue;
        }
        if(task.getState().typeIsPause() || task.getState().typeIsCancel()){
            logger.info("task {} stopped, the state is {}", task.getName(), task.getState());
        }else{
            // task添加到priorityQueue
            addTaskToStandByList(task);
        }
    }
}

/**
 * handling the list of tasks to be submitted
 */
private void submitStandByTask(){

    try {
        int length = readyToSubmitTaskQueue.size();
        for (int i=0;i<length;i++) {
            // 从队列里面取task, 提交给worker执行
            TaskInstance task = readyToSubmitTaskQueue.peek();
            // 先判断task的前置依赖有没有都运行成功，如果运行成功，在提交该task运行

            // 如果运行失败，或者没有执行，则不提交
            DependResult dependResult = getDependResultForTask(task);
            if(DependResult.SUCCESS == dependResult){
                if(retryTaskIntervalOverTime(task)){
                    submitTaskExec(task);
                    removeTaskFromStandbyList(task);
                }
            }else if(DependResult.FAILED == dependResult){
                // if the dependency fails, the current node is not submitted and the state changes to failure.
                dependFailedTask.put(task.getName(), task);
                removeTaskFromStandbyList(task);
                logger.info("task {},id:{} depend result : {}",task.getName(), task.getId(), dependResult);
            } else if (DependResult.NON_EXEC == dependResult) {
            // for some reasons(depend task pause/stop) this task would not be submit
            removeTaskFromStandbyList(task);
            logger.info("remove task {},id:{} , because depend result : {}", task.getName(), task.getId(), dependResult);
           }
        }
    } catch (Exception e) {
        logger.error("submit standby task error",e);
    }
}

/**
* 创建TaskExecThread
*/
private TaskInstance submitTaskExec(TaskInstance taskInstance) {
    MasterBaseTaskExecThread abstractExecThread = null;
    if(taskInstance.isSubProcess()){
        abstractExecThread = new SubProcessTaskExecThread(taskInstance);
    }else if(taskInstance.isDependTask()){
        abstractExecThread = new DependentTaskExecThread(taskInstance);
    }else if(taskInstance.isConditionsTask()){
        abstractExecThread = new ConditionsTaskExecThread(taskInstance);
    }else {
        abstractExecThread = new MasterTaskExecThread(taskInstance);
    }
    Future<Boolean> future = taskExecService.submit(abstractExecThread);
    activeTaskNode.putIfAbsent(abstractExecThread, future);
    return abstractExecThread.getTaskInstance();
}

MasterBaseTaskExecThread

MasterBaseTaskExecThread是SubProcessTaskExecThread，DependentTaskExecThread，ConditionsTaskExecThread，MasterTaskExecThread的父类，实现Callable接口。

• SubProcessTaskExecThread

  任务实例不会下发到worker节点执行，在submitTask(TaskInstance taskInstance)方法中，针对子流程，会增加一条子流程实例命令，然后在waitTaskQuit方法中循环等待子流程执行完成。在当前工作流运行结束后会继续运行子工作流并做相关状态更新，子工作流完全完成才同步状态为子工作流的状态。

• DependentTaskExecThread

  Dependent 节点，就是依赖检查节点。比如 A 流程依赖昨天的 B 流程执行成功，依赖节点会去检查 B 流程在昨天是否有执行成功的实例。

• ConditionsTaskExecThrea

  Conditions 是一个条件节点，根据上游任务运行状态，判断应该运行哪个下游任务。截止目前 Conditions 支持多个上游任务，但只支持两个下游任务。当上游任务数超过一个时，可以通过且以及或操作符实现复杂上游依赖。

• MasterTaskExecThread

  将任务实例下发到worker节点执行，并在waitTaskQuit方法中循环等待任务实例执行完成，任务完成后则即出。例如SQKL，Shell等任务类型。

MasterBaseTaskExecThread初始化方法

public MasterBaseTaskExecThread(TaskInstance taskInstance){
    this.processService = SpringApplicationContext.getBean(ProcessService.class);
    this.alertDao = SpringApplicationContext.getBean(AlertDao.class);
    this.cancel = false;
    this.taskInstance = taskInstance;
    this.masterConfig = SpringApplicationContext.getBean(MasterConfig.class);
    this.taskUpdateQueue = SpringApplicationContext.getBean(TaskPriorityQueueImpl.class);
    initTaskParams();
}

MasterBaseTaskExecThread执行方法

@Override
public Boolean call() throws Exception {
    this.processInstance = processService.findProcessInstanceById(taskInstance.getProcessInstanceId());
    return submitWaitComplete(); // 由各子类实现
}

MasterBaseTaskExecThread公共方法

submit()

protected TaskInstance submit(){
    // 提交任务重试次数. master.task.commit.retryTimes=5
    Integer commitRetryTimes = masterConfig.getMasterTaskCommitRetryTimes();
    // 提交任务失败，重试间隔时间 master.task.commit.interval=1000
    Integer commitRetryInterval = masterConfig.getMasterTaskCommitInterval();

    int retryTimes = 1;
    boolean submitDB = false;
    boolean submitTask = false;
    TaskInstance task = null;
    while (retryTimes <= commitRetryTimes){
        try {
            if(!submitDB){
                // 持久化TaskInstance到数据库
                task = processService.submitTask(taskInstance);
                if(task != null && task.getId() != 0){
                    submitDB = true;
                }
            }
            if(submitDB && !submitTask){
                // 分发任务到Woroker执行
                submitTask = dispatchTask(task);
            }
            if(submitDB && submitTask){
                return task;
            }
            if(!submitDB){
                logger.error("task commit to db failed , taskId {} has already retry {} times, please check the database", taskInstance.getId(), retryTimes);
            }else if(!submitTask){
                logger.error("task commit  failed , taskId {} has already retry {} times, please check", taskInstance.getId(), retryTimes);
            }
            Thread.sleep(commitRetryInterval);
        } catch (Exception e) {
            logger.error("task commit to mysql and dispatcht task failed",e);
        }
        retryTimes += 1;
    }
    return task;
}

dispatchTask(TaskInstance task)

public Boolean dispatchTask(TaskInstance taskInstance) {
    try{
        // 如果是子流程，条件任务，依赖任务，直接返回true，不提交给worker执行
        if(taskInstance.isConditionsTask()
                || taskInstance.isDependTask()
                || taskInstance.isSubProcess()){
            return true;
        }
        if(taskInstance.getState().typeIsFinished()){
            logger.info(String.format("submit task , but task [%s] state [%s] is already  finished. ", taskInstance.getName(), taskInstance.getState().toString()));
            return true;
        }
        // task cannot submit when running
        if(taskInstance.getState() == ExecutionStatus.RUNNING_EXECUTION){
            logger.info(String.format("submit to task, but task [%s] state already be running. ", taskInstance.getName()));
            return true;
        }
        logger.info("task ready to submit: {}", taskInstance);

        /**
         *  taskPriority
         */
        TaskPriority taskPriority = buildTaskPriority(processInstance.getProcessInstancePriority().getCode(),
                                                      processInstance.getId(),
                                                      taskInstance.getProcessInstancePriority().getCode(),
                                                      taskInstance.getId(),
                                                      org.apache.dolphinscheduler.common.Constants.DEFAULT_WORKER_GROUP);
        // 放入TaskPriorityQueue中，
        // org.apache.dolphinscheduler.service.queue.TaskPriorityQueueImpl用于消费，从队列里取出TaskInstance，提交给Worker执行
        taskUpdateQueue.put(taskPriority);
        logger.info(String.format("master submit success, task : %s", taskInstance.getName()) );
        return true;
    }catch (Exception e){
        logger.error("submit task  Exception: ", e);
        logger.error("task error : %s", JSONUtils.toJson(taskInstance));
        return false;
    }
}

MasterTaskExecThread

submitWaitComplete()

public Boolean submitWaitComplete() {
    Boolean result = false;
    // 提交任务
    this.taskInstance = submit();
    if(this.taskInstance == null){
        logger.error("submit task instance to mysql and queue failed , please check and fix it");
        return result;
    }
    if(!this.taskInstance.getState().typeIsFinished()) {
        // 等待任务执行结果
        result = waitTaskQuit();
    }
    taskInstance.setEndTime(new Date());
    processService.updateTaskInstance(taskInstance);
    logger.info("task :{} id:{}, process id:{}, exec thread completed ",
                this.taskInstance.getName(),taskInstance.getId(), processInstance.getId() );
    return result;
}

waitTaskQuit()

public Boolean waitTaskQuit(){
    // query new state
    taskInstance = processService.findTaskInstanceById(taskInstance.getId());
    logger.info("wait task: process id: {}, task id:{}, task name:{} complete",
                this.taskInstance.getProcessInstanceId(), this.taskInstance.getId(), this.taskInstance.getName());

    while (Stopper.isRunning()){
        try {
            if(this.processInstance == null){
                logger.error("process instance not exists , master task exec thread exit");
                return true;
            }
            // task instance add queue , waiting worker to kill
            if(this.cancel || this.processInstance.getState() == ExecutionStatus.READY_STOP){
                cancelTaskInstance();
            }
            if(processInstance.getState() == ExecutionStatus.READY_PAUSE){
                pauseTask();
            }
            // task instance finished
            if (taskInstance.getState().typeIsFinished()){
                // if task is final result , then remove taskInstance from cache
                // taskInstanceCacheManager其实现类为：org.apache.dolphinscheduler.server.master.cache.impl.TaskInstanceCacheManagerImpl
                // taskInstance在触发ack和response Command会被添加到taskInstanceCache里
                taskInstanceCacheManager.removeByTaskInstanceId(taskInstance.getId());
                break;
            }
            if (checkTaskTimeout()) {
                this.checkTimeoutFlag = !alertTimeout();
            }
            // updateProcessInstance task instance
            taskInstance = processService.findTaskInstanceById(taskInstance.getId());
            processInstance = processService.findProcessInstanceById(processInstance.getId());
            Thread.sleep(Constants.SLEEP_TIME_MILLIS);
        } catch (Exception e) {
            logger.error("exception",e);
            if (processInstance != null) {
                logger.error("wait task quit failed, instance id:{}, task id:{}",
                             processInstance.getId(), taskInstance.getId());
            }
        }
    }
    return true;
}

SubProcessTaskExecThread

submitWaitComplete()

public Boolean submitWaitComplete() {

    Boolean result = false;
    try{
        // submit task instance
        this.taskInstance = submit();

        if(taskInstance == null){
            logger.error("sub work flow submit task instance to mysql and queue failed , please check and fix it");
            return result;
        }
        setTaskInstanceState();
        waitTaskQuit();
        subProcessInstance = processService.findSubProcessInstance(processInstance.getId(), taskInstance.getId());

        // at the end of the subflow , the task state is changed to the subflow state
        if(subProcessInstance != null){
            if(subProcessInstance.getState() == ExecutionStatus.STOP){
                this.taskInstance.setState(ExecutionStatus.KILL);
            }else{
                this.taskInstance.setState(subProcessInstance.getState());
            }
        }
        taskInstance.setEndTime(new Date());
        processService.updateTaskInstance(taskInstance);
        logger.info("subflow task :{} id:{}, process id:{}, exec thread completed ",
                    this.taskInstance.getName(),taskInstance.getId(), processInstance.getId() );
        result = true;

    }catch (Exception e){
        logger.error("exception: ",e);
        if (null != taskInstance) {
            logger.error("wait task quit failed, instance id:{}, task id:{}",
                         processInstance.getId(), taskInstance.getId());
        }
    }
    return result;
}

waitTaskQuit()

private void waitTaskQuit() throws InterruptedException {

    logger.info("wait sub work flow: {} complete", this.taskInstance.getName());
    
    if (taskInstance.getState().typeIsFinished()) {
        logger.info("sub work flow task {} already complete. task state:{}, parent work flow instance state:{}",
                    this.taskInstance.getName(),
                    this.taskInstance.getState(),
                    this.processInstance.getState());
        return;
    }
    while (Stopper.isRunning()) {
        // waiting for subflow process instance establishment
        if (subProcessInstance == null) {
            Thread.sleep(Constants.SLEEP_TIME_MILLIS);
            if(!setTaskInstanceState()){
                continue;
            }
        }
        subProcessInstance = processService.findProcessInstanceById(subProcessInstance.getId());
        if (checkTaskTimeout()) {
            this.checkTimeoutFlag = !alertTimeout();
            handleTimeoutFailed();
        }
        updateParentProcessState();
        if (subProcessInstance.getState().typeIsFinished()){
            break;
        }
        if(this.processInstance.getState() == ExecutionStatus.READY_PAUSE){
            // parent process "ready to pause" , child process "pause"
            pauseSubProcess();
        }else if(this.cancel || this.processInstance.getState() == ExecutionStatus.READY_STOP){
            // parent Process "Ready to Cancel" , subflow "Cancel"
            stopSubProcess();
        }
        Thread.sleep(Constants.SLEEP_TIME_MILLIS);
    }
}

ConditionsTaskExecThread

submitWaitComplete()

public Boolean submitWaitComplete() {
    try{
        this.taskInstance = submit();
        logger = LoggerFactory.getLogger(LoggerUtils.buildTaskId(LoggerUtils.TASK_LOGGER_INFO_PREFIX,
                                                                 taskInstance.getProcessDefinitionId(),
                                                                 taskInstance.getProcessInstanceId(),
                                                                 taskInstance.getId()));
        String threadLoggerInfoName = String.format(Constants.TASK_LOG_INFO_FORMAT, processService.formatTaskAppId(this.taskInstance));
        Thread.currentThread().setName(threadLoggerInfoName);
        initTaskParameters();
        logger.info("dependent task start");
        // 等待判断
        waitTaskQuit();
        // 更新最终依赖结果
        updateTaskState();
    }catch (Exception e){
        logger.error("conditions task run exception" , e);
    }
    return true;
}

waitTaskQuit

private void waitTaskQuit() {
    List<TaskInstance> taskInstances = processService.findValidTaskListByProcessId(
            taskInstance.getProcessInstanceId()
    );
    for(TaskInstance task : taskInstances){
        completeTaskList.putIfAbsent(task.getName(), task.getState());
    }

    // 获取所有依赖结果
    List<DependResult> modelResultList = new ArrayList<>();
    for(DependentTaskModel dependentTaskModel : dependentParameters.getDependTaskList()){

        List<DependResult> itemDependResult = new ArrayList<>();
        for(DependentItem item : dependentTaskModel.getDependItemList()){
            itemDependResult.add(getDependResultForItem(item));
        }
        DependResult modelResult = DependentUtils.getDependResultForRelation(dependentTaskModel.getRelation(), itemDependResult);
        modelResultList.add(modelResult);
    }
    // 根据逻辑运算符，合并依赖结果
    conditionResult = DependentUtils.getDependResultForRelation(
            dependentParameters.getRelation(), modelResultList
    );
    logger.info("the conditions task depend result : {}", conditionResult);
}

DependentTaskExecThread

submitWaitComplete()

public Boolean submitWaitComplete() {
    try{
        logger.info("dependent task start");
        this.taskInstance = submit();
        logger = LoggerFactory.getLogger(LoggerUtils.buildTaskId(LoggerUtils.TASK_LOGGER_INFO_PREFIX,
                taskInstance.getProcessDefinitionId(),
                taskInstance.getProcessInstanceId(),
                taskInstance.getId()));
        String threadLoggerInfoName = String.format(Constants.TASK_LOG_INFO_FORMAT, processService.formatTaskAppId(this.taskInstance));
        Thread.currentThread().setName(threadLoggerInfoName);
        initTaskParameters();
        initDependParameters();
        waitTaskQuit();
        updateTaskState();
    }catch (Exception e){
        logger.error("dependent task run exception" , e);
    }
    return true;
}

waitTaskQuit()

private Boolean waitTaskQuit() {
    logger.info("wait depend task : {} complete", this.taskInstance.getName());
    if (taskInstance.getState().typeIsFinished()) {
        logger.info("task {} already complete. task state:{}",
                this.taskInstance.getName(),
                this.taskInstance.getState());
        return true;
    }
    while (Stopper.isRunning()) {
        try{
            if(this.processInstance == null){
                logger.error("process instance not exists , master task exec thread exit");
                return true;
            }
            // 省略部分代码
    
            // allDependentTaskFinish()等待所有依赖任务执行结束
            if ( allDependentTaskFinish() || taskInstance.getState().typeIsFinished()){
                break;
            }
            // update process task
            taskInstance = processService.findTaskInstanceById(taskInstance.getId());
            processInstance = processService.findProcessInstanceById(processInstance.getId());
            Thread.sleep(Constants.SLEEP_TIME_MILLIS);
        } catch (Exception e) {
            logger.error("exception",e);
            if (processInstance != null) {
                logger.error("wait task quit failed, instance id:{}, task id:{}",
                        processInstance.getId(), taskInstance.getId());
            }
        }
    }
    return true;
}

TaskPriorityQueueConsumer

@Override
public void run() {
    List<TaskPriority> failedDispatchTasks = new ArrayList<>();
    while (Stopper.isRunning()){
        try {
            // 每一批次分发任务数量，master.dispatch.task.num = 3
            int fetchTaskNum = masterConfig.getMasterDispatchTaskNumber();
            failedDispatchTasks.clear();
            for(int i = 0; i < fetchTaskNum; i++){
                if(taskPriorityQueue.size() <= 0){
                    Thread.sleep(Constants.SLEEP_TIME_MILLIS);
                    continue;
                }
                // if not task , blocking here
                // 从队列里面获取task
                TaskPriority taskPriority = taskPriorityQueue.take();
                // 分发给worker执行
                boolean dispatchResult = dispatch(taskPriority);
                if(!dispatchResult){
                    failedDispatchTasks.add(taskPriority);
                }
            }
            if (!failedDispatchTasks.isEmpty()) {
                // 分发失败的任务，需要重新加入队列中，等待重新分发
                for (TaskPriority dispatchFailedTask : failedDispatchTasks) {
                    taskPriorityQueue.put(dispatchFailedTask);
                }
                // If there are tasks in a cycle that cannot find the worker group,
                // sleep for 1 second
                if (taskPriorityQueue.size() <= failedDispatchTasks.size()) {
                    TimeUnit.MILLISECONDS.sleep(Constants.SLEEP_TIME_MILLIS);
                }
            }
        }catch (Exception e){
            logger.error("dispatcher task error",e);
        }
    }
}


/**
 * dispatch task
 *
 * @param taskPriority taskPriority
 * @return result
 */
protected boolean dispatch(TaskPriority taskPriority) {
    boolean result = false;
    try {
        int taskInstanceId = taskPriority.getTaskId();
        TaskExecutionContext context = getTaskExecutionContext(taskInstanceId);
        ExecutionContext executionContext = new ExecutionContext(context.toCommand(), ExecutorType.WORKER, context.getWorkerGroup());

        if (taskInstanceIsFinalState(taskInstanceId)){
            // when task finish, ignore this task, there is no need to dispatch anymore
            return true;
        }else{
            // 分发任务
            // 分发算法支持：低负载优先算法，随机算法， 轮询算法。
            result = dispatcher.dispatch(executionContext);
        }
    } catch (ExecuteException e) {
        logger.error("dispatch error: {}",e.getMessage());
    }
    return result;
}

通过对 Apache DolphinScheduler 1.3.9 的源码分析，我们深入了解了其核心模块的设计和实现。DolphinScheduler 的 Master 架构充分保证了任务调度的高可用性和扩展性，而通过 Zookeeper 实现的集群协调则为系统提供了强大的容错机制。

如果你对 Apache DolphinScheduler 的源码有兴趣，可以深入研究其任务调度策略的细节部分，或者根据自身业务场景进行二次开发，充分发挥 DolphinScheduler 的调度能力。

本文完！

本文由 白鲸开源 提供发布支持！