队列与优先队列（堆）选型指南：从BFS到单调队列、TopK与Dijkstra实战

刚摸队列和优先队列堆这两个结构的时候我一度以为它们只是 BFS 的附属品和sort()的廉价替代品。刷了二十来道题之后才发现完全想反了队列管的是顺序控制堆管的是动态最值这两件事几乎贯穿了所有需要按某种次序处理数据的场景。很多人卡在这一块不是不会调 API而是不知道什么时候该掏哪个结构。这篇就把我在力扣刷题过程中对队列和优先队列堆的整套理解摊开说从什么时候用deque、什么时候非得手写循环队列到单调队列的模板怎么记、heapq的负数取反坑在哪、手写堆到底值不值得花时间。适合已经能写基础 BFS、但对为什么这里要用堆为什么这题堆会超时还比较模糊的同学也适合刷完一轮想系统复盘的人。1. 别把队列当成 BFS 的附属品这两个结构的真实分工1.1 队列解决的问题是处理顺序不是存数据很多人对队列的认知停在先进先出BFS 用。这个描述没错但它掩盖了队列真正的价值。队列的本质是一个待处理任务的调度器你把一批还没处理完的东西按某个顺序排进去然后按这个顺序吐出来。BFS 之所以用它是因为图或者树上的层这个概念天然对应先进先出——先被发现的节点离起点更近应该先被展开。一旦你接受这个视角就会发现队列的用武之地远不止层序遍历。比如滑动窗口里维护候选最大值的单调队列、0-1 BFS 里区分零权边和单位权边的双端队列、设计循环队列时对数组空间的复用全都是顺序控制这个内核在不同约束下的变形。区别只在于允许从哪一端进出、允不允许插队、允不允许丢弃队内元素。我在复盘笔记里给自己总结了一句话普通队列管公平双端队列管优先级插队单调队列管淘汰。这三种语义对应三类题混淆了就会写出又慢又难调的实现。1.2 优先队列解决的问题是每次都要拿到当前最值优先队列堆的定位则完全不同。它不关心元素进来的顺序只保证你能以 O(1) 拿到当前的最值、以 O(log n) 的代价插入和弹出。注意这句话里有个非常容易被忽略的部分堆不保证整体有序。它只保证堆顶是极值其余元素的相对顺序是部分有序——父节点永远优于子节点兄弟之间没有任何约束。这个区别决定了选型。如果你需要在处理过程中反复问现在最大的/最小的/最紧急的是哪个那就是堆如果你只是最后要一个完整排序结果那是sort()如果你要的是按到达顺序依次处理那是队列。还有个经常被搞混的第三类需要同时维护最大和最小比如数据流中位数这时候单堆不够得用两个堆对顶一个存较小的一半大顶堆一个存较大的一半小顶堆靠 sizes 平衡。这个套路后面单独拆。提示判断该不该用堆就问自己一句我是不是在处理过程中反复需要当前最值。如果答案是只在最后需要一次堆就是过度设计。2. 普通队列与双端队列的选择从层序遍历到 0-1 BFS2.1 三层递进遍历、最短路、状态搜索队列相关题的难度大致可以分成三层每一层的实现细节都不一样。第一层是结构遍历比如二叉树层序遍历、N 叉树层序、右视图。这类题的核心是分层写法上用for _ in range(len(q))把当前层一次性取完避免把下一层和当前层混在一起。第二层是无权图最短路比如网格里从起点到终点的最少步数、腐烂橘子扩散的最短时间。这里的队列里存的不是节点本身而是到达该节点的步数,通常把坐标和步数打包成元组或者用外部数组记录距离。我习惯用外部二维数组记dist因为很多题最后要返回整个距离矩阵。第三层是状态搜索比如转盘锁、单词接龙、打开锁的最少次数。这类题的难点不在队列而在状态表示和去重——你得把当前状态编码成一个可哈希的东西塞进visited否则队列会无限膨胀。这里最容易犯的错是入队时才标记 visited正确做法是入队时就标记出队时才标记会导致同一状态被重复入队内存直接爆掉。2.2 0-1 BFS双端队列的正确插法有一类图边权只有 0 和 1 两种。这时候普通 BFS 会失效因为它默认所有边权相同而 Dijkstra 又有点重。标准解法是 0-1 BFS用双端队列边权为 0 的邻居appendleft边权为 1 的邻居append。from collections import deque def bfs_01(grid, start): n, m len(grid), len(grid[0]) INF float(inf) dist [[INF] * m for _ in range(n)] dist[start[0]][start[1]] 0 dq deque([start]) while dq: x, y dq.popleft() for dx, dy, w in ((0, 1, 0), (0, -1, 0), (1, 0, 1), (-1, 0, 1)): nx, ny x dx, y dy if 0 nx n and 0 ny m and dist[nx][ny] dist[x][y] w: dist[nx][ny] dist[x][y] w if w 0: dq.appendleft((nx, ny)) else: dq.append((nx, ny)) return dist这里有个细节值得说清楚为什么边权 0 要插到队头因为双端队列在模拟当前距离相同的一批节点。边权 0 意味着不增加距离新节点和当前节点属于同一层应该优先处理边权 1 意味着距离加一排到后面。这样队列里的dist天然保持单调不减等价于一个小根堆但常数小得多。2.3 什么时候必须自己写循环队列绝大多数刷题场景直接用collections.deque就够了。但有一类题会明确要求用固定大小的数组实现不许用现成队列典型是设计循环队列。它的核心难点是判空和判满head tail到底表示空还是满主流有两种写法。第一种是浪费一个位置容量开k1队空条件head tail队满条件(tail 1) % cap head。第二种是额外维护 size容量就开k判空看size 0判满看size cap。我个人更偏好第二种因为不用在脑子里反复推导取模关系代码可读性明显更好。class MyCircularQueue: def __init__(self, k: int): self.data [0] * k self.cap k self.head 0 self.size 0 def enQueue(self, value: int) - bool: if self.size self.cap: return False self.data[(self.head self.size) % self.cap] value self.size 1 return True def deQueue(self) - bool: if self.size 0: return False self.head (self.head 1) % self.cap self.size - 1 return True def Front(self) - int: return -1 if self.size 0 else self.data[self.head] def Rear(self) - int: return -1 if self.size 0 else self.data[(self.head self.size - 1) % self.cap]Rear()那一行的取模是最容易写错的尾元素位置是head size - 1不是head size。我第一次写的时候就是这里少减了个一测了好几组才发现。还有一类是用两个栈实现队列用队列实现栈本质是在考察你对这两种结构约束条件的理解栈只能操作一端队列两端分工。两个栈实现队列的思路是入队压 in 栈出队时如果 out 栈空就把 in 栈全倒过去均摊复杂度 O(1)。这类题不难但面试里出现频率不低值得手写一遍。3. 单调队列维护窗口内候选最值而不是全部元素3.1 它到底在维护什么单调队列这个词听起来玄其实就一句话队列里的元素值保持单调队头永远是当前窗口的最值。关键在于淘汰——当新元素进来时把所有比它更差且更早过期的元素直接从队尾踢掉。举个直观的例子。求滑动窗口最大值窗口向右滑新来一个数x。如果队尾有个数y x那么y永远不可能成为未来的最大值了——它比x小而且比x先离开窗口两头都不占优。既然毫无价值直接弹掉。弹完之后队列从队头到队尾是单调递减的队头就是答案。这里的两头都不占优是整个单调队列的核心逻辑也是必须理解透彻的地方。如果你只是背模板遇到变体比如求最小值、或者带前缀和的变体就很容易翻车。3.2 三个必须吃透的模板变体变体一滑动窗口最大值。队列里存下标而不是值因为要靠下标判断是否越界。from collections import deque def maxSlidingWindow(nums, k): q deque() # 存下标对应的值单调递减 res [] for i, x in enumerate(nums): while q and nums[q[-1]] x: q.pop() # 队尾比新元素小的永久淘汰 q.append(i) if q[0] i - k: q.popleft() # 队头已经滑出窗口 if i k - 1: res.append(nums[q[0]]) return res注意顺序先按值淘汰队尾再入队最后按位置淘汰队头。这三步的顺序如果打乱某些边界情况就会出错比如k 1的时候。变体二绝对值差不超过限制的最长连续子数组。这题需要同时维护窗口最大值和最小值所以要开两个单调队列一个递减一个递增。窗口右边界每扩一格两个队都要更新当max - min limit时收缩左边界直到满足条件。这类题的关键是收缩时怎么同步更新两个队列——其实不用特殊处理只要在每次循环开头按位置把队列头部过期元素弹掉即可。变体三和至少为 K 的最短子数组。这题是单调队列里最绕的一个因为它结合了前缀和。核心观察是如果pre[j] - pre[i] K且j i那么pre[j] pre[i]意味着pre[j]对后续任何位置都不可能比pre[i]更优它更大能凑出的子数组更短不了。所以队列维护的是前缀和的递增序列。from collections import deque def shortestSubarray(nums, k): n len(nums) pre [0] * (n 1) for i, x in enumerate(nums): pre[i 1] pre[i] x q deque() ans n 1 for i in range(n 1): while q and pre[i] - pre[q[0]] k: ans min(ans, i - q.popleft()) while q and pre[q[-1]] pre[i]: q.pop() q.append(i) return ans if ans n else -1这两段 while 的顺序不能换先尝试用当前i去消费队头能凑出合法区间就把队头弹掉因为后面的i只会更长不可能更优然后再维护队列的单调性。3.3 单调队列和堆在同一道题上的取舍滑动窗口最大值这题用堆也能做维护一个(-value, index)的大顶堆每次取堆顶检查下标是否过期。复杂度是 O(n log n)。单调队列是 O(n)。那是不是所有场景都该用单调队列不是。我的判断标准是如果窗口里只需要最值、且元素只增不减单调滑动用单调队列如果窗口是动态增删、删的位置不固定堆更省心。因为单调队列的淘汰逻辑依赖元素按顺序进入这个前提一旦允许任意位置删除队内被提前淘汰的元素可能复活整个结构就崩了。堆的另一个优势是支持取前 k 个这类需求单调队列只能给你一个最值。所以别迷信 O(n)先看清需求。4. heapq 的真实语义它只承诺堆顶不承诺有序4.1 最小堆约定与负数取反的三个陷阱Python 的heapq只提供最小堆。要做最大堆标准做法是存负数入堆-x出堆再取反。听起来简单实际有三个坑。坑一混用正负号。如果你在同一个堆里既存x又存-x比较逻辑会彻底错乱。我的习惯是给堆变量起名max_heap然后所有出入操作都严格配对写成-x绝不偷懒。坑二对负数取反后语义翻转。比如数组里有负数你想求第 k 大直接用-x存进去是对的但如果某道题本身要求最大堆里存的是成本成本可能为负那么-(-5) 5堆顶对应的是最小的绝对值……这种场景我建议干脆写成最小堆存成本然后在弹出时取反语义更清晰。坑三浮点数取反的精度。-0.0和0.0在比较时相等heapq不会报错但可能引发排序歧义。涉及浮点优先级的场景加一个自增计数器做第二关键字比较稳妥。4.2 元组比较失效的场景与计数器方案heapq比较的是元组的字典序。这就带来一个经典问题如果你入堆的是(priority, task)而两个 task 优先级相同时Python 会去比较task对象本身。如果 task 是字典、列表或者自定义类且没实现__lt__直接抛TypeError。解决方案有两个。方案一是加一个全局自增计数器import heapq import itertools counter itertools.count() heap [] heapq.heappush(heap, (priority, next(counter), task))这样第二关键字永远唯一绝不会比到 task 上。方案二是给自定义类实现__lt__只比较优先级字段。我一般用方案一因为改数据结构不如加个计数器省事而且计数器的自增顺序还能保证同优先级先进先出这个稳定性质。还有个小技巧如果你想让同优先级按后进先出排就把计数器取负即(priority, -next(counter), task)。这个在模拟任务调度的时候挺有用。4.3 heapify、heappushpop、heapreplace 的性能与语义差异这三个函数经常被混用其实差别不小。函数语义复杂度典型场景heapify(lst)原地把列表整理成堆O(n)一次性给一批数据建堆heappushpop(h, x)先 push 再 pop返回最小值O(log n)维持固定大小 k 的堆新元素进、最小出heapreplace(h, x)先 pop 再 push返回被弹出的最小值O(log n)已经确定要替换堆顶时heappushheappop分开写两次操作O(log n) × 2逻辑需要分步时关键区别在于heappushpop和heapreplace的执行顺序不同前者是先入后出如果新元素比堆顶还小它会立刻被弹出来堆内容不变后者是先出后入无论新元素多小堆顶一定会被弹掉。求 TopK 时用heappushpop做守门员特别合适——堆还没满就heappush满了就heappushpop比写 if-else 干净。至于heapify它的 O(n) 是很多人忽略的性能点。如果你有一个长度为 n 的数组要建堆逐个heappush是 O(n log n)heapify是 O(n)。数据量大的时候差别很明显而且代码还短一行。5. 堆的四条高频套路TopK、多路归并、双堆对顶、堆上贪心5.1 TopK为什么求第 K 大要用小顶堆这是最反直觉的一个点也是面试最爱问的。求第 K 大的元素你用什么堆很多人脱口而出大顶堆然后就错了。正确的做法是维护一个大小为 K 的小顶堆。理由遍历数组每次把元素塞进堆堆大小超过 K 就弹掉堆顶当前最小值。遍历结束后堆里留下的是最大的 K 个元素而堆顶恰好是这 K 个里最小的那个也就是整个数组的第 K 大。import heapq def find_kth_largest(nums, k): heap [] for x in nums: heapq.heappush(heap, x) if len(heap) k: heapq.heappop(heap) return heap[0]复杂度 O(n log k)比全排序的 O(n log n) 好而且空间只用 O(k)。如果 k 很小、n 很大这个优化的收益非常可观。求前 K 个高频元素是同一套路的变体只是比较的关键字从值变成了频率。我的写法是先Counter统计再用heapq.nlargest(k, freq.items(), keylambda p: p[1])。nlargest内部就是小顶堆实现比自己写更省事。5.2 多路归并堆里放当前参赛者合并 K 个有序链表、合并 K 个有序数组、在有序矩阵里找第 K 小本质都是同一件事有 K 个有序序列每次要选出最小的那个头元素。import heapq def merge_k_lists(lists): heap [] for i, node in enumerate(lists): if node: heapq.heappush(heap, (node.val, i, node)) dummy tail ListNode(0) while heap: val, i, node heapq.heappop(heap) tail.next node tail node if node.next: heapq.heappush(heap, (node.next.val, i, node.next)) return dummy.next元组里的i是必需的因为链表节点对象之间不可比较。这个细节我已经在好几道多路归并的题上栽过现在成了肌肉记忆。复杂度是 O(N log K)N 是总元素数。如果 K 很小这就是最优解如果 K 和 N 同量级那还不如直接全部收集起来排序。5.3 双堆对顶数据流中位数的维护逻辑中位数要求随时能拿到中间那个数单堆做不到必须用两个堆small存较小的一半大顶堆用负数实现large存较大的一半小顶堆。不变式是len(small) len(large)或len(small) len(large) 1。import heapq class MedianFinder: def __init__(self): self.small [] # 大顶堆存负数 self.large [] # 小顶堆 def addNum(self, num: int) - None: heapq.heappush(self.small, -num) heapq.heappush(self.large, -heapq.heappop(self.small)) if len(self.large) len(self.small): heapq.heappush(self.small, -heapq.heappop(self.large)) def findMedian(self) - float: if len(self.small) len(self.large): return -self.small[0] return (-self.small[0] self.large[0]) / 2addNum这三行是整个结构最精妙的地方先无脑进small再把它最大的那个挪到large保证small里所有元素都不大于large里的最后调整 sizes。这三步顺序不能变换任何一个都会破坏不变式。双堆的变体很多滑动窗口中位数需要配合延迟删除、两个有序数组的中位数这题其实应该用二分等。但核心骨架就这一套。5.4 堆上贪心与图算法里的堆堆在贪心题里的用法是每次都拿当前最优。典型场景会议室安排按结束时间取最早结束的、任务调度按剩余执行次数取最多的、K 次取反后最大化数组和每次取最小的取反。图算法里最重量级的应用是 Dijkstra。用优先队列优化后复杂度从 O(V²) 降到 O((VE) log V)。但 Python 的heapq不支持 decrease-key所以标准写法是懒惰删除import heapq def dijkstra(graph, src, n): INF float(inf) dist [INF] * n dist[src] 0 pq [(0, src)] while pq: d, u heapq.heappop(pq) if d dist[u]: continue # 过期条目直接丢弃 for v, w in graph[u]: nd d w if nd dist[v]: dist[v] nd heapq.heappush(pq, (nd, v)) return distif d dist[u]: continue这一行是全部关键。同一个节点可能被多次入堆只有距离等于当前最优的那次是有效的其余都是历史遗留。少了这一行某些图上的复杂度会退化甚至死循环。6. 复盘六个让我调试到凌晨的细节坑6.1 堆里存值还是存索引这个问题我纠结了很久。存值的好处是代码短坏处是无法判断元素是否已过期。比如滑动窗口最大值用堆做你必须知道堆顶元素的下标是否还在窗口内那就只能存(value, index)。判断标准很简单如果堆里的元素有有效期或者需要和外部状态对应存索引如果只是纯粹取最值存值就够了。求第 K 大那种题存值完全没问题因为元素之间没有时效性。6.2 单调队列忘记存索引导致的越界我最初写单调队列习惯存值不存下标觉得省事。结果遇到窗口大小固定的题就开始出错——因为没法判断队头是否已经滑出窗口。更隐蔽的问题是如果数组里有重复值你根本分不清队头那个值和窗口边界的关系。后来我定了个死规矩单调队列一律存下标取值的时机再nums[index]。代价是多一层索引收益是永不越界。6.3 大顶堆取负时把 0 和负数搞混有一次做一道合并石头的最低成本类型的题我用-cost存大顶堆但成本本身可能是 0。结果-0和0在比较时一样堆的稳定性被破坏输出顺序不对。修法很简单加第二关键字索引(-cost, i)但当时排查花了快四十分钟。这个教训是任何用负数模拟大顶堆的场景都要考虑同值情况下的第二关键字。要么加索引要么加计数器。6.4 用 list 当队列做 pop(0)这是新手最常犯的性能错误。list.pop(0)是 O(n)因为它要把后面所有元素前移一位。如果你在一个循环里对长度为 n 的列表反复pop(0)整体就是 O(n²)。BFS 里节点数一大直接超时。正确做法是用collections.dequepopleft()是 O(1)。这个坑我踩过一次之后现在只要看到队列两个字就条件反射写deque。6.5 建堆用错方法前面提过heapify是 O(n)逐个heappush是 O(n log n)。有一次做最接近原点的 K 个点我先用列表推导生成所有距离然后循环heappush跑了 800ms 才过。改成heapify之后降到 300ms 以内。不是所有场景都能用heapify因为它要求你一次性把所有元素都放进堆。但对于给定静态数组求 TopK这类题完全可以用heapq.nsmallest(k, points, key...)内部就是heapify加筛选。6.6 状态去重的时机前面提过一次这里再强调BFS 的状态去重必须在入队时标记不能在出队时标记。出队时标记意味着同一个状态可能被多个父节点重复入队队列长度指数级膨胀。这个坑在打开转盘锁单词接龙这类题上特别致命。我当时的做法是在出队时visited.add本地小数据能过测试用例一到三位数就超时。改成入队即标记之后立刻通过。7. 手写堆到底值不值sift up 与 sift down 的完整模板7.1 什么情况下值得自己写我的答案是刷题阶段值得写两遍实战阶段九成场景用现成的。值得手写的场景有三类。第一类是题目明确要求不能使用内置堆第二类是需要decrease-key或者delete-any这类高级操作内置库不支持第三类是需要堆里存自定义结构且比较逻辑很复杂包一层比到处写元组更清晰。至于刷题阶段为什么要写两遍是因为只有自己实现过一遍你才会真正理解堆只保证父子关系这件事。用过heapq的人很多能说清sift_down为什么从最后一个非叶子节点开始的人不多。7.2 一个大顶堆的完整实现class MaxHeap: def __init__(self): self.h [] def push(self, x): self.h.append(x) self._sift_up(len(self.h) - 1) def pop(self): top self.h[0] last self.h.pop() if self.h: self.h[0] last self._sift_down(0) return top def peek(self): return self.h[0] if self.h else None def _sift_up(self, i): while i 0: p (i - 1) // 2 if self.h[p] self.h[i]: break self.h[p], self.h[i] self.h[i], self.h[p] i p def _sift_down(self, i): n len(self.h) while True: left, right, big 2 * i 1, 2 * i 2, i if left n and self.h[left] self.h[big]: big left if right n and self.h[right] self.h[big]: big right if big i: break self.h[i], self.h[big] self.h[big], self.h[i] i bigpop的实现有个细节先pop()出最后一个元素如果堆非空才把它放到堆顶并下沉。如果写成先取出堆顶再pop()最后一个在堆只有一个元素时会出问题——pop()之后堆空了再访问self.h[0]就 IndexError。_sift_down里的big变量名不是随意起的它记录的是父节点、左孩子、右孩子三者中的最大值下标。三个 if 的顺序无所谓但最后一定要判断big i相等说明已经满足堆性质可以退出。少了这个判断会死循环。7.3 手写堆和 heapq 的性能对比说实话纯 Python 手写的堆比heapq慢不少因为heapq是 C 实现的。我做过一个粗略测试10 万次 push/pop 的操作heapq大概 0.05 秒手写的大概 0.35 秒差了七倍。所以除非有特殊需求生产环境一律用内置库。但手写堆在刷题时有个隐性收益它能帮你验证你对堆的理解。比如数据流的中位数这题如果你能手写双堆并且保证 sizes 平衡那说明你真的懂了而不是背了个模板。8. 题单怎么分组、模板怎么沉淀才不忘8.1 我自己的分组方式刷完这个专题之后我把题单重新按结构职责分了一遍而不是按难度或者题号。四组分组核心特征代表题型队列基础先进先出、分层处理层序遍历、网格最短路、状态搜索双端队列两端进出、优先级插队0-1 BFS、循环队列设计、双栈实现队列单调队列淘汰劣质候选、维护窗口最值滑动窗口最大值、限制差值的最长子数组、前缀和型最短子数组优先队列动态最值、多路归并、双堆对顶第 K 大、合并 K 个有序结构、数据流中位数、Dijkstra这样分的好处是遇到新题时能快速定位这题属于哪一组然后直接调那一组的模板。以前我是按简单/中等/困难刷的刷完感觉啥都会一遇到变形就懵。8.2 模板怎么记才不会忘我的做法是给每个模板配一个最小可运行骨架而不是完整题解。比如单调队列的骨架就三行注释# 1. 按值淘汰队尾保持单调 # 2. 新元素入队 # 3. 按位置淘汰队头处理过期这三行的顺序本身携带信息比背一整段代码管用。真正写题的时候只要想起这三步代码自然就出来了。完整题解反而容易让人形成抄写依赖换个场景就不会用。还有一个习惯每道题做完之后我会在笔记里写一句如果这题改一个条件我的解法会不会崩。比如滑动窗口最大值我会问自己如果窗口大小不固定会怎样——答案是单调队列仍然能用只是过期判断要改再问如果要求窗口第二大呢——答案是单调队列不够得上平衡树或者两个堆。这种自我追问比刷十道同类题更能加深理解。提示模板背不下来的根本原因通常是没搞清楚每个步骤在防什么错。把每一步对应的反例想明白模板就变成推理结果而不是记忆负担。最后分享一个我自己在用的小技巧。刷完一个专题我会挑一道最有代表性的题从零开始手写一遍完整解法中间不看任何资料写完再对照之前的代码找差异。差异点往往就是理解不到位的地方。这个方法比反复重刷题单效率高得多尤其是队列和堆这种模板短但细节多的专题差异往往就藏在取模、越界判断、去重时机这几个地方。