Python堆与优先队列 — heapq模块/最大堆/K路合并/中位数查找堆是一种完全二叉树结构Python的heapq模块实现的是最小堆。堆常用于优先队列、Top-K问题、任务调度等场景。1. heapq 基本操作import heapq# heapq提供对Python列表的就地堆操作data [5, 3, 8, 1, 9, 2]heapq.heapify(data) # 将列表转换为最小堆原地操作print(最小堆:, data) # [1, 3, 2, 5, 9, 8]# heappush: 向堆中插入元素保持堆性质heapq.heappush(data, 0)print(插入0后:, data) # [0, 3, 1, 5, 9, 8, 2]# heappop: 弹出并返回堆顶元素最小值smallest heapq.heappop(data)print(弹出最小值:, smallest) # 0print(弹出后:, data)# heappushpop: 先push再pop比先后调用更高效result heapq.heappushpop(data, -1)print(pushpop结果:, result) # -1堆中最小值被弹出# heapreplace: 先pop再push与pushpop相反result heapq.heapreplace(data, 10)print(replace结果:, result)2. nlargest / nsmallest — Top-K 问题scores [42, 67, 13, 89, 55, 90, 24, 71, 88, 33]top3 heapq.nlargest(3, scores) # 获取最大的3个元素print(Top 3:, top3) # [90, 89, 88]bottom3 heapq.nsmallest(3, scores) # 获取最小的3个元素print(Bottom 3:, bottom3) # [13, 24, 33]# nlargest和nsmallest也支持key参数用于复杂对象的比较students [(Alice, 88), (Bob, 95), (Charlie, 72), (David, 91)]top_students heapq.nlargest(2, students, keylambda s: s[1])print(分数最高的两名学生:, top_students) # [(Bob, 95), (David, 91)]3. 模拟最大堆Max-Heapclass MaxHeap:Python的heapq只支持最小堆通过存入负值来模拟最大堆def __init__(self):self._data [] # 实际存储负值的堆def push(self, val):heapq.heappush(self._data, -val) # 存入负值def pop(self):return -heapq.heappop(self._data) # 弹出后取反def peek(self):return -self._data[0] if self._data else Nonedef __len__(self):return len(self._data)max_heap MaxHeap()for v in [3, 1, 4, 1, 5, 9]:max_heap.push(v)print(最大堆弹出:, max_heap.pop()) # 9print(最大堆弹出:, max_heap.pop()) # 54. 自定义比较器的优先队列import dataclassesdataclasses.dataclass(orderTrue)class Task:任务类优先级高的先执行priority: int # 数值越小优先级越高name: str dataclasses.field(compareFalse) # 不参与比较# 直接使用heapq存储Task对象tasks [Task(3, 写报告), Task(1, 修复Bug), Task(2, 开晨会)]heapq.heapify(tasks)while tasks:t heapq.heappop(tasks)print(f执行任务: {t.name} (优先级{t.priority}))# 输出顺序: 修复Bug - 开晨会 - 写报告5. 合并K个有序列表def merge_k_sorted_lists(lists):使用堆合并K个有序列表时间复杂度O(NlogK)heap []# 将每个列表的头元素值和所在列表索引、元素索引加入堆for i, lst in enumerate(lists):if lst:heapq.heappush(heap, (lst[0], i, 0))result []while heap:val, list_idx, elem_idx heapq.heappop(heap)result.append(val)# 如果该列表还有下一个元素将其加入堆if elem_idx 1 len(lists[list_idx]):next_val lists[list_idx][elem_idx 1]heapq.heappush(heap, (next_val, list_idx, elem_idx 1))return resultlists [[1, 4, 7], [2, 5, 8], [3, 6, 9]]print(合并K个有序列表:, merge_k_sorted_lists(lists)) # [1,2,3,4,5,6,7,8,9]6. 中位数查找双堆法class MedianFinder:使用最大堆最小堆维护数据流的中位数def __init__(self):self.small [] # 最大堆存较小的一半用负值模拟self.large [] # 最小堆存较大的一半def add_num(self, num):插入新数字if not self.small or num -self.small[0]:heapq.heappush(self.small, -num)else:heapq.heappush(self.large, num)# 平衡两个堆的大小保证small比large多0或1个元素if len(self.small) len(self.large) 1:heapq.heappush(self.large, -heapq.heappop(self.small))elif len(self.large) len(self.small):heapq.heappush(self.small, -heapq.heappop(self.large))def find_median(self):返回当前中位数if len(self.small) len(self.large):return -self.small[0] # 奇数个small堆顶就是中位数return (-self.small[0] self.large[0]) / 2.0 # 偶数个取平均mf MedianFinder()for v in [5, 2, 8, 1, 9]:mf.add_num(v)print(f添加{v}后中位数: {mf.find_median()})7. 任务调度器模式def least_interval(tasks, n):任务调度器相同任务之间至少间隔n个时间片freq [0] * 26for t in tasks:freq[ord(t) - ord(A)] 1max_freq max(freq) # 出现次数最多的任务的频率max_count freq.count(max_freq) # 有多少种任务出现了max_freq次part_count max_freq - 1 # 间隔段数part_length n 1 # 每段的长度empty_slots part_count * part_length # 总的理论插槽数available len(tasks) - max_count # 可填充的任务数idles max(0, empty_slots - available) # 需要的空闲时间片return len(tasks) idlesprint(任务调度最少时间:, least_interval([A,A,A,B,B,B], 2)) # 8
