4. 关键路径分析：任务流的时间命脉

关键路径分析，说白了就是找出整个任务流里，哪条路最“拖不起”。

我在做分布式调度系统的时候，经常遇到这种情况：明明大部分任务都跑完了，整个流程却卡在一个不起眼的小任务上。后来我才意识到，这就是关键路径在作祟。你想想看，一个 DAG 里可能有几十上百个节点，但真正决定总工期的，往往就是那么一条路径。

4.1 关键路径的定义

关键路径（Critical Path）是 DAG 中从起点到终点，总耗时最长的那条路径。它决定了整个任务流的最短完成时间。

为什么叫“关键”？因为这条路径上的任何一个任务延迟了，整个项目就得跟着延迟。其他路径上的任务就算有点余量，也不影响大局。

举个例子，一个简单的 DAG：A(2) → B(3) → C(4)，总耗时 9。如果 A 和 C 之间还有一条捷径 A → D(1) → C，那关键路径就是 A→B→C，耗时 9。D 任务就算慢一点，也不影响大局。

4.2 最早开始时间与最晚开始时间

要找到关键路径，得先搞懂两个核心概念：最早开始时间（ES）和最晚开始时间（LS）。

4.2.1 最早开始时间（Earliest Start, ES）

一个任务在不依赖任何前置条件的情况下，最早什么时候能开始。说白了，就是等它所有前置任务都完成后，才能启动。

计算公式：

ES(node) = max(ES(predecessor) + duration(predecessor))

起点任务的 ES 通常设为 0。

4.2.2 最晚开始时间（Latest Start, LS）

一个任务在不影响整个项目工期的前提下，最晚什么时候必须开始。如果超过了这个时间，整个项目就会延期。

计算公式：

LS(node) = min(LS(successor) - duration(node))

终点任务的 LS 等于它的 ES。

4.2.3 浮动时间（Slack/Float）

浮动时间 = LS - ES。如果浮动时间为 0，这个任务就在关键路径上。

概念	含义	计算方式
ES	最早能开始的时间	前置任务 ES + 耗时 的最大值
LS	最晚必须开始的时间	后继任务 LS - 当前耗时 的最小值
浮动时间	可以拖延的余地	LS - ES

4.3 关键路径算法实现

关键路径算法分两步走：正向遍历算 ES，反向遍历算 LS。嗯，这里要注意，DAG 必须是无环的，否则算法会死循环。

4.3.1 算法步骤

1. 拓扑排序：先给 DAG 排个序，保证每个节点都在它的前置节点之后处理。
2. 正向遍历：按拓扑顺序，计算每个节点的 ES。
3. 反向遍历：按逆拓扑顺序，计算每个节点的 LS。
4. 找关键路径：筛选出 ES == LS 的节点，它们连起来就是关键路径。

4.3.2 代码实现（Python）

from collections import defaultdict, deque

def critical_path(nodes, edges, durations):
    """
    nodes: 节点列表
    edges: 边列表 [(u, v)]
    durations: 节点耗时 {node: duration}
    """
    # 构建邻接表和入度表
    adj = defaultdict(list)
    in_degree = {n: 0 for n in nodes}
    for u, v in edges:
        adj[u].append(v)
        in_degree[v] += 1

    # 拓扑排序
    queue = deque([n for n in nodes if in_degree[n] == 0])
    topo_order = []
    while queue:
        u = queue.popleft()
        topo_order.append(u)
        for v in adj[u]:
            in_degree[v] -= 1
            if in_degree[v] == 0:
                queue.append(v)

    # 正向遍历：计算 ES
    es = {n: 0 for n in nodes}
    for u in topo_order:
        for v in adj[u]:
            es[v] = max(es[v], es[u] + durations[u])

    # 反向遍历：计算 LS
    ls = {n: es[n] for n in nodes}  # 终点 LS = ES
    for u in reversed(topo_order):
        for v in adj[u]:
            ls[u] = min(ls[u], ls[v] - durations[u])

    # 找关键路径
    critical = [n for n in nodes if es[n] == ls[n]]
    return critical, es, ls

# 示例
nodes = ['A', 'B', 'C', 'D', 'E']
edges = [('A','B'), ('A','C'), ('B','D'), ('C','D'), ('D','E')]
durations = {'A':2, 'B':3, 'C':1, 'D':4, 'E':2}

critical, es, ls = critical_path(nodes, edges, durations)
print("关键路径节点:", critical)
print("ES:", es)
print("LS:", ls)

4.4 关键路径的 SVG 示意图

下面这张图展示了一个典型 DAG 的关键路径计算过程。蓝色节点是关键路径，灰色节点有浮动时间。

4.5 实际应用中的注意事项

• 多关键路径：有时候 DAG 里会有多条关键路径，任何一条上的任务延迟都会影响工期。我遇到过同时维护 3 条关键路径的情况，调度策略得格外小心。
• 动态更新：任务执行过程中，耗时可能会变化（比如某个任务跑慢了）。这时候需要重新计算关键路径。我在系统里加了个“增量更新”机制，只重新计算受影响的部分，而不是全量重算。
• 资源约束：经典关键路径算法不考虑资源竞争。如果两个关键路径上的任务争抢同一个资源，实际工期会比理论值更长。嗯，这个坑我踩过，后来引入了资源约束的关键链法（Critical Chain）才解决。