优先队列=堆+业务逻辑，是算法设计中最优雅的抽象之一
优先队列（Priority Queue）是堆的抽象接口。普通队列 FIFO，优先队列按优先级出队。优先队列的本质是：在动态集合中，始终能高效获取最值。它在算法设计中无处不在——从操作系统进程调度到网络路由，从搜索引擎排序到数据流处理。
| 场景 | 优先级定义 | 经典题目 | 时间优化 |
|---|---|---|---|
| 多路归并 | 当前元素值 | 合并K个排序链表 | O(N log K) |
| 任务调度 | 截止时间/优先级 | 任务调度器 | O(N log N) |
| 最短路径 | 当前距离 | Dijkstra算法 | O((V+E) log V) |
| 数据流 | 元素值/频率 | 中位数/TopK | O(log N) 每次 |
| 哈夫曼编码 | 字符频率 | 最优前缀编码 | O(N log N) |
| A*搜索 | f=g+h估价 | 迷宫最短路 | 依赖启发函数 |
# 方式一: heapq (推荐，最快，线程不安全)
import heapq
heap = []
heapq.heappush(heap, (priority, item))
item = heapq.heappop(heap)
# 方式二: queue.PriorityQueue (线程安全，稍慢)
from queue import PriorityQueue
pq = PriorityQueue()
pq.put((priority, item))
item = pq.get()
# 方式三: 自定义比较 (需要__lt__方法)
class Node:
def __init__(self, val, cost):
self.val = val
self.cost = cost
def __lt__(self, other):
return self.cost < other.cost
# 性能对比:
# heapq: 最快, 适合单线程算法题
# PriorityQueue: 线程安全, 适合多线程场景
# 自定义类: 适合复杂比较逻辑优先队列在图论中最著名的应用是Dijkstra最短路算法。每一步取出当前距离最小的未访问节点，松弛其邻居，更新距离后重新入堆。
def dijkstra(graph, start):
n = len(graph)
dist = [float('inf')] * n
dist[start] = 0
heap = [(0, start)] # (距离, 节点)
while heap:
d, u = heapq.heappop(heap)
if d > dist[u]: # 延迟删除：过时记录跳过
continue
for v, w in graph[u]:
if dist[u] + w < dist[v]:
dist[v] = dist[u] + w
heapq.heappush(heap, (dist[v], v))
return dist给定一个链表数组，每个链表都已按升序排列，将所有链表合并到一个升序链表中，返回合并后的链表。
# Definition for singly-linked list.
# class ListNode:
# def __init__(self, val=0, next=None):
# self.val = val
# self.next = next
class Solution:
def mergeKLists(self, lists: List[Optional[ListNode]]) -> Optional[ListNode]:
dummy = ListNode(0)
curr = dummy
heap = []
# 将各链表头节点入堆
for i, node in enumerate(lists):
if node:
# i作为第二关键字避免ListNode比较
heapq.heappush(heap, (node.val, i, node))
while heap:
val, i, node = heapq.heappop(heap)
curr.next = node
curr = curr.next
if node.next:
heapq.heappush(heap, (node.next.val, i, node.next))
return dummy.next
# 分治合并版本（空间更优 O(log K)）
class Solution:
def mergeKLists(self, lists: List[Optional[ListNode]]) -> Optional[ListNode]:
if not lists:
return None
def merge2(a, b):
dummy = ListNode(0)
curr = dummy
while a and b:
if a.val <= b.val:
curr.next = a
a = a.next
else:
curr.next = b
b = b.next
curr = curr.next
curr.next = a or b
return dummy.next
# 分治：每次合并相邻两个链表
step = 1
while step < len(lists):
for i in range(0, len(lists) - step, step * 2):
lists[i] = merge2(lists[i], lists[i + step])
step *= 2
return lists[0]| 方法 | 时间 | 空间 | 特点 |
|---|---|---|---|
| 优先队列 | O(N log K) | O(K) | 推荐 |
| 两两合并 | O(NK) | O(1) | 简单但慢 |
| 分治合并 | O(N log K) | O(log K) | 空间更优 |
(node.val, i, node)。这是Python特有的坑！设计一个类，初始时传入 k 和 nums，每次调用 add(val) 添加元素并返回当前第k大的元素。
class KthLargest:
def __init__(self, k: int, nums: List[int]):
self.k = k
self.heap = []
for n in nums:
heapq.heappush(self.heap, n)
if len(self.heap) > self.k:
heapq.heappop(self.heap)
def add(self, val: int) -> int:
heapq.heappush(self.heap, val)
if len(self.heap) > self.k:
heapq.heappop(self.heap)
return self.heap[0]
# 使用示例
k = KthLargest(3, [4, 5, 8, 2])
print(k.add(3)) # 4
print(k.add(5)) # 5
print(k.add(10)) # 5
print(k.add(9)) # 8
print(k.add(4)) # 8初始化: O(n log k) add: O(log k) 空间: O(k)
给你一个用字符数组表示的CPU任务列表和一个冷却时间n。每个相同任务之间必须有n个冷却时间。计算完成所有任务的最短时间。
class Solution:
def leastInterval(self, tasks: List[str], n: int) -> int:
# 统计频率
freq = list(collections.Counter(tasks).values())
max_freq = max(freq)
# 出现最大频率的任务数
max_count = freq.count(max_freq)
# 公式: (max_freq-1)*(n+1) + max_count
# 但不能少于任务总数
return max(len(tasks), (max_freq - 1) * (n + 1) + max_count)