2、任务编排模式：串行、并行、并发、依赖关系图、DAG（有向无环图）在异步任务中的应用

在异步任务编排中，不同的执行模式决定了任务如何被调度、资源如何分配以及错误如何传播。理解串行、并行、并发、依赖关系图与DAG（有向无环图）的核心差异，是构建可靠日志采集与错误追踪体系的基础。本节将逐一剖析这些模式，并展示它们在异步任务中的具体应用场景。

2.1 串行与并行：基础执行模式

串行执行是最直观的模式：任务按顺序一个接一个执行，前一个任务完成后，后一个任务才开始。这种模式逻辑简单，但吞吐量低，且单个任务的失败会阻塞整个链条。

并行执行则允许多个任务在同一时刻同时运行，通常依赖多线程或多进程实现。并行能显著提升吞吐量，但需要处理资源竞争与数据一致性问题。

2.2 并发：逻辑上的“同时”执行

并发与并行不同：并发是逻辑上的同时执行，通过任务切换（如协程、事件循环）在单线程内实现多任务交错运行。在异步编程中，并发是核心手段，它允许一个线程在等待I/O（如网络请求、磁盘写入）时切换到另一个任务，从而避免CPU空转。

• 并发模型：基于事件循环（Event Loop）和协程（Coroutine），如Python的asyncio。
• 优势：轻量级，上下文切换开销远小于线程切换，适合高I/O密集型任务。
• 与并行的关系：并发可以在单核上实现，而并行必须依赖多核；并发是并行的基础调度单元。

2.3 依赖关系图：任务间的逻辑约束

实际业务中，任务之间往往存在依赖关系。例如，日志聚合任务必须等待所有日志解析任务完成后才能开始。依赖关系图（Dependency Graph）用节点表示任务，用有向边表示依赖方向，直观地描述了任务间的先后约束。

依赖关系图的核心要素：

• 节点（Node）：代表一个异步任务，如“日志采集”、“格式校验”、“写入存储”。
• 有向边（Directed Edge）：从前置任务指向后置任务，表示后置任务必须在前置任务完成后才能启动。
• 入度（In-degree）：指向某节点的边数，表示该任务有多少个前置依赖。
• 出度（Out-degree）：从某节点出发的边数，表示该任务有多少个后续任务。

2.4 DAG（有向无环图）：异步任务编排的核心模型

DAG（Directed Acyclic Graph）是依赖关系图的一种特殊形式，它不允许存在任何环路。DAG是异步任务编排的事实标准，广泛应用于工作流引擎、数据管道、CI/CD流水线等场景。

DAG在异步任务中的关键特性：

• 拓扑排序：将DAG中的节点线性化，保证所有依赖关系被满足。调度器按拓扑序依次提交任务。
• 并行度控制：入度为0的节点可以同时执行，天然支持并行。
• 错误传播：某个节点失败后，其下游节点可以选择跳过、重试或进入错误处理分支。
• 可观测性：每个节点的状态（等待、运行、成功、失败）可以被独立追踪，便于日志采集与错误定位。

下面是一个典型的日志采集DAG结构，展示了任务间的依赖关系与并行执行路径：

2.5 任务编排模式对比与选型

在实际的Development环境异步任务编排中，需要根据业务场景选择合适的模式。下表总结了各模式的核心差异与适用场景：

2.6 DAG在异步任务中的实现要点

在Python中实现DAG任务编排时，通常需要关注以下关键点：

1. 任务定义：每个任务封装为可调用的异步函数（async def），并携带元数据（任务ID、依赖列表、重试策略）。
2. 拓扑排序：使用Kahn算法或DFS对DAG进行拓扑排序，确保调度顺序正确。
3. 状态管理：为每个节点维护状态（PENDING、RUNNING、SUCCESS、FAILED、SKIPPED），便于日志采集与错误追踪。
4. 调度器：基于事件循环，当某个节点的所有前置节点状态为SUCCESS时，将该节点提交到执行队列。
5. 错误处理：节点失败时，根据配置决定是否重试、跳过下游或触发错误处理分支。

import asyncio
from enum import Enum

class TaskStatus(Enum):
    PENDING = 1
    RUNNING = 2
    SUCCESS = 3
    FAILED = 4

class DAGNode:
    def __init__(self, task_id, coro, dependencies=None):
        self.task_id = task_id
        self.coro = coro
        self.dependencies = dependencies or []
        self.status = TaskStatus.PENDING

class DAGScheduler:
    def __init__(self, nodes):
        self.nodes = {n.task_id: n for n in nodes}
        self.status_map = {n.task_id: TaskStatus.PENDING for n in nodes}

    def get_ready_nodes(self):
        ready = []
        for n in self.nodes.values():
            if n.status != TaskStatus.PENDING:
                continue
            deps_all_done = all(
                self.nodes[d].status == TaskStatus.SUCCESS
                for d in n.dependencies
            )
            if deps_all_done:
                ready.append(n)
        return ready

    async def execute_node(self, node):
        node.status = TaskStatus.RUNNING
        try:
            await node.coro()
            node.status = TaskStatus.SUCCESS
        except Exception as e:
            node.status = TaskStatus.FAILED
            # 错误追踪：记录失败上下文
            print(f"Task {node.task_id} failed: {e}")

    async def run(self):
        while True:
            ready = self.get_ready_nodes()
            if not ready:
                break
            tasks = [self.execute_node(n) for n in ready]
            await asyncio.gather(*tasks)

下面用另一张SVG图展示DAG调度器内部的状态流转与调度决策过程，帮助理解任务从提交到完成的完整生命周期：

2.7 日志采集与错误追踪中的DAG实践

在Development环境的异步任务编排中，DAG为日志采集与错误追踪提供了天然的结构化支持：

• 日志采集：每个DAG节点在执行前后输出结构化日志（包含节点ID、状态、时间戳、输入输出摘要），形成完整的执行轨迹。
• 错误追踪：当某个节点失败时，DAG调度器可以记录失败节点的上下文（依赖链、输入参数、异常堆栈），并自动将错误信息传播到下游的错误处理分支。
• 可视化：DAG的拓扑结构可以直接映射到监控面板，运维人员可以直观地看到哪个环节出现瓶颈或异常。