这道题我去年就遇到过当时很多人一看到“任务编排系统”这六个字就上头以为输出一个拓扑序列就能交差。实际上这题的核心不是“能不能排序”而是“当多个任务同时满足执行条件时先做谁、做完之后整个系统要花多久、机器数量不够时怎么塞”。今天不绕圈子直接把题目拆开给出 Python 和 JavaScript 两套能跑的代码并把我踩过的几个坑一起写清楚。先说结论这道“任务编排系统”的本质是“拓扑排序 贪心选任务 事件模拟”。它比单纯背一个拓扑排序模板要难一点但只要把依赖图、就绪队列、执行单元三者的关系理清楚不管题目把 C 卷换成 D 卷还是 E 卷你都能在十分钟内写出稳定通过的版本。1. 题目背景与核心考点1.1 “双机位 C 卷”到底是什么意思先把这个标题里的信息剥开。华为 OD 机试中的“双机位”指的是考试时的监控方式一个摄像头对着考生正面另一个放在侧后方用来防作弊。这个信息本身和算法无关但它说明了一个事实机试是真实计时、真实监考的你没法靠查手机把答案抄完考前把题目套路吃透才是正路。“C 卷”是考生圈子里对机试随机分卷的一种叫法。不同批次的考生可能拿到题面类似、但顺序或小参数不同的试卷所以网上的分享里会看到 A 卷、B 卷、C 卷这样的关键词。说到底卷型不影响我们准备因为核心考点就那些图论、动态规划、字符串处理、模拟。还有标题里那个“100%通过率”。我直接说这种词基本都是培训机构或资料商的营销话术。没有哪套题能保证 100% 通过尤其是 OD 机试这种多用例判分模式边界条件一多稍不留神就会挂。但反过来说像“任务编排系统”这种高频考点你把依赖建模和调度逻辑吃透通过率确实能拉到很高的水平。1.2 任务编排系统到底考什么这题在市面上的各种版本里常见的核心要素是这几个任务编号从 1 到 N每个任务有一个正整数耗时。任务之间存在依赖关系比如“任务 A 必须在任务 B 开始前完成”。有 K 个执行单元可以理解成 K 个 CPU 核心或 K 个 worker同一时刻最多执行 K 个任务。调度规则只要出现空闲执行单元就从所有满足依赖条件且尚未执行的任务中选耗时最短的任务先执行如果耗时相同选编号更小的。问所有任务完成后系统总共花了多长时间。这个描述是我基于题目关键词和常见变体重构出来的标准形态。实际考试时有的版本会问“输出任务执行顺序”有的会问“最少需要多少时间”但解题的内核都是一样的。如果考试时你拿到的题面和我这里的略有不同只要把这几个要素对应上思路完全可以直接迁移。1.3 输入输出格式示例我按自己平时练习的习惯把输入格式整理成下面这样第一行一个整数 N表示任务数量 第二行 N 个整数表示每个任务的耗时 第三行一个整数 K表示并行执行单元数量 第四行一个整数 M表示依赖关系的数量 接下来 M 行每行两个整数 a b表示任务 a 必须在任务 b 开始前完成 输出一个整数表示完成所有任务所需的总时间举例5 2 3 1 4 5 2 3 1 3 2 3 4 5这个例子我后面会拿来做完整手工推演。现在先记住一件事这题的输出不是拓扑序列而是一个时间数值。2. 算法核心为什么这样设计2.1 用图来建模任务依赖任务之间的依赖关系天然就是一张有向无环图。比如“1 必须在 3 之前完成”就是一条从 1 指向 3 的边。当某一个任务的所有入边对应的前置任务都完成之后这个任务才进入“可执行”状态。所以数据结构上我们需要三样东西邻接表adj记录每个任务完成后能解锁哪些后续任务。入度数组indeg记录每个任务还剩下几个前置任务没完成。就绪堆ready存放所有已经满足依赖条件、可以随时开始执行的任务。这里的“入度”不要和图论教材里的概念割裂开理解。你就把它当成“这个任务还有多少个‘爹’没完事儿”。每完成一个前置任务就把后继任务的入度减 1当入度减到 0说明所有前置任务都完事了它才有资格进入就绪堆。2.2 为什么必须用最小堆而不是直接数组排序题目里那句“选耗时最短的任务先执行”是最容易让人掉坑的地方。很多人的第一反应是我每次把所有就绪任务都扫一遍挑个最小的不就行了吗这样做在小数据量下确实能跑对。但 OD 机试的用例里 N 往往给到10^5甚至更大如果每个任务完成时都扫描一遍所有就绪节点复杂度会退化到O(N^2)。后面的大用例必然超时。正确的做法是用一个最小堆优先队列来维护就绪任务。每次有新任务进入就绪状态就把它按(耗时, 任务编号)的键值压入堆中需要选任务时从堆顶弹出最小的那个就行。堆的插入和弹出都是O(log N)整体复杂度就是O((N M) log N)能稳稳扛住大数据量。这里还有一个细节为什么元组里要把任务编号放在第二位因为 Python 的heapq在比较元组时会先比较第一个元素如果相同再比较第二个。把(耗时, 编号)作为整体放进堆里天然就满足了“耗时相同选编号小”的规则不用额外写排序逻辑。2.3 时间推进要按“事件”走不要按“秒”走任务并行执行时最忌讳的做法是写一个从 0 到最终时间的循环每秒检查一次状态。如果最终时间是10^9这个循环直接就炸了。正确的做法是按“事件时间”推进。当前正在执行的任务里哪一个最先完成下一个事件时间就是它的完成时刻。比如现在有两个任务在跑一个会在第 5 秒完成一个会在第 8 秒完成那系统时间直接跳到第 5 秒处理第 5 秒完成的那个任务然后再看这个任务解锁了哪些新任务。这种事件驱动模拟循环次数只和任务完成事件的数量有关不会受总时间大小影响。2.4 完整算法流程梳理我把整个流程写成下面这个可落地的步骤序列读入所有输入构建邻接表和入度数组。把所有入度为 0 的任务按(耗时, 编号)压入就绪小顶堆。定义一个“正在执行”的集合或堆用来存放(预计完成时间, 任务编号)。只要就绪堆不为空或者正在执行的任务不为空就循环先把空闲执行单元填满如果当前正在执行的任务数小于 K并且就绪堆里有任务就弹出堆顶任务开始执行把它的预计完成时间记为“当前时间 耗时”放入正在执行集合。找到正在执行任务中预计完成时间最小的那个把当前时间推进到那个时刻。处理所有在同一时刻完成的任务。对每个完成的任务遍历它的后继任务把后继任务的入度减 1如果入度变成 0就把后继任务压入就绪堆。从正在执行集合中移除这些已完成任务进入下一轮循环。全部循环结束时输出当前时间。这个流程里有一个很容易被忽略的点当某个任务完成并解锁了后续任务时如果此时还有空闲执行单元新解锁的任务应该立刻开始执行而不是等到下一批任务全部完成后再统一开始。所以循环开头必须有一个“填满空闲执行单元”的步骤。3. Python 完整实现与逐段解读3.1 输入读取与数据结构初始化Python 的机试环境里读取输入我推荐用sys.stdin.read().split()一次性读入所有 token而不是用input()一行一行读。这样做的好处是不依赖具体输入行怎么换行哪怕耗时数组被拆成了多行也能完整读进来。初始化部分的核心是给编号留一个下标偏移。任务编号从 1 开始所以我创建数组时长度统一是N 1数组下标 0 的位置直接弃用。这样duration[i]就是任务 i 的耗时adj[i]就是任务 i 的后继列表下标对不上这种低级错误就不会出现。import sys import heapq def solve(): data sys.stdin.read().split() idx 0 n int(data[idx]) idx 1 duration [0] [int(x) for x in data[idx:idx n]] idx n k int(data[idx]) idx 1 m int(data[idx]) idx 1 adj [[] for _ in range(n 1)] indeg [0] * (n 1) for _ in range(m): a int(data[idx]) b int(data[idx 1]) idx 2 adj[a].append(b) indeg[b] 1 ready [] for i in range(1, n 1): if indeg[i] 0: heapq.heappush(ready, (duration[i], i))这里有一个细节想提醒你data[idx:idxn]转成数字后前面直接拼一个[0]是为了让任务编号和位置对齐。如果不做这个偏移任务 1 会被存在下标 0 上后面写duration[1]就会拿到任务 2 的耗时这种错位在复杂题目里非常难查。3.2 正在执行任务的数据结构选择“正在执行”这个集合我用的是一个小顶堆键值为(预计完成时间, 任务编号)。这样每次要找“谁最先结束”直接看堆顶就行不需要遍历整个正在执行列表。但这里有个隐患堆只能保证堆顶最小不能快速判断“有哪些任务在同一时刻完成”。不过没关系我们的处理方式是先弹出堆顶记录它的完成时间now然后只要堆顶的完成时间仍然等于now就继续弹出。因为同一时刻完成的任务键值里的完成时间相同堆顶会连续弹出它们。running [] now 0 while ready or running: while len(running) k and ready: d, job heapq.heappop(ready) heapq.heappush(running, (now d, job)) if not running: break now running[0][0] finished [] while running and running[0][0] now: _, job heapq.heappop(running) finished.append(job) for job in finished: for nxt in adj[job]: indeg[nxt] - 1 if indeg[nxt] 0: heapq.heappush(ready, (duration[nxt], nxt))这段代码里最需要注意的是while len(running) k and ready这一步。它保证了只要有空位就绪任务就会立刻顶上。我见过不少人把这步写成if len(running) k结果一个任务完成解锁新任务后同一轮循环里新任务不能立刻开始必须等下一轮导致整体时间偏大。虽然有时候不影响答案但一旦用例卡这个细节就白丢分了。3.3 完整 Python 源码把上面的片段拼起来再加上输出部分就是可以直接提交的版本import sys import heapq def solve(): data sys.stdin.read().split() idx 0 n int(data[idx]) idx 1 duration [0] [int(x) for x in data[idx:idx n]] idx n k int(data[idx]) idx 1 m int(data[idx]) idx 1 adj [[] for _ in range(n 1)] indeg [0] * (n 1) for _ in range(m): a int(data[idx]) b int(data[idx 1]) idx 2 adj[a].append(b) indeg[b] 1 ready [] for i in range(1, n 1): if indeg[i] 0: heapq.heappush(ready, (duration[i], i)) running [] now 0 while ready or running: while len(running) k and ready: d, job heapq.heappop(ready) heapq.heappush(running, (now d, job)) if not running: break now running[0][0] finished [] while running and running[0][0] now: _, job heapq.heappop(running) finished.append(job) for job in finished: for nxt in adj[job]: indeg[nxt] - 1 if indeg[nxt] 0: heapq.heappush(ready, (duration[nxt], nxt)) print(now) if __name__ __main__: solve()这段代码的结构其实就是把“事件模拟”压缩成了两个堆一个管等待执行的任务一个管正在执行的任务。只要这两个堆的状态同步正确最终时间就一定算得准。4. JavaScript 完整实现与关键差异4.1 Node.js 环境下的输入处理JS 版本在机试环境里输入输出方式和 Python 差别很大。Node.js 没有input()这种同步读入必须通过readline模块监听line事件把所有行收集完之后再统一处理。我先说一个容易踩的坑机试的输入末尾有时会多一个空行有时不会。所以收集完lines数组后处理时一定要用trim()去掉首尾空白否则数字解析会出错。还有split(/\s/)能一次处理多个空格比按单个空格切分更稳。const readline require(readline); const rl readline.createInterface({ input: process.stdin, output: process.stdout }); const lines []; rl.on(line, (line) { if (line.trim().length 0) { lines.push(line.trim()); } }); rl.on(close, () { let idx 0; const n Number(lines[idx]); const times lines[idx].split(/\s/).map(Number); const duration [0, ...times]; const k Number(lines[idx]); const m Number(lines[idx]); const adj Array.from({ length: n 1 }, () []); const indeg new Array(n 1).fill(0); for (let i 0; i m; i) { const parts lines[idx].split(/\s/).map(Number); const a parts[0]; const b parts[1]; adj[a].push(b); indeg[b]; } // 主逻辑在这里继续 });4.2 手写最小堆还是直接用内置优先队列这是很多 JS 考生纠结的问题。Node.js 目前没有内置的优先队列虽然有一种实验性质的MinPriorityQueue在不稳定版本里出现过但机试环境千万不要依赖它。最稳妥的做法是手写一个通用最小堆。这里的手写最小堆我用的是一个构造函数接收一个比较函数compare。这样同一个堆类既能存(耗时, 编号)也能存(预计完成时间, 编号)两种用途都能复用。需要注意堆的下沉和上浮操作都要用比较函数来决定顺序不能写死。class MinHeap { constructor(compare) { this.heap []; this.compare compare; } size() { return this.heap.length; } peek() { return this.size() 0 ? this.heap[0] : null; } push(val) { this.heap.push(val); this._up(this.heap.length - 1); } pop() { if (this.size() 0) return null; const top this.heap[0]; const last this.heap.pop(); if (this.size() 0) { this.heap[0] last; this._down(0); } return top; } _up(i) { while (i 0) { const parent (i - 1) 1; if (this.compare(this.heap[i], this.heap[parent]) 0) { [this.heap[i], this.heap[parent]] [this.heap[parent], this.heap[i]]; i parent; } else { break; } } } _down(i) { while (true) { let smallest i; const left i * 2 1; const right i * 2 2; if (left this.size() this.compare(this.heap[left], this.heap[smallest]) 0) { smallest left; } if (right this.size() this.compare(this.heap[right], this.heap[smallest]) 0) { smallest right; } if (smallest ! i) { [this.heap[i], this.heap[smallest]] [this.heap[smallest], this.heap[i]]; i smallest; } else { break; } } } }这个实现里我用了解构赋值的骚操作来交换数组元素[this.heap[i], this.heap[parent]] [this.heap[parent], this.heap[i]]。这是 ES6 的合法语法机试的 Node.js 版本一般都支持。如果你担心版本问题老老实实写一个临时变量交换也没问题。4.3 JS 完整源码把堆类、输入处理和主逻辑拼起来就得到了完整可运行的 JS 版本class MinHeap { constructor(compare) { this.heap []; this.compare compare; } size() { return this.heap.length; } peek() { return this.size() 0 ? this.heap[0] : null; } push(val) { this.heap.push(val); this._up(this.heap.length - 1); } pop() { if (this.size() 0) return null; const top this.heap[0]; const last this.heap.pop(); if (this.size() 0) { this.heap[0] last; this._down(0); } return top; } _up(i) { while (i 0) { const parent (i - 1) 1; if (this.compare(this.heap[i], this.heap[parent]) 0) { [this.heap[i], this.heap[parent]] [this.heap[parent], this.heap[i]]; i parent; } else { break; } } } _down(i) { while (true) { let smallest i; const left i * 2 1; const right i * 2 2; if (left this.size() this.compare(this.heap[left], this.heap[smallest]) 0) { smallest left; } if (right this.size() this.compare(this.heap[right], this.heap[smallest]) 0) { smallest right; } if (smallest ! i) { [this.heap[i], this.heap[smallest]] [this.heap[smallest], this.heap[i]]; i smallest; } else { break; } } } } const compareTask (a, b) { if (a[0] ! b[0]) return a[0] - b[0]; return a[1] - b[1]; }; const readline require(readline); const rl readline.createInterface({ input: process.stdin, output: process.stdout }); const lines []; rl.on(line, (line) { if (line.trim().length 0) { lines.push(line.trim()); } }); rl.on(close, () { let idx 0; const n Number(lines[idx]); const times lines[idx].split(/\s/).map(Number); const duration [0, ...times]; const k Number(lines[idx]); const m Number(lines[idx]); const adj Array.from({ length: n 1 }, () []); const indeg new Array(n 1).fill(0); for (let i 0; i m; i) { const parts lines[idx].split(/\s/).map(Number); adj[parts[0]].push(parts[1]); indeg[parts[1]]; } const ready new MinHeap(compareTask); for (let i 1; i n; i) { if (indeg[i] 0) { ready.push([duration[i], i]); } } const running new MinHeap(compareTask); let now 0; while (ready.size() 0 || running.size() 0) { while (running.size() k ready.size() 0) { const item ready.pop(); const d item[0]; const job item[1]; running.push([now d, job]); } if (running.size() 0) { break; } now running.peek()[0]; const finished []; while (running.size() 0 running.peek()[0] now) { const item running.pop(); finished.push(item[1]); } for (const job of finished) { for (const nxt of adj[job]) { indeg[nxt]--; if (indeg[nxt] 0) { ready.push([duration[nxt], nxt]); } } } } console.log(now); });JS 和 Python 的实现逻辑完全一致只是在语法上有几个差异需要提醒你JS 的数组解构赋值能对heap里的元素直接进行操作非常方便但要确保不会越界。running.peek()返回的是null或一个数组所以主逻辑里要先判断 running 是否为空再取[0]。比较函数里a[0] - b[0]可能会导致大整数溢出吗在本题耗时是正整数且不超过2^31 - 1的情况下完全不会。但如果耗时的数量级极大建议改成a[0] b[0] ? -1 : (a[0] b[0] ? 1 : a[1] - b[1])从根本上避免减法溢出。5. 测试用例与手工推演5.1 基础依赖用例先回到前面的例子5 2 3 1 4 5 2 3 1 3 2 3 4 5初始时任务 1、2、4 入度为 0任务 3 依赖 1 和 2任务 5 依赖 4。就绪堆里是(2,1)、(3,2)、(4,4)取前两个执行任务 1 耗时 2任务 2 耗时 3。时间推进到 2任务 1 完成。任务 3 的入度从 2 变成 1还不能执行。此时正在执行的任务 2 还没结束就绪堆为空。时间推进到 3任务 2 完成。任务 3 入度从 1 变成 0进入就绪堆。此时执行单元全空于是把任务 3耗时 1和任务 4耗时 4同时启动。时间推进到 4任务 3 完成。任务 4 还在执行就绪堆为空。注意任务 5 此时还不能执行因为任务 4 还没完成。时间推进到 7任务 4 完成。任务 5 入度归零进入就绪堆此时只有一个空闲执行单元启动任务 5。时间推进到 12任务 5 完成。最终输出 12。这个用例最大的价值在于它展示了“机器满了但任务还没就绪”的等待场景任务 3 在 4 秒就完成了但任务 5 必须等到任务 4 在 7 秒完成才能开始。如果不做事件模拟只看拓扑关键路径很容易算成错误答案。5.2 单任务无依赖用例1 7 1 0只有一个任务耗时 7一个执行单元。就绪堆初始只有任务 1直接执行输出 7。这个用例是边界测试主要验证数组下标和输入解析没有越界问题。5.3 多任务无依赖、并行度不足用例5 2 3 1 4 5 3 0五个任务之间没有任何依赖三个执行单元同时工作。初始就绪堆为(1,3)、(2,1)、(3,2)、(4,4)、(5,5)。时间 0 启动任务 3、1、2。任务 3 耗时 1最先完成此时空出一个执行单元立刻启动任务 4。时间 2任务 1 完成空出第二个执行单元立刻启动任务 5。时间 3任务 2 完成。此时三个执行单元分别跑着任务 4 到 5 秒、任务 5 到 7 秒没有空位等待。时间 5任务 4 完成。空出一个执行单元但已经没有等待任务了。时间 7任务 5 完成。最终输出 7。这个用例想说明一个概念即使没有依赖关系并行执行单元的数量 K 也会限制整体完成时间。如果 K 改成 5五个任务全部同时启动输出就是最大耗时 5。千万不要在代码里把所有任务的耗时简单相加。5.4 深度依赖链用例3 2 3 1 3 2 1 2 2 3任务 1 依赖前置为空任务 2 依赖任务 1任务 3 依赖任务 2。任务 3 耗时 1但因为必须先等任务 1 和任务 2所以即使 K 为 3它也得等链上的任务走完。时间 0 启动任务 1耗时 2同时任务 3 入度不是 0不能启动。时间 2任务 1 完成任务 2 就绪启动任务 2耗时 3。时间 5任务 2 完成任务 3 就绪启动任务 3耗时 1。时间 6任务 3 完成输出 6。这条链式用例展示了依赖关系中最“朴素”的情况不管你有多少并行执行单元只要形成一条链整体时间就是链上所有任务耗时之和。用代码跑这个用例可以验证入度更新的链路是否正确。6. 考场常见问题与避坑思路6.1 就绪堆填满执行单元的时机我在前面反复强调过while len(running) k and ready这个循环。这里再展开讲一下为什么它必须是while而不是if。假设 K 等于 3某一轮结束时正在执行的任务只剩 1 个而就绪堆里有 3 个任务。如果是if这一轮只启动 1 个任务剩下 2 个要等下一轮才能开始白白浪费了执行单元。用while才能一次性把所有空位都填满。这个点虽然代码只有一行却决定了模拟结果是否正确。6.2 多个任务同时完成的处理我正在执行堆里用的键值是(预计完成时间, 任务编号)。如果两个任务预计都在第 5 秒完成其中编号更小的会排在堆顶。处理完成事件时必须用一个while循环把完成时间等于now的任务全部弹出而不是只弹一个。这里还有一个很微妙的点多个任务同时完成时它们的后继任务可能互为依赖。比如任务 A 完成后解锁任务 C任务 B 完成后解锁任务 CC 的入度要经过两次减 1 才到 0。在同一个完成事件里必须把所有已完成任务遍历完并且全部更新完入度才能继续下一轮启动。如果你在遍历过程中就把新就绪的任务压入堆并且立刻开始下一轮填充可能会出现“C 的入度还没减到底就被错误地当成就绪任务”的问题吗不会因为入度更新一定是减到 0 才压堆但只要你把入度更新和“启动新任务”混在同一个阶段里调度顺序就容易乱。我的建议是严格分阶段先集中处理完成事件再统一填满执行单元。6.3 循环依赖的处理这道题默认输入是一张有向无环图但实际考试中用例有时会故意给你一个环比如 1 依赖 2、2 依赖 1。这种用例往往是用来卡“死循环”实现的。我见过一些考生用递归拓扑排序遇到环直接栈溢出。我这套解法用的是迭代 入度数组天然不会递归爆栈。但需要注意如果存在环环上的任务入度永远不会变成 0就绪堆一直为空而正在执行的任务也可能很快清空主循环就会退出输出一个偏小的答案。这虽然不是标准答案但至少不会让程序崩溃。如果你在本地调试时发现输出明显不合理第一反应就应该是检查输入里是否有环而不是怀疑堆写错了。6.4 编号下标从 1 开始还是从 0 开始这个坑非常隐蔽。题目说任务编号从 1 到 N你就必须保证数组下标也是从 1 开始。我见过有人图省事把数组长度直接设为 N然后用duration[i - 1]来读耗时结果在邻接表的循环里忘了减 1导致建图全错。我的建议很简单所有数组长度统一用N 1把下标 0 的位置当成垃圾位。虽然多开一个整数数组的空间不算什么但它能让你在写代码时少操一份心。6.5 大整数边界任务的耗时可能很大所有任务完成的总时间可能超过2^31 - 1。在 Python 里完全不用担心int 可以无限大。但在 JS 里如果你用Number存储精度可能出问题尤其是涉及now d这种加法时超过Number.MAX_SAFE_INTEGER就会丢精度。机试用例一般不会出这种极端数据但稳妥起见你可以把所有时间相关变量保持为普通数字因为只要每个耗时本身不超过2^31 - 1加法的中间结果通常也安全。如果实在不放心可以用BigInt全程存储但要注意BigInt不能和普通数字直接比较写起来麻烦一点。6.6 关于“100%通过率”的正确心态最后再说回标题。市面上的“100%通过率”基本都是营销词真正决定你能不能过的东西是你对题型的熟练度和边界处理的完整度。OD 机试的判分系统是多用例判分核心用例过了只能拿到部分分边界用例全过才能满分。所以刷题时别只盯着一个题解跑通就完事多想一想“如果任务编号从 1 开始”、“如果并行度大于任务数”、“如果耗时相等”这些边界场景你的代码是不是依然正确。我个人在实际练习里的一个小习惯是每次写完一个题都会用至少五组不同形态的测试数据去验证。一组无依赖、一组全依赖形成的单链、一组有多个同时完成事件、一组 K 大于 N、一组带环。把这五组跑完代码里的低级错误基本都能暴露出来。这个方法不仅适用于“任务编排系统”也适用于所有图论模拟题。你可以把这个思路沉淀成自己的刷题流程机试时心里会踏实很多。