点击蓝字，关注我们

A graph is formed by vertices and by edges connecting pairs of vertices, where the
vertices can be any kind of object that is connected in pairs by edges.
In the case of a directed graph, each edge has an orientation, from one vertex to another vertex. A path in a directed graph is a sequence of edges having the property that the ending vertex of each
edge in the sequence is the same as the starting vertex of the next edge in the sequence; a path forms a cycle if the starting vertex of its first edge equals the ending vertex of its last edge.

A directed acyclic graph is a directed graph that has no cycles.[1][2][3]

A vertex v of a directed graph is said to be reachable from another
vertex u when there exists a path that starts at u and ends at v.
As a special case, every vertex is considered to be reachable from itself (by a path with zero edges). If a vertex can reach itself via a nontrivial path (a path with one or more edges), then that path is a cycle, so another way to define directed acyclic graphs is that they are the graphs in which no vertex can reach itself via a nontrivial path.

1. vertex
   一个实体或者元素，可以是任何抽象的object
2. edge
   一条有方向直线，包含两个vertex，分别扮演起点和终点

• Dag约束

1. 在Dag中，一个edge（a,b）的终点可以作为另一个edge(b,c)的起点，这个链路中所有的vertex都是可到达的, c是从a可达的。
2. 在Dag中允许vertex不存在于任何一个edge中，这个节点可以从自己到达自己（一个孤岛，不和其他vertex有任何联系）
3. 如果一个vertex可以从自己到达自己，但是中间经过了其他的vertex，那么这就存在一个环circle
4. 在Dag中没有环

public class DAG<Node, NodeInfo, EdgeInfo> {

    private final ReadWriteLock lock = new ReentrantReadWriteLock();

    **
     * node map, key is node, value is node information
     */
    private final Map<Node, NodeInfo> nodesMap;

    **
     * edge map. key is node of origin;value is Map with key for destination node and value for edge
     */
    private final Map<Node, Map<Node, EdgeInfo>> edgesMap;

    **
     * reversed edge set，key is node of destination, value is Map with key for origin node and value for edge
     */
    private final Map<Node, Map<Node, EdgeInfo>> reverseEdgesMap;
}

• Node表示任务的id
• NodeInfo表示任务的详细信息
• EdgeInfo包含任务id和依赖任务id

Command分发流程

处理方式

生产者

{
    "commandType": "START_PROCESS",
    "processDefinitionCode": 14285512555584,
    "executorId": 1,
    "commandParam": "{}",
    "taskDependType": "TASK_POST",
    "failureStrategy": "CONTINUE",
    "warningType": "NONE",
    "startTime": 1723444881372,
    "processInstancePriority": "MEDIUM",
    "updateTime": 1723444881372,
    "workerGroup": "default",
    "tenantCode": "default",
    "environmentCode": -1,
    "dryRun": 0,
    "processInstanceId": 0,
    "processDefinitionVersion": 1,
    "testFlag": 0
}

消费者

<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>

Dag遍历执行任务

Master本地cache缓冲

public interface ProcessInstanceExecCacheManager {

    **
     * get WorkflowExecuteThread by process instance id
     *
     * @param processInstanceId processInstanceId
     * @return WorkflowExecuteThread
     */
    WorkflowExecuteRunnable getByProcessInstanceId(int processInstanceId);

    **
     * judge the process instance does it exist
     *
     * @param processInstanceId processInstanceId
     * @return true - if process instance id exists in cache
     */
    boolean contains(int processInstanceId);

    **
     * remove cache by process instance id
     *
     * @param processInstanceId processInstanceId
     */
    void removeByProcessInstanceId(int processInstanceId);

    **
     * cache
     *
     * @param processInstanceId     processInstanceId
     * @param workflowExecuteThread if it is null, will not be cached
     */
    void cache(int processInstanceId, @NonNull WorkflowExecuteRunnable workflowExecuteThread);

    **
     * get all WorkflowExecuteThread from cache
     *
     * @return all WorkflowExecuteThread in cache
     */
    Collection<WorkflowExecuteRunnable> getAll();

    void clearCache();
}

生产者

消费者

可执行任务Dispatch

Master进城内优先级队列

生产者

消费者

1. MasterTaskDispatcher
2. WorkerTaskDispatcher

任务执行状态回调通知

生产者

1. TaskExecutionStatus.FAILURE 执行抛出异常，运行失败
2. TaskExecutionStatus.RUNNING_EXECUTION 开始执行
3. TaskExecutionStatus.KILL 被杀死
4. TaskExecutionStatus.SUCCESS 执行成功

消费者

任务状态处理

缓冲队列

生产者

1. api-server用户行为
2. master节点任务调度
3. work节点任务执行
4. master任务执行

消费者