dolphinscheduler master实现去中心化源码分析

dolphinscheduler Master服务是去中心化的，也就是没有master和slave之分，每个master都参与工作，那么它是如何每个Master服务去取任务执行时，每个Master都取到不同的任务，并且不会漏掉，不会重复的呢 ，下面从源码角度来分析这个问题

MasterServer.java

/**
 * run master server
 */
@PostConstruct
public void run() throws SchedulerException {
    ......

    this.masterSlotManager.start();

    // self tolerant

    ......
    
    this.masterSchedulerBootstrap.start();
    ......
}

在DS（后面dolphinschedule的简称）中提出了一个Slot（插槽）的概念，每个Master都有专属于自己的SlotId，当master减少或增加时，会刷新这个Id。

2. 先看SlotId的生成原理

this.masterSlotManager.start();

public void start() {
    serverNodeManager.addMasterInfoChangeListener(new MasterSlotManager.SlotChangeListener());
}

这里注册了addMasterInfoChangeListener , 从字面上分析，这里会监听卸载Zookeeper中的Master信息的变化；先分析SlotChangeListener

public class SlotChangeListener implements MasterInfoChangeListener {

    private final Lock slotLock = new ReentrantLock();

    private final MasterPriorityQueue masterPriorityQueue = new MasterPriorityQueue();

    @Override
    public void notify(Map<String, MasterHeartBeat> masterNodeInfo) {
    // 一旦master信息发生变化会走到这里，masterNodeInfo是所有master的最新信息
        List<Server> serverList = masterNodeInfo.values().stream()
                .filter(heartBeat -> !heartBeat.getServerStatus().equals(ServerStatus.BUSY))
                .map(this::convertHeartBeatToServer).collect(Collectors.toList());
        syncMasterNodes(serverList);
    }

    /**
     * sync master nodes
     */
    private void syncMasterNodes(List<Server> masterNodes) {
        slotLock.lock();
        try {
            //masterPriorityQueue 是一个阻塞优先队列
            this.masterPriorityQueue.clear();
            // 把masterNodes，会获得一个根据createTime排序的队列，这样在各个master节点中，这个优先队列中的排序顺序就都是固定一样的
            this.masterPriorityQueue.putAll(masterNodes);
            // 这里是根据当前节点的IP来获取队列中的Index来当做SlotId
            int tempCurrentSlot = masterPriorityQueue.getIndex(masterConfig.getMasterAddress());
            int tempTotalSlot = masterNodes.size();
            if (tempCurrentSlot < 0) {
                totalSlot = 0;
                currentSlot = 0;
                log.warn("Current master is not in active master list");
            } else if (tempCurrentSlot != currentSlot || tempTotalSlot != totalSlot) {
                totalSlot = tempTotalSlot;
                currentSlot = tempCurrentSlot;
                log.info("Update master nodes, total master size: {}, current slot: {}", totalSlot, currentSlot);
            }
        } finally {
            slotLock.unlock();
        }
    }
    ......
 }

从SlotChangeListener源码中，可以知道SlotId的来源，来源于Zookeeper中masterInfo信息的变化，而masterinfo信息是由master启动时主动注册上去的（临时节点，下线自动删除）。

3. 如何利用SlotId来取任务Command

MasterSchedulerBootstrap.start()

这个方法最终会执行到MasterSchedulerBootstrap的run()方法

@Override
public void run() {
    MasterServerLoadProtection serverLoadProtection = masterConfig.getServerLoadProtection();
    while (!ServerLifeCycleManager.isStopped()) {
        try {
.......
            //SlotId就在这个findCommands方法中，实现不同的master取不同的command
            List<Command> commands = findCommands();
            if (CollectionUtils.isEmpty(commands)) {
                // indicate that no command ,sleep for 1s
                Thread.sleep(Constants.SLEEP_TIME_MILLIS);
                continue;
            }

            commands.parallelStream()
                    .forEach(command -> {
                        try {
                            Optional<WorkflowExecuteRunnable> workflowExecuteRunnableOptional =
                                    workflowExecuteRunnableFactory.createWorkflowExecuteRunnable(command);
                            if (!workflowExecuteRunnableOptional.isPresent()) {
                                log.warn(
                                        "The command execute success, will not trigger a WorkflowExecuteRunnable, this workflowInstance might be in serial mode");
                                return;
                            }
                            WorkflowExecuteRunnable workflowExecuteRunnable = workflowExecuteRunnableOptional.get();
                            ProcessInstance processInstance = workflowExecuteRunnable
                                    .getWorkflowExecuteContext().getWorkflowInstance();
                            if (processInstanceExecCacheManager.contains(processInstance.getId())) {
                                log.error(
                                        "The workflow instance is already been cached, this case shouldn't be happened");
                            }
                            processInstanceExecCacheManager.cache(processInstance.getId(), workflowExecuteRunnable);
                            workflowEventQueue.addEvent(
                                    new WorkflowEvent(WorkflowEventType.START_WORKFLOW, processInstance.getId()));
                        } catch (WorkflowCreateException workflowCreateException) {
                            log.error("Master handle command {} error ", command.getId(), workflowCreateException);
                            commandService.moveToErrorCommand(command, workflowCreateException.toString());
                        }
                    });
            MasterServerMetrics.incMasterConsumeCommand(commands.size());
        } catch (InterruptedException interruptedException) {
            log.warn("Master schedule bootstrap interrupted, close the loop", interruptedException);
            Thread.currentThread().interrupt();
            break;
        } catch (Exception e) {
            log.error("Master schedule workflow error", e);
            // sleep for 1s here to avoid the database down cause the exception boom
            ThreadUtils.sleep(Constants.SLEEP_TIME_MILLIS);
        }
    }
}

private List<Command> findCommands() throws MasterException {
    try {
        long scheduleStartTime = System.currentTimeMillis();
        // 这里拿到自己的SlotId
        int thisMasterSlot = masterSlotManager.getSlot();
        int masterCount = masterSlotManager.getMasterSize();
        if (masterCount <= 0) {
            log.warn("Master count: {} is invalid, the current slot: {}", masterCount, thisMasterSlot);
            return Collections.emptyList();
        }
        int pageSize = masterConfig.getFetchCommandNum();
        //这里通过slotId去取command
        final List<Command> result =
                commandService.findCommandPageBySlot(pageSize, masterCount, thisMasterSlot);
        if (CollectionUtils.isNotEmpty(result)) {
            long cost = System.currentTimeMillis() - scheduleStartTime;
            log.info(
                    "Master schedule bootstrap loop command success, fetch command size: {}, cost: {}ms, current slot: {}, total slot size: {}",
                    result.size(), cost, thisMasterSlot, masterCount);
            ProcessInstanceMetrics.recordCommandQueryTime(cost);
        }
        return result;
    } catch (Exception ex) {
        throw new MasterException("Master loop command from database error", ex);
    }
}

最终会走到Mapper中

<select id="queryCommandPageBySlot" resultType="org.apache.dolphinscheduler.dao.entity.Command">
    select *
    from t_ds_command
    where id % #{masterCount} = #{thisMasterSlot}
    order by process_instance_priority, id asc
        limit #{limit}
</select>

这里通过id 对master的总数取余，如果等于当前的SlotId，则取出，实现多master取同一张表中的command，而master之间相互不冲突