模拟消息队列这个标题第一次在华为OD B卷题单里刷到的时候我第一反应是完了是不是要手写一个Kafka或者RabbitMQ消息队列诶听着就是分布式中间件那一挂的。结果认认真真把题读完才发现题目要的只是一个纯数据结构层面的模拟若干个队列、若干条消息、三种操作仅此而已。这是一道非常典型的名字唬人、内核基础的100分题也是华为OD机试里那种人人都能会、但未必人人拿满分的送分题。如果你正在准备华为OD机试或者想通过真题巩固Python队列操作这篇分享会给你一个可以直接抄作业的解法以及我实际刷题和测试过程中踩过的几个坑。1. 华为OD B卷里的这道题到底在考什么1.1 100分题的真实定位华为OD机试的题目一般分几个梯度100分题承担的角色是基础能力筛选你代码能不能写利索、边界条件能不能想全、常用数据结构能不能熟练用。它不要求你掌握动态规划或者复杂的图论算法但如果连基础的模拟题都写不稳后面的200分题基本也没时间做完。模拟消息队列这类题在B卷里属于非常标准的模拟题——题目描述往往很长包装也很唬人会讲一堆消息队列在分布式系统里的应用背景甚至可能出现生产者消费者异步解耦这些名词。但剥掉这些包装核心考的就是你能不能把队列这个数据结构的FIFO特性用代码准确表达出来。1.2 题目大意与输入输出格式以我看到的B卷版本为例题目的核心模型是这样的系统中有若干条消息队列每条队列用一个整数ID标识。你需要支持三种操作PUSH 队列ID 消息ID将一条消息放入指定队列的队尾POP 队列ID从指定队列的队首取出一条消息并输出其消息ID若队列为空输出 -1SIZE 队列ID输出指定队列当前的消息数量若队列不存在输出 0输入第一行是一个正整数 N代表操作次数。接下来 N 行每行一条操作指令。输出要求是每条POP和SIZE指令输出一行结果。不同批次的题面细节可能会有出入——比如操作名可能是PUT/GET/COUNT消息ID可能是字符串队列ID也可能不是从1开始连续——但核心模型不变解题思路可以完全复用。1.3 样例推演先在纸上走一遍我建议刷题时遇到任何模拟题第一步不是写代码而是手动模拟一遍样例。比如这样一个输入7 PUSH 1 100 PUSH 1 200 POP 1 SIZE 1 POP 1 POP 1 SIZE 1手推过程PUSH 1 100队列1变为[100]PUSH 1 200队列1变为[100, 200]POP 1队首100弹出输出100队列1变为[200]SIZE 1长度为1输出1POP 1队首200弹出输出200队列1变为[]POP 1队列为空输出 -1SIZE 1队列为空但存在长度为0输出0手动推完这遍整个程序的逻辑框架就已经清晰了。很多同学代码写混乱问题就出在没在纸上理清楚空队列和不存在的队列这两种情况的区别后面我会专门讲这个点。2. 拆掉消息队列的包装本质是FIFO模拟2.1 为什么核心是FIFO队列最本质的特性就是先进先出First In First Out也就是先进入队列的元素会先被取出。题目里的消息队列不管你前面包装了多少异步解耦的概念落到数据结构层就是一件事push往队尾放pop从队首取。# 放入 queue.append(item) # 取出 first queue.popleft()这里最容易犯的错误是把队列和栈搞混。栈是后进先出Python 的list.append()和list.pop()天然就是栈的操作组合。如果你不小心用了queue.pop()而不是queue.popleft()那你就把一个 FIFO队列 写成了 LIFO栈输出结果会完全不对。这种错误特别隐蔽因为代码看起来是正常的逻辑也没问题但就是不满足题目要求。2.2 Python里队列数据结构的选型对比Python里实现队列有好几种选择我列个表对比一下实现方式队尾插入队首弹出适用场景list.append()list.pop(0)O(1)O(n)需要移动所有元素数据量小不推荐collections.deque.append()deque.popleft()O(1)O(1)算法题首选queue.Queue()O(1)O(1)多线程场景才需要有锁开销自定义链表O(1)O(1)没必要Python里重复造轮子我在第一章的时候就强调过机试环境里 N 可能到 10^5 甚至 10^6 量级。如果用list.pop(0)每一次弹出都要把后面所有元素往前挪一位时间复杂度是 O(n)整个程序最坏会退化到 O(N²)。数据量一大超时几乎是必然的。2.3 多队列管理字典就是天然的队列仓库题目不是说只有一条队列而是若干条队列。这意味着你不能只维护一个deque你需要一种能根据队列ID快速找到对应队列的数据结构。在这个场景下Python的字典dict是最合适的键是队列ID值是对应的deque对象。你可以把字典理解成一个仓库每个仓库格子有一个编号格子里放一个真实的队列。queues {} if qid not in queues: queues[qid] deque() queues[qid].append(msg)有的同学可能会想我一开始就把所有可能出现的队列ID都初始化好行不行比如题目说队列ID从1到M那我创建{i: deque() for i in range(1, M1)}。这样做的问题在于如果M很大比如10^6光是初始化字典就要浪费大量时间和内存。用惰性创建的方式——只有第一次PUSH到某个队列时才创建它——才是最优的。3. 代码实现40行Python搞定三种操作3.1 完整代码可以直接跑下面是我在机试环境里采用的完整实现你可以直接复制去OJ上验证import sys from collections import deque def main(): data sys.stdin.read().splitlines() if not data: return n int(data[0].strip()) queues {} output [] for i in range(1, n 1): line data[i].strip() if not line: continue parts line.split() op parts[0] if op PUSH: qid int(parts[1]) msg int(parts[2]) if qid not in queues: queues[qid] deque() queues[qid].append(msg) elif op POP: qid int(parts[1]) q queues.get(qid) if q is None or not q: output.append(-1) else: output.append(str(q.popleft())) elif op SIZE: qid int(parts[1]) q queues.get(qid) if q is None: output.append(0) else: output.append(str(len(q))) sys.stdout.write(\n.join(output)) if __name__ __main__: main()3.2 逐步拆解主循环与操作分发这段代码的核心是一个主循环遍历每一行操作指令通过op判断当前操作类型分发到不同的分支去处理。我逐段说下关键点sys.stdin.read().splitlines()一次读入所有行避免逐行调用input()的开销。queues {}用来管理所有队列。output []用来暂存所有输出结果最后一次性写入标准输出。PUSH分支里用了惰性创建if qid not in queues: queues[qid] deque()。POP分支用queues.get(qid)而不是queues[qid]这样如果队列ID不存在返回的是None而不是抛KeyError。SIZE分支同样用get队列不存在就输出0存在就输出长度。3.3 这段代码里藏着的几个关键设计你可能觉得这段代码很简单但简单背后有几个设计决策是值得展开说说的。第一为什么用queues.get(qid)而不是if qid in queues其实两种写法都能用但get写法更紧凑它把判断队列是否存在和获取队列对象两步合成了一步。如果先if qid in queues: q queues[qid]再else手动处理代码会多出几行逻辑上也更容易出错。第二为什么POP分支要同时判断q is None or not qq is None表示队列ID从来没有出现过not q表示队列存在但是空的。这两种情况在输出上是一样的——输出 -1——但它们的语义不同。不存在的队列你连查询它里面有什么的资格都没有存在的空队列说明曾经有过消息但都被消费完了。如果不把这两种情况分开判断直接用queues[qid]去访问不存在的队列会直接让程序崩溃。第三为什么SIZE分支对空队列不做特殊处理这是很多同学会纠结的地方。队列为空长度为0len(q)天然返回0所以SIZE对空队列的输出就是0。而队列不存在时你不能调用len()所以要单独用q is None判断输出0。这两条路径得到的输出数值是一样的但代码上必须有区别——这就是边界条件处理的精细度。4. 测试驱动不只要跑通样例边界情况必须全部覆盖4.1 常规样例验证把题目给的样例输入丢进程序输出是100 1 200 -1 0和手推结果完全一致。到这一步只说明主流程没大问题真正的考验在边界情况。4.2 队列不存在的场景4 SIZE 99 POP 99 PUSH 1 5 SIZE 1手动分析SIZE 99队列99从未出现过输出0POP 99队列99不存在输出 -1PUSH 1 5队列1里放入消息5SIZE 1输出1程序输出0 -1 1这个用例覆盖了两个分支里最容易被忽略的场景。很多人在写SIZE时直接output.append(str(len(queues[qid])))遇到SIZE 99就直接KeyError崩溃了。4.3 连续POP直到空队列的边界6 PUSH 10 1 PUSH 10 2 POP 10 POP 10 POP 10 SIZE 10前两个POP分别弹出1和2第三个POP时队列已经空了输出 -1。最后SIZE 10输出0。这个用例验证的是队列从有到空的完整生命周期。程序输出1 2 -1 0跟上文代码跑出来的结果一致说明空队列判断逻辑是正确的。4.4 大量数据输入的性能测试我自己在本地用Python脚本生成过10万条随机操作去压测这个程序deque的popleft()是 O(1)总共10万个操作运行时间大概几十毫秒级别完全不会有超时的风险。如果你非要用list.pop(0)同样的10万次操作时间会飙到几秒甚至十几秒那就是典型的超时警告。所以选对数据结构对这道题来说不是优化技巧而是生死线。5. 机试实战避坑手册这些细节能救你10分5.1 输入解析的坑别用input()硬扛大数据我记得第一次在OJ上刷类似题目时习惯性地用了input()逐行读取结果跑到第 5000 行的时候明显感觉到卡顿。Python 的input()本身就是相对慢的当 N 达到 10^5 量级这一丁点性能差距就可能成为超时的最后一根稻草。正确做法是import sys data sys.stdin.read().splitlines()一次把所有输入读进内存然后对列表逐行处理。splitlines()会自动按行分割而且会保留行内容比split(\n)更规范。如果操作数真的巨大比如 N 超过 10^6可以考虑用sys.stdin.buffer.read()读二进制再decode()转字符串。但一般情况下sys.stdin.read()就足够了不要过度优化。5.2 输出的坑别每算一个就print一次有些同学写代码时喜欢每得到一个结果就print()一次。这在小数据时没什么感觉但机试环境的IO性能差别会在大输出量时暴露问题。我的习惯是output [] output.append(str(result)) sys.stdout.write(\n.join(output))把结果全部收集到一个列表最后一次性用换行符拼起来输出。这样做还有一个附加好处调试时可以直接print(output)看到所有中间结果排查问题更方便。5.3 防御性编写的习惯代码风格也是隐形的分数华为OD机试不是只判题目对错代码的可读性、健壮性也会影响综合评分至少在面试复盘的时候面试官会看你的代码风格。几个小习惯保留main()函数和if __name__ __main__:入口不要把所有逻辑堆在最外层。操作符和变量之间加空格queues[qid].append(msg)这样的写法比queues[qid].append(msg)可读性好太多。每一段逻辑加一行注释说明这个分支在干什么。尽量不写花哨的一行流代码机试是看稳定性的不是看炫技的。6. 升级方向如果题目稍微变形你还能不能接住6.1 带优先级的消息队列如果题目改成每条消息有一个优先级POP操作取出的是优先级最高的消息而不是最早进入的那数据结构就需要换成heapq实现的优先队列。import heapq # 入队 heapq.heappush(queue, (priority, msg_id)) # 出队取出优先级最高的 _, msg heapq.heappop(queue)这种变体在B卷的200分题里偶尔会出现。核心思想不变变的是用哪个数据结构承载队列。如果遇到这个变体你要立刻联想到heapq而不是deque。6.2 带容量限制的队列如果题目规定每条队列最多只能放 K 条消息PUSH时如果队列已满可能要拒绝入队并输出 -1或者丢弃队首消息。这种变体其实就是在PUSH分支增加一个长度判断if len(q) K: output.append(-1) else: q.append(msg)思路依然很直接本质上还是队列模拟只是多了一个条件分支。6.3 生产者-消费者模型的阻塞问题如果题目加一个场景POP一个空队列时不是输出 -1而是等待直到有消息入队才返回那就变成一个阻塞队列问题。这种情况在OD机试里很少见因为涉及时间维度的模拟复杂度会高很多。万一碰到可以把等待理解为把POP操作先挂起等后续的PUSH操作唤醒它。实现上可以用延迟处理的思路先把挂起的操作记录下来等对应队列有消息了再按顺序处理。不过说实话机试里遇到这种题的概率很低如果真的遇到了说明你已经进入后半程的高分区题目了。6.4 我的一个实操建议刷华为OD真题的时候建议把同一类型题归个类。比如模拟消息队列和LRU缓存任务调度器本质上都是用一个或多个容器去忠实模拟题目描述的过程。你把其中一道题的代码思路吃透其他模拟题其实就是换了个壳。我个人的刷题顺序是先把这一类的模拟题都找出来集中在一天内做完边做边总结它们之间共同的套路——读输入、核心容器、循环分支、边界判断、收集输出这套流程在任何模拟题里都是通用的。最后分享一个我自己用得很顺的调试技巧在代码里保留data sys.stdin.read().splitlines()之后先用一行# print(data[:20])打印前20行输入确认数据读对了再往下写。这个习惯帮我避免过至少三次题目读错但代码写得很嗨的惨案。机试时时间就是分数这种小技巧能在关键时刻帮你稳住节奏。