Python TopologicalSorter 怎样分批取得可执行节点

如果一组任务之间存在前置依赖，想把“现在能做的任务”一次取出来，可以使用 Python 标准库 graphlib.TopologicalSorter。它不会每次只给一个节点，而是通过 get_ready() 返回一个元组；等这一批任务真正处理完，再用 done(*nodes) 标记完成，下一轮才会释放新的节点。

官方地址：https://docs.python.org/3/library/graphlib.html

分批调度的核心只有三件事：用字典值写清前置节点，调用 prepare() 固定图，然后循环执行“取一批、处理一批、done 一批”。

先把依赖方向写对

TopologicalSorter 的输入是“节点到全部前置节点”的映射。比如发布任务依赖渲染和检查，渲染、检查又依赖编译，那么 publish 的值应该是 {"render", "lint"}，而不是把后继任务写进去。没有前置条件的节点可以使用空集合。

调用 prepare() 后，图进入调度状态，不能再继续通过 add() 修改。准备阶段会检查循环；如果存在循环，会抛出 CycleError。因此，批量取节点前要先完成建图和准备。

from graphlib import CycleError, TopologicalSorter

# 每个键表示任务，每个值表示它必须等待的前置任务。
graph = {
    "publish": {"render", "lint"},
    "render": {"compile"},
    "lint": {"compile"},
    "compile": set(),
}

sorter = TopologicalSorter(graph)
try:
    # prepare 会固定图，并在发现循环时抛出 CycleError。
    sorter.prepare()
except CycleError as exc:
    # 生产代码可把 exc.args 记录到任务诊断日志中。
    raise RuntimeError("任务依赖存在循环") from exc

get_ready() 返回的就是一批节点

准备完成后，第一次调用 get_ready() 会返回所有没有前置依赖的节点。后续调用则返回前置节点都已经通过 done() 标记的节点。返回值是 tuple，所以可以把它作为一个调度批次交给线程池、进程池或自己的任务队列。

下面用同步函数模拟消费侧。示例重点是批次边界：一次循环拿到的所有节点都属于同一轮，不能只处理其中一个就把整批传给 done()。

def run_task(node: str) -> None:
    # 这里替换成实际任务；示例只保留调度关系，不代表真实执行结果。
    print(f"处理任务: {node}")


while sorter.is_active():
    ready_nodes = sorter.get_ready()
    if not ready_nodes:
        # 异步场景可能仍有尚未完成的任务，此处交给完成回调处理。
        break

    for node in ready_nodes:
        run_task(node)

    # 只有当前批次的任务都处理完，后继节点才可能在下一轮 ready。
    sorter.done(*ready_nodes)

如果图中 compile 是唯一的根节点，第一批会包含它；完成标记后，render 和 lint 才能同时进入下一批。最后，两个后继都完成，publish 才会变为 ready。这里的“同时”表示同一次 get_ready() 返回，不表示必须并行执行。

用 done() 解锁下一批

done(*nodes) 不是“从队列删除节点”这么简单，它会把已经返回且已处理的节点记为完成，并解除对应后继节点的等待条件。传入的节点必须来自此前的 get_ready()，重复调用、传入未知节点或尚未 ready 的节点，都会触发 ValueError。

接入真正的并发 worker 时，不要在任务刚投递到队列后立即调用 done()，否则调度器会误以为任务已经完成。正确做法是：worker 完成一个节点后，把节点交给完成处理器，再调用对应的 done(node)；等下一次 get_ready() 取得的新元组后，再继续投递。

并发与异常边界怎么处理

is_active() 表示调度器仍有进展空间：可能还有 ready 节点尚未取出，也可能已有节点被取出但还没有全部调用 done()。因此异步实现不能把一次空的 get_ready() 当作整个调度结束，而应等待正在执行的 worker 完成。

如果只需要一个不参与并发的拓扑顺序，可以使用 static_order()；它内部完成准备、取 ready 批次和 done 标记，不要再对同一个 sorter 手动调用这些方法。需要并发或按批次投递时，才使用 prepare()、get_ready()、done() 这组接口。

还有两个容易忽略的边界：节点必须是可哈希对象；同一层 ready 节点之间没有由依赖图决定的固定顺序，结果顺序可能受插入顺序影响。如果业务需要稳定顺序，应在消费侧定义自己的排序规则，但不能改变“前置节点完成后才释放后继”的依赖约束。

常见问题

为什么 get_ready() 一开始返回多个节点？

因为这些节点都没有未完成的前置依赖。它们属于同一 ready 批次，可以交给多个 worker，也可以在单线程中逐个处理。

调用 done() 后为什么没有新节点？

检查传入的节点是否真的已经完成，以及它们是否覆盖某个后继节点的全部前置条件。只完成部分前置节点时，后继节点仍然不会 ready。

什么时候应该用 static_order()？

只想获得一个拓扑顺序、没有任务并发和完成回调时使用它；需要控制批次或把工作交给外部 worker 时，使用显式的批次接口。

总结一下，get_ready() 负责给出当前可执行的一批节点，done(*nodes) 负责反馈真实完成状态，is_active() 负责判断调度是否还有进展。把“取出”和“完成”分开，才能在串行和并发场景中都保持依赖关系正确。