Python heapq 标准库，涵盖小顶堆原理、核心函数（heapify, heappush, heappop 等）、批量查询及合并功能。通过优先级队列、中位数查找、堆排序等实战案例展示应用场景，并提供最大堆模拟、自定义对象比较及性能优化技巧，帮助开发者高效处理数据流与算法问题。
Python 的 heapq 库是基于堆数据结构实现的标准库模块，它提供了对小顶堆（min-heap）的高效操作支持。堆是一种特殊的完全二叉树结构，其中父节点的值总是小于或等于其所有子节点的值（小顶堆特性）。该库的时间复杂度为 O(log n)，在需要频繁插入和删除最小元素的场景下表现出色。
| 函数名 | 功能描述 | 时间复杂度 | 使用场景 |
|---|
heapify(x) | 将列表 x 转换为堆结构 | O(n) | 列表初始化堆 |
heappush(heap, item) | 向堆中插入新元素 | O(log n) | 动态添加元素 |
heappop(heap) | 弹出并返回最小元素 | O(log n) | 获取最小元素 |
heapreplace(heap, item) | 弹出最小元素并插入新元素 | O(log n) | 替换堆顶元素 |
heappushpop(heap, item) | 先插入再弹出最小元素 | O(log n) | 高效插入弹出 |
代码示例：基础堆操作
import heapq
# 初始化列表
data = [3, 1, 4, 1, 5, 9, 2, 6]
# 将列表转换为堆（原地操作）
heapq.heapify(data)
print(f"堆化后的列表：{data}")
# 输出：[1, 1, 2, 3, 5, 9, 4, 6]
# 向堆中插入元素
heapq.heappush(data, 0)
print(f"插入 0 后的堆：{data}")
# 输出：[0, 1, 2, 1, 5, 9, 4, 6, 3]
# 弹出最小元素
min_element = heapq.heappop(data)
print(f"弹出的最小元素：{min_element}")
# 输出：0
print(f"弹出后的堆：{data}")
# 输出：[1, 1, 2, 3, 5, 9, 4, 6]
| 函数名 | 功能描述 | 时间复杂度 | 适用场景 |
|---|---|---|---|
nlargest(n, iterable) | 返回前 n 个最大元素 | O(n log k) | Top-K 最大元素 |
nsmallest(n, iterable) | 返回前 n 个最小元素 | O(n log k) | Top-K 最小元素 |
代码示例：Top-K 问题解决
import heapq
import random
# 生成测试数据
numbers = [random.randint(1, 1000) for _ in range(100)]
# 获取最大的 5 个元素
largest_5 = heapq.nlargest(5, numbers)
print(f"最大的 5 个元素：{largest_5}")
# 获取最小的 5 个元素
smallest_5 = heapq.nsmallest(5, numbers)
print(f"最小的 5 个元素：{smallest_5}")
# 使用 key 参数进行自定义比较
words = ['apple', 'banana', 'cherry', 'date', 'elderberry']
longest_3 = heapq.nlargest(3, words, key=len)
print(f"最长的 3 个单词：{longest_3}")
# 输出：['elderberry', 'banana', 'cherry']
heapq.merge(*iterables) 函数用于合并多个已排序的输入序列，返回一个排序后的迭代器。
import heapq
# 合并多个有序序列
list1 = [1, 3, 5, 7]
list2 = [2, 4, 6, 8]
list3 = [0, 9, 10]
merged = list(heapq.merge(list1, list2, list3))
print(f"合并后的有序列表：{merged}")
# 输出：[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
import heapq
class PriorityQueue:
def __init__(self):
self._heap = []
self._index = 0 # 用于处理优先级相同的情况
def push(self, item, priority):
"""添加元素到优先级队列"""
heapq.heappush(self._heap, (priority, self._index, item))
self._index += 1
def pop(self):
"""弹出优先级最高的元素"""
if self._heap:
return heapq.heappop(self._heap)[-1]
raise IndexError("优先级队列为空")
def is_empty(self):
return len(self._heap) == 0
# 使用示例
pq = PriorityQueue()
pq.push("任务 A", 3)
pq.push("任务 B", 1) # 最高优先级
pq.push("任务 C", 2)
while not pq.is_empty():
print(f"执行：{pq.pop()}")
# 输出：执行：任务 B → 执行：任务 C → 执行：任务 A
import heapq
class MedianFinder:
def __init__(self):
# 最大堆（使用负数模拟）和最小堆
self.max_heap = [] # 存储较小的一半
self.min_heap = [] # 存储较大的一半
def add_num(self, num):
if not self.max_heap or num <= -self.max_heap[0]:
heapq.heappush(self.max_heap, -num)
else:
heapq.heappush(self.min_heap, num)
# 平衡两个堆
if len(self.max_heap) > len(self.min_heap) + 1:
heapq.heappush(self.min_heap, -heapq.heappop(self.max_heap))
elif len(self.min_heap) > len(self.max_heap):
heapq.heappush(self.max_heap, -heapq.heappop(self.min_heap))
def find_median(self):
if len(self.max_heap) == len(self.min_heap):
return (-self.max_heap[0] + self.min_heap[0]) / 2
else:
return -self.max_heap[0]
# 使用示例
finder = MedianFinder()
for num in [1, 3, 2, 6, 4, 5]:
finder.add_num(num)
print(f"当前中位数：{finder.find_median()}")
import heapq
def heap_sort(iterable):
"""使用堆排序算法对可迭代对象进行排序"""
heap = list(iterable)
heapq.heapify(heap) # 构建最小堆
return [heapq.heappop(heap) for _ in range(len(heap))]
# 排序示例
unsorted_data = [9, 2, 7, 5, 1, 8, 3, 6, 4]
sorted_data = heap_sort(unsorted_data)
print(f"堆排序结果：{sorted_data}")
# 输出：[1, 2, 3, 4, 5, 6, 7, 8, 9]
由于 heapq 默认实现的是最小堆，可以通过存储负值来模拟最大堆：
import heapq
class MaxHeap:
def __init__(self):
self._heap = []
def push(self, item):
heapq.heappush(self._heap, -item)
def pop(self):
return -heapq.heappop(self._heap)
def peek(self):
return -self._heap[0] if self._heap else None
# 最大堆使用示例
max_heap = MaxHeap()
for num in [3, 1, 4, 1, 5]:
max_heap.push(num)
print("最大堆元素弹出顺序:")
while max_heap._heap:
print(max_heap.pop())
# 输出：5, 4, 3, 1, 1
import heapq
class Task:
def __init__(self, name, priority, duration):
self.name = name
self.priority = priority
self.duration = duration
def __lt__(self, other):
# 定义比较规则：优先级高的在前，相同优先级时持续时间短的在前
if self.priority == other.priority:
return self.duration < other.duration
return self.priority > other.priority
def __repr__(self):
return f"Task({self.name}, priority:{self.priority}, duration:{self.duration})"
# 自定义对象堆操作
tasks = [
Task("紧急任务", 3, 2),
Task("普通任务", 1, 5),
Task("重要任务", 2, 3)
]
heap = []
for task in tasks:
heapq.heappush(heap, task)
print("任务执行顺序:")
while heap:
print(heapq.heappop(heap))
| 操作场景 | 推荐方法 | 时间复杂度 | 优势 |
|---|---|---|---|
| 一次性获取 Top-K | nlargest()/nsmallest() | O(n log k) | 代码简洁 |
| 持续插入和弹出 | heappush() + heappop() | O(log n) | 动态高效 |
| 多个有序序列合并 | heapq.merge() | O(n log k) | 内存友好 |
import heapq
# 流式处理大数据集
def process_large_dataset(data_stream, top_n=10):
"""使用堆处理大数据流，只维护 Top-N 元素"""
heap = []
for item in data_stream:
if len(heap) < top_n:
heapq.heappush(heap, item)
elif item > heap[0]: # 对于最大 Top-N，使用最小堆
heapq.heapreplace(heap, item)
return sorted(heap, reverse=True)
# 模拟大数据流处理
import random
data_stream = (random.randint(1, 10000) for _ in range(100000))
top_10 = process_large_dataset(data_stream, 10)
print(f"大数据流中的 Top-10: {top_10}")
Python 的 heapq 库通过提供高效的堆操作函数，在算法优化、数据处理和系统设计等多个领域发挥着重要作用。掌握这些函数的正确使用方法和适用场景，能够显著提升程序的性能和代码的可维护性。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。 在线工具，Markdown转HTML在线工具，online
将 HTML 片段转为 GitHub Flavored Markdown，支持标题、列表、链接、代码块与表格等；浏览器内处理，可链接预填。 在线工具，HTML转Markdown在线工具，online