最近一次华为OD机试,双机位C卷里出现了一道《任务编排系统》。很多人一听“系统”两个字就紧张,以为要设计数据库、接口、消息队列,等看到题目才发现,它其实就是一张带依赖关系的任务图,让你判断任务能不能排得通、按什么顺序执行、最早什么时候全部跑完——说白了一句:考拓扑排序,外加一个关键路径。
这篇文章把我复盘后的完整思路和两种语言的实现都放出来,Python版和JavaScript版都能直接在机试环境里跑。无论你是刚开始刷OD真题,还是已经在牛客上被各种题折磨过,这道题都值得认真吃透。它把“图论建模”和“实际工程”结合得挺自然,面试复盘时也常被追问。
至于标题里的“100%通过率”,我不太信玄学。上机能一遍过,靠的是把建模、边界、输入解析这三个环节都做扎实了。下面我会把每一步拆开讲清楚,尤其是我自己踩过的坑、考场上容易翻车的地方,全部摊开给你看。
1. 考题复现:任务编排系统到底让你做什么
先把题目场景复述一遍。假设你负责一个分布式任务编排平台,系统里有N个任务,编号从0到N-1。每个任务有固定的执行耗时,还有一些前置任务——前置任务全部完成后,当前任务才能开始。平台有足够多的执行器,不存在资源竞争,所以互相之间没有依赖关系的任务完全可以同时跑。
需要你输出三样东西:
- 判断当前的依赖关系是否存在环。如果有环,说明这些任务永远排不出来,直接输出ERROR。
- 如果无环,给出一个合法的执行顺序。多个任务同时可以开始的时候,按任务编号从小到大输出。
- 计算整个编排的最短完成时间,也就是从第一个任务启动,到最后一个任务结束的总耗时。
输入格式长这样,每一行表示一个任务:
5 0 3 0 1 2 1 0 2 4 1 0 3 1 1 1 4 5 2 1 2解释一下:第一行是任务总数N=5。接下来N行,每行依次是:任务编号、执行耗时、前置任务个数K、K个前置任务编号。
拿第4行举例:4 5 2 1 2表示任务4耗时5个时间单位,有2个前置任务,分别是任务1和任务2。也就是说,任务1和任务2都结束之后,任务4才能开工。
这个输入对应的依赖关系是:
- 任务0无前置,耗时3
- 任务1依赖任务0,耗时2
- 任务2依赖任务0,耗时4
- 任务3依赖任务1,耗时1
- 任务4依赖任务1和任务2,耗时5
合法的输出应该包含两行:
0 1 2 3 4 12第一行是执行顺序,第二行是最短完成时间。
等一下,细心的读者会发现:任务1和任务2都依赖任务0,按道理任务0完成后它们可以同时开始,那谁先谁后?题目规定“同时可开始的任务按编号升序”,所以先输出1再输出2。任务3只依赖任务1,任务1在时间5结束时它就能开始;任务4必须等任务1和任务2都结束,而任务2要到时间7才结束,所以任务4真正开始是时间7。整体最后一个任务结束是时间12。
我考场上拿到这题的第一反应是:这不就是一张有向无环图(DAG)套了个最长路吗?但真正动手写的时候还是有几个细节需要考虑清楚,下面逐个说。
2. 建模思路:把“任务编排”翻译成一张图
无论题目描述包装得多工程化,只要出现“前置依赖”“同时执行”“完成时间”这些词,第一步永远是建模。任务之间的依赖关系天然就是一张有向图:
- 每个任务是一个节点。
- 如果任务A是任务B的前置,就画一条从A指向B的边,表示“A要排在B前面”。
- 任务开始执行的唯一条件:所有指向它的节点都已完成。
2.1 为什么一定是DAG,环意味着什么
如果这张图里有环,比如A依赖B、B又依赖A,那这两个任务永远互相等待。放到真实系统里就是死锁,放到题目里就是“无法编排”。
所以题目要求的“先判断是否有环”,本质上就是在问:这组依赖关系能不能构成DAG。所有无环有向图都能做拓扑排序,反过来,能做完整拓扑排序的图一定是无环的。判断环的办法,就是跑一遍拓扑排序,看最后排出来的节点个数是不是等于N。少于N,说明有环,直接给ERROR。
这个结论我在考场上没有多想,但复盘时觉得值得强调:千万不要上来就写DFS判环,然后把判断环和输出顺序分开做。一次拓扑排序就能同时解决“有没有环”和“执行顺序是什么”两个问题,多写一套逻辑反而容易出bug。
2.2 每个任务的最早开始时间,是取最大值还是最小值
这是整道题最容易理解错的地方。一个任务有多个前置任务,它的最早开始时间应该是:
start[任务] = max(所有前置任务的结束时间)为什么是max而不是min?因为前置任务必须“全部”完成。现实中也好理解:你要等最慢的那个供应商到货了才能开工,不能因为最快的先到了就提前生产。
对应到图上,每个任务的结束时间是:
finish[任务] = start[任务] + cost[任务]整个编排的总耗时是:
totalTime = max(所有任务的finish)这其实就是DAG上的关键路径问题。严格说,关键路径通常是找最长路径,但这里节点带权重、边不带权重,所以直接在拓扑序上做动态规划就能得到每个任务的最早开始时间。这也是我把这题归类为“拓扑排序 + DP”的原因。
2.3 为什么用拓扑排序能同时算出时间
拓扑排序的流程大家都熟:先找所有入度为0的节点,这些任务没有前置依赖,可以直接开始;处理完一个节点后,把它的所有后继节点的入度减1,一旦某个后继的入度变成0,说明它的前置任务都完成了,可以开始调度。
关键点在于:当一个节点入度变成0、被弹出堆的时候,它的start值其实已经被所有前置任务更新过了,而且一定是“最大值”。因为每条指向它的边,在边起点结束的那一刻,都会去更新它的start,取max。所以等到它入度清零弹出时,这个值就是它真正的最早开始时间。
用这个思路去写代码,就不会出现“时间算错”的问题。
3. Python实现:邻接表加最小堆,一次跑通
先贴出完整可运行的Python代码,再逐个解释关键细节。
import sys import heapq def solve_case(n, cost, pres): adj = [[] for _ in range(n)] indeg = [0] * n for i in range(n): for p in pres[i]: adj[p].append(i) indeg[i] += 1 heap = [i for i in range(n) if indeg[i] == 0] heapq.heapify(heap) start = [0] * n order = [] total_time = 0 while heap: u = heapq.heappop(heap) order.append(u) finish_u = start[u] + cost[u] total_time = max(total_time, finish_u) for v in adj[u]: if finish_u > start[v]: start[v] = finish_u indeg[v] -= 1 if indeg[v] == 0: heapq.heappush(heap, v) if len(order) != n: return None return order, total_time def main(): data = sys.stdin.read().strip().split() if not data: return idx = 0 outputs = [] while idx < len(data): n = int(data[idx]) idx += 1 cost = [0] * n pres = [[] for _ in range(n)] for _ in range(n): tid = int(data[idx]) c = int(data[idx + 1]) k = int(data[idx + 2]) idx += 3 ps = [] for _ in range(k): ps.append(int(data[idx])) idx += 1 cost[tid] = c pres[tid] = ps result = solve_case(n, cost, pres) if result is None: outputs.append("ERROR") else: order, total_time = result outputs.append(" ".join(map(str, order))) outputs.append(str(total_time)) sys.stdout.write("\n".join(outputs)) if __name__ == "__main__": main()3.1 为什么要用最小堆而不是普通队列
大部分拓扑排序的教程用的是队列或栈,先进先出就能保证正确性。但这道题多了一个要求:同时可以开始的任务,按编号升序输出。
举个例子,任务0完成后,任务1、任务2、任务5的入度同时变成0。如果用普通队列,它们的出队顺序取决于入队的顺序,也就是遍历adj[0]时后继的顺序。这个顺序并不一定是编号升序。
用最小堆就很简单:把入度为0的节点全部丢进堆,每次弹出编号最小的那个。堆天然保证了“同时可开始”的任务按升序输出。这个细节,考场上如果忽略了,样例可能都过不了。
3.2 输入解析最容易翻车的一个细节
我见过不少同学卡在这:解析输入时默认任务编号是按0到N-1有序出现的,于是直接用数组下标去存cost和pres。
但题目只说了任务编号从0到N-1,并没有承诺输入行按编号有序。稳妥起见,解析时必须按照第一列的任务编号,把耗时和前置列表放到正确的位置:
cost[tid] = c pres[tid] = ps这样即使输入顺序是乱的,代码也一样能跑。机试环境里输入格式偶尔会有各种非预期情况,这里多写一行,省一整个调试周期,非常划算。
3.3 更新后继start值的位置有讲究
看这段:
for v in adj[u]: if finish_u > start[v]: start[v] = finish_u indeg[v] -= 1 if indeg[v] == 0: heapq.heappush(heap, v)注意,更新start[v]是在递减入度之前做的。这不是随意的顺序,而是有逻辑原因的:不管v的其他前置任务是否已经处理完,当前节点u的结束时间都是v的一个参考值,要用max逻辑不断刷新v的“最早可能开始时间”。
如果把这个更新放到indeg[v] == 0之后再去做,逻辑上就反了——v都已经入堆了,再改它的start就晚了。
3.4 Python版本的复杂度
用邻接表存储图,每个节点和每条边都只被访问一次。堆操作的时间是O(logN)。所以总复杂度是:
O((N + M) * log N)其中M是依赖边的数量。在机试的常规数据规模下,这个复杂度是毫无压力的。N到了10万级别,也依然能跑得动。
4. JavaScript实现:手写小顶堆是绕不开的
JavaScript版本的整体思路和Python完全一致,但有几个JS特有的坑要处理。第一个就是:标准库没有堆。
4.1 手写一个够用的小顶堆
很多人第一反应是用数组加sort模拟:
heap.push(x); heap.sort((a, b) => a - b); const min = heap.shift();N很小的测试下,这样确实能过。但sort每次O(N log N),shift还是O(N),数据一多就废了。既然已经决定用堆策略,不如自己写一个小顶堆,几十行代码的事,稳定高效。
class MinHeap { constructor() { this.heap = []; } isEmpty() { return this.heap.length === 0; } push(x) { const h = this.heap; h.push(x); let i = h.length - 1; while (i > 0) { const p = (i - 1) >> 1; if (h[p] <= h[i]) break; [h[p], h[i]] = [h[i], h[p]]; i = p; } } pop() { const h = this.heap; const top = h[0]; const last = h.pop(); if (h.length) { h[0] = last; let i = 0; while (true) { let l = i * 2 + 1; let r = i * 2 + 2; let m = i; if (l < h.length && h[l] < h[m]) m = l; if (r < h.length && h[r] < h[m]) m = r; if (m === i) break; [h[i], h[m]] = [h[m], h[i]]; i = m; } } return top; } }这个堆实现虽然简短,但该有的都有。push时从底部上浮,pop时把最后一个元素放到堆顶再下沉,每次都能在O(log N)内完成操作。
4.2 完整JavaScript代码
const readline = require('readline'); const rl = readline.createInterface({ input: process.stdin, output: process.stdout }); const tokens = []; rl.on('line', line => { line.trim().split(/\s+/).forEach(t => { if (t.length) tokens.push(t); }); }); rl.on('close', () => { let idx = 0; const outputs = []; while (idx < tokens.length) { const n = parseInt(tokens[idx++]); const cost = new Array(n).fill(0); const pres = Array.from({ length: n }, () => []); for (let i = 0; i < n; i++) { const tid = parseInt(tokens[idx++]); const c = parseInt(tokens[idx++]); const k = parseInt(tokens[idx++]); const ps = []; for (let j = 0; j < k; j++) { ps.push(parseInt(tokens[idx++])); } cost[tid] = c; pres[tid] = ps; } const result = solve(n, cost, pres); if (!result) { outputs.push('ERROR'); } else { outputs.push(result.order.join(' ')); outputs.push(String(result.totalTime)); } } console.log(outputs.join('\n')); }); function solve(n, cost, pres) { const adj = Array.from({ length: n }, () => []); const indeg = new Array(n).fill(0); for (let i = 0; i < n; i++) { for (const p of pres[i]) { adj[p].push(i); indeg[i]++; } } const heap = new MinHeap(); for (let i = 0; i < n; i++) { if (indeg[i] === 0) heap.push(i); } const start = new Array(n).fill(0); const order = []; let totalTime = 0; while (!heap.isEmpty()) { const u = heap.pop(); order.push(u); const finishU = start[u] + cost[u]; totalTime = Math.max(totalTime, finishU); for (const v of adj[u]) { if (finishU > start[v]) start[v] = finishU; indeg[v]--; if (indeg[v] === 0) heap.push(v); } } if (order.length !== n) return null; return { order, totalTime }; }4.3 JS版三个值得说的细节
第一,Node环境下读取输入。我用了readline把所有token收集到数组,再统一解析。这样处理的好处是:不管测试数据的换行格式是Windows还是Linux,也不管是空格分隔还是换行分隔,都能稳定读取。这个方法在机试里非常实用,我后来刷其他题也一直沿用。
第二,new Array(n).fill(0)做数组初始化没问题,但建二维数组时千万别写new Array(n).fill([])——这样所有元素会指向同一个数组,改一个全变。要写Array.from({ length: n }, () => [])。
第三,手写堆的pop方法里,注意h.pop()弹出的是最后一个元素,但堆顶保留的仍是那个被弹出的元素引用。代码里先取top = h[0],再last = h.pop(),最后重新把last放到h[0]并下沉。顺序不能乱。我一开始写成先pop再取top,结果堆空时h[0]变成undefined,排查了好一阵子。
5. 边界用例、自测习惯和通过率的真相
代码能跑通样例只是第一步。上机考试真正拉开差距的,是你有没有把边界情况想全。
5.1 环检测还有一种隐蔽形式:自环
最常见的环是A依赖B、B依赖A这种双向依赖。还有一种容易被忽略的:任务自己依赖自己,比如输入2 0 1 2表示任务2依赖自己。这种自环用拓扑排序同样能识别出来——因为任务2永远无法入度清零,最终order.length一定小于N,输出ERROR。
很多同学看到“环”只想到互相依赖,忘了自环,结果代码没覆盖到。好在拓扑排序这个方案天然免疫这类问题,不需要单独写判断。
5.2 单任务和无依赖的极端情况
N=1、且它没有前置依赖时,拓扑序只有一个元素,总耗时就是它自己的执行时间。这个用例用来验证数组初始化和边界输出。
另一种情况:所有任务都互不依赖,那图的入度全部为0,堆里一开始就塞满了N个节点。此时弹出的顺序就是编号升序,总耗时是所有任务里最大的那个耗时。这个用例能验证“同时开始按编号升序”的处理逻辑。
5.3 多组测试用例一起跑
机试的测试数据经常不止一组,或者系统会同时跑多个样例。我的Python和JS代码都支持无限读取,直到token耗尽。这里有个小习惯值得分享:写完代码后,把两三个样例拼接在一起,中间不要用额外分隔符,直接顺序粘贴进测试环境跑一遍。如果输出结果各自独立、没有串行错位,说明循环解析的边界是安全的。
5.4 我在提交前固定做的三分钟自测
这一步我每次机试都会做,花不了三分钟,但能挡掉大部分低级失误:
- 用题目给的样例跑一遍,确认输出格式完全一致,包括空格和换行。
- 构造一个最小用例:N=1,耗时任意,确认输出编号和耗时。
- 构造一个环用例,确认输出ERROR。
- 构造一个多组用例拼接的输入,确认循环解析正常。
- 最后看一眼代码里有没有残留调试用的print或console.log。
这套自测做完,基本可以安心提交了。所谓的“一次通过”,在我看来就是建模正确、输入解析稳健、边界覆盖充分的结果,不是靠运气。
6. 从这道题延伸到真实的调度系统
复盘的时候我还想明白一件事:这道题虽然以机试真题的形式出现,但它的模型和真实世界的任务调度非常接近。
你可以把每个任务想象成一段构建流程,把前置依赖想象成编译依赖库的链式关系。构建系统比如Make、Gradle,底层干的都是同一件事:解析依赖图,判断有没有循环依赖,找到合法的执行顺序,尽量并行执行互不依赖的任务。Go语言的go build为什么能自动分析包依赖关系?本质上也是在做DAG拓扑排序。
所以这道题的代码,换个皮就是一套极简版的依赖调度引擎。理解了这一点,刷题就不再是为了应付考试,而是真的在积累工程经验。面试官问“你怎么理解任务编排系统”,你完全可以拿这题的建模思路去回答:建图、判环、拓扑序、找关键路径,四步到位。
我个人在实际编码中还有一个体会:能用一次遍历解决的,不要拆成两套逻辑。这题的判环、排序、算时间,全部在一次堆弹射过程中完成,代码短、状态少、不容易出错。很多人在面试时把这三件事拆成三个函数,反而因为状态传递出错而翻车。保持简单,本身就是一种通过率。