优先级队列 模版题单
优先级队列 (堆)
用到优先级队列 (priority queue) 或堆 (heap) 的题一般需要维护一个动态更新的池,元素会被频繁加入到池中或从池中被取走,每次取走的元素为池中优先级最高的元素 (可以简单理解为最大或者最小)。
用堆来实现优先级队列是效率非常高的方法,加入或取出都只需要 O(log N) 的复杂度。
Python的heapq
模块实现了一个最小堆(小顶堆),底层基于**二叉堆(Binary Heap)**数据结构。以下是其核心实现原理和常用操作的时间复杂度分析:
底层实现原理
-
二叉堆结构
heapq
使用完全二叉树实现堆,即除了最后一层外,每一层都被完全填充,且最后一层的节点从左到右排列。- 堆的特性:每个节点的值都小于或等于其子节点的值(最小堆)。
-
数组存储
- 堆以一维数组形式存储,通过索引计算父子关系:
- 节点
i
的左子节点:2i + 1
- 节点
i
的右子节点:2i + 2
- 节点
i
的父节点:(i - 1) // 2
- 节点
- 堆以一维数组形式存储,通过索引计算父子关系:
-
核心操作
- 上浮(Heapify Up):插入元素时,新元素从数组末尾开始向上调整,直到满足堆性质。
- 下沉(Heapify Down):删除堆顶元素后,将最后一个元素移到堆顶,然后向下调整,直到满足堆性质。
常用操作的时间复杂度
操作 | 时间复杂度 | 说明 |
---|---|---|
heapq.heappush(h, x) | O(log n) | 插入元素到堆中,需执行上浮操作调整堆结构。 |
heapq.heappop(h) | O(log n) | 删除并返回堆顶元素,需执行下沉操作调整堆结构。 |
heapq.heapify(x) | O(n) | 将列表转换为堆,通过从最后一个非叶子节点开始下沉调整,效率高于逐个插入。 |
h[0] | O(1) | 访问堆顶元素(最小值)。 |
关键代码实现(简化版)
以下是heapq
核心操作的简化实现,帮助理解原理:
def heappush(heap, item):heap.append(item)_siftup(heap, len(heap) - 1) # 上浮操作def heappop(heap):lastelt = heap.pop()if heap:returnitem = heap[0]heap[0] = lastelt_siftdown(heap, 0) # 下沉操作return returnitemreturn lasteltdef _siftup(heap, pos):# 上浮:从pos位置开始向上调整newitem = heap[pos]while pos > 0:parentpos = (pos - 1) // 2parent = heap[parentpos]if newitem < parent:heap[pos] = parentpos = parentposcontinuebreakheap[pos] = newitemdef _siftdown(heap, pos):# 下沉:从pos位置开始向下调整startpos = posnewitem = heap[pos]childpos = 2 * pos + 1while childpos < len(heap):rightpos = childpos + 1if rightpos < len(heap) and not heap[childpos] < heap[rightpos]:childpos = rightposheap[pos] = heap[childpos]pos = childposchildpos = 2 * pos + 1heap[pos] = newitem_siftup(heap, pos) # 最后可能需要上浮微调
应用场景
- 优先队列:任务调度、Dijkstra算法等。
- Top-K问题:求最大/最小的K个元素。
- 合并有序序列:多路归并排序。
总结
heapq
通过数组实现二叉堆,保证插入和删除操作的时间复杂度为O(log n),适合处理动态数据集合。其设计巧妙,无需显式维护树结构,通过数组索引即可高效操作,是Python中处理堆相关问题的标准库。
Kth largest/smallest
找数据中第 K 个最大/最小数据是堆的一个典型应用。以找最大为例,遍历数据时,使用一个最小堆来维护当前最大的 K 个数据,堆顶数据为这 K 个数据中最小,即是你想要的第 k 个最大数据。每检查一个新数据,判断是否大于堆顶,若大于,说明堆顶数据小于了 K 个值,不是我们想找的第 K 个最大,则将新数据 push 进堆并 pop 掉堆顶,若小于则不操作,这样当遍历完全部数据后堆顶即为想要的结果。找最小时换成最大堆即可。
kth-largest-element-in-a-stream
设计一个找到数据流中第K大元素的类。
import heapqclass Item:def __init__(self, value):self.value = value# 定义小于比较,使堆变成最大堆def __lt__(self, other):return self.value > other.value # 反向比较max_heap = []
heapq.heappush(max_heap, Item(3))
heapq.heappush(max_heap, Item(1))
heapq.heappush(max_heap, Item(4))# 弹出时取 value
print(heapq.heappop(max_heap).value) # 输出: 4
print(heapq.heappop(max_heap).value) # 输出: 3
import heapq# 创