什么是优先队列？

优先队列（Priority Queue）是一种特殊的队列，它不再按照"先入先出"的顺序处理元素，而是根据元素的优先级决定处理顺序。优先级最高的元素总是最先被取出。

优先队列的核心操作包括：

  • 插入（Insert）：添加一个新元素，并根据优先级放入合适位置。
  • 提取最高优先级元素（Extract-Min/Max）：移除并返回优先级最高的元素。

常见实现方式

方式一：普通数组/链表

最简单的实现是使用数组或链表存储元素，每次提取时遍历整个结构找到优先级最高的元素。

  • 插入：O(1)
  • 提取：O(n)

这种方式在元素较多时效率很低，不适用于大规模数据。

方式二：二叉堆（Binary Heap）

二叉堆是优先队列最常用的高效实现方式。它是一种完全二叉树，同时满足堆属性：

  • 最小堆：每个父节点的值小于或等于其子节点的值，根节点是最小值。
  • 最大堆：每个父节点的值大于或等于其子节点的值，根节点是最大值。

通常优先队列使用最小堆实现。

二叉堆的核心操作

上浮（Sift-Up）

插入元素时，先将其放到堆的末尾，然后与父节点比较，若优先级更高则交换，重复直到满足堆属性。

function sift_up(heap, index):
    while index > 0:
        parent = (index - 1) // 2
        if heap[index] >= heap[parent]:
            break
        swap(heap[index], heap[parent])
        index = parent

下沉（Sift-Down）

提取根节点后，将最后一个元素放到根节点位置，然后与子节点比较，与优先级更高的子节点交换，重复直到满足堆属性。

function sift_down(heap, index):
    n = len(heap)
    while True:
        left = 2 * index + 1
        right = 2 * index + 2
        smallest = index
        
        if left < n and heap[left] < heap[smallest]:
            smallest = left
        if right < n and heap[right] < heap[smallest]:
            smallest = right
        if smallest == index:
            break
        swap(heap[index], heap[smallest])
        index = smallest

复杂度分析

  • 插入：O(log n) —— 最多需要从叶节点上浮到根节点，路径长度为 log n
  • 提取最小元素：O(log n) —— 最多需要从根节点下沉到叶节点
  • 查找最小元素：O(1) —— 直接访问根节点

Python 示例：使用 heapq

Python 的标准库 heapq 模块提供了最小堆的实现。

import heapq

# 创建一个空堆
heap = []

# 插入元素
heapq.heappush(heap, 5)
heapq.heappush(heap, 3)
heapq.heappush(heap, 8)
heapq.heappush(heap, 1)

print(heap)  # [1, 3, 8, 5] —— 注意：不是完全有序，只保证根节点最小

# 提取最小元素
print(heapq.heappop(heap))  # 1
print(heapq.heappop(heap))  # 3
print(heap)  # [5, 8]

# 堆化一个列表
arr = [4, 1, 7, 3, 2]
heapq.heapify(arr)
print(arr)  # [1, 2, 7, 3, 4]

实现最大堆

heapq 默认实现的是最小堆，要实现最大堆可以将元素取负值后插入：

import heapq

max_heap = []
heapq.heappush(max_heap, -5)
heapq.heappush(max_heap, -3)
heapq.heappush(max_heap, -8)

print(-heapq.heappop(max_heap))  # 8（最大元素）

优先队列的应用：Dijkstra 算法

优先队列最经典的应用之一是 Dijkstra 最短路径算法。该算法在带权图中寻找从起点到所有其他节点的最短路径。

import heapq
from typing import Dict, List, Tuple


def dijkstra(graph: Dict[str, List[Tuple[str, int]]], start: str) -> Dict[str, int]:
    distances = {node: float('inf') for node in graph}
    distances[start] = 0
    
    # 优先队列：(距离, 节点)
    pq = [(0, start)]
    
    while pq:
        current_dist, current_node = heapq.heappop(pq)
        
        # 跳过已经处理过的节点
        if current_dist > distances[current_node]:
            continue
        
        for neighbor, weight in graph[current_node]:
            distance = current_dist + weight
            
            # 如果找到更短的路径
            if distance < distances[neighbor]:
                distances[neighbor] = distance
                heapq.heappush(pq, (distance, neighbor))
    
    return distances


graph = {
    "A": [("B", 4), ("C", 2)],
    "B": [("A", 4), ("C", 1), ("D", 5)],
    "C": [("A", 2), ("B", 1), ("D", 8)],
    "D": [("B", 5), ("C", 8)],
}

distances = dijkstra(graph, "A")
print(distances)  # {'A': 0, 'B': 3, 'C': 2, 'D': 8}

适用场景

优先队列在以下场景中非常有用：

  • 任务调度：根据优先级处理任务
  • 图算法：Dijkstra、Prim 最小生成树
  • 数据压缩：Huffman 编码
  • 事件驱动模拟：按时间顺序处理事件

选择合适的实现方式取决于具体的使用场景和性能需求。