堆与优先队列

#数据结构 · #堆 · #优先队列 · #二叉堆 · #斐波那契堆 · #TopK

堆是一种特殊的完全二叉树，满足"父节点 ≥（或 ≤）子节点"的性质。优先队列是基于堆实现的数据结构，广泛应用于调度系统、TopK 问题、Dijkstra 最短路径等场景。

二叉堆

性质

最大堆（父 ≥ 子）：            最小堆（父 ≤ 子）：
        100                        1
       /   \                     /   \
     50     80                  5     3
    /  \   /  \               /  \   /  \
   30  20 60  70             10   8 6   9

关键操作复杂度：

操作	时间复杂度	说明
Insert	O(log n)	插入末尾，上浮
ExtractMax/Min	O(log n)	取根，末尾替根，下沉
Peek	O(1)	返回根节点
BuildHeap	O(n)	从后往前下沉
Increase/Decrease Key	O(log n)	修改值后上浮或下沉

数组表示

        [0]
       /   \
    [1]     [2]
   /   \   /   \
 [3]  [4] [5]  [6]

index i 的：
  父节点: (i-1) / 2
  左子: 2*i + 1
  右子: 2*i + 2

Go 实现（最小堆）

type MinHeap struct {
    data []int
}

func (h *MinHeap) Insert(val int) {
    h.data = append(h.data, val)
    h.siftUp(len(h.data) - 1)
}

func (h *MinHeap) ExtractMin() (int, bool) {
    if len(h.data) == 0 {
        return 0, false
    }
    min := h.data[0]
    last := len(h.data) - 1
    h.data[0] = h.data[last]
    h.data = h.data[:last]
    if len(h.data) > 0 {
        h.siftDown(0)
    }
    return min, true
}

func (h *MinHeap) siftUp(i int) {
    for i > 0 {
        parent := (i - 1) / 2
        if h.data[parent] <= h.data[i] {
            break
        }
        h.data[parent], h.data[i] = h.data[i], h.data[parent]
        i = parent
    }
}

func (h *MinHeap) siftDown(i int) {
    n := len(h.data)
    for {
        smallest := i
        left := 2*i + 1
        right := 2*i + 2
        if left < n && h.data[left] < h.data[smallest] {
            smallest = left
        }
        if right < n && h.data[right] < h.data[smallest] {
            smallest = right
        }
        if smallest == i {
            break
        }
        h.data[i], h.data[smallest] = h.data[smallest], h.data[i]
        i = smallest
    }
}

Go 标准库用法

import "container/heap"

type Item struct {
    value    string
    priority int
    index    int  // heap.Interface 要求的
}

type PriorityQueue []*Item

func (pq PriorityQueue) Len() int           { return len(pq) }
func (pq PriorityQueue) Less(i, j int) bool { return pq[i].priority < pq[j].priority }
func (pq PriorityQueue) Swap(i, j int)      { pq[i], pq[j] = pq[j], pq[i]; pq[i].index = i; pq[j].index = j }
func (pq *PriorityQueue) Push(x any)        { *pq = append(*pq, x.(*Item)) }
func (pq *PriorityQueue) Pop() any          { old := *pq; n := len(old); x := old[n-1]; *pq = old[:n-1]; return x }

// 使用
pq := &PriorityQueue{}
heap.Init(pq)
heap.Push(pq, &Item{value: "task1", priority: 3})
heap.Push(pq, &Item{value: "task2", priority: 1})
top := heap.Pop(pq).(*Item)  // task2 (priority=1)

堆排序

堆排序在 pdqsort（Go 1.22+ 排序算法）中作为快速排序退化时的回退策略。完整排序算法对比与各语言实现见排序与选择算法。

func HeapSort(arr []int) {
    n := len(arr)
    // 1. 建堆 O(n)
    for i := n/2 - 1; i >= 0; i-- {
        heapify(arr, n, i)
    }
    // 2. 逐个取堆顶 O(n log n)
    for i := n - 1; i > 0; i-- {
        arr[0], arr[i] = arr[i], arr[0]
        heapify(arr, i, 0)
    }
}

func heapify(arr []int, n, i int) {
    largest := i
    left, right := 2*i+1, 2*i+2
    if left < n && arr[left] > arr[largest] {
        largest = left
    }
    if right < n && arr[right] > arr[largest] {
        largest = right
    }
    if largest != i {
        arr[i], arr[largest] = arr[largest], arr[i]
        heapify(arr, n, largest)
    }
}

特点：

原地排序，O(1) 额外空间
不稳定排序
适合需要部分排序的场景（TopK）

TopK 问题

方案对比

方案	时间复杂度	适用
全排序	O(n log n)	K ≈ n
大小为K的最小堆	O(n log K)	K << n ✅
快速选择	平均 O(n)	不需要有序结果
桶排序/计数	O(n)	数据范围有限

// 用大小为 K 的最小堆求 TopK 大
func TopK(nums []int, k int) []int {
    if k <= 0 || len(nums) == 0 {
        return nil
    }
    h := &MinHeap{data: make([]int, 0, k)}
    for _, num := range nums {
        if len(h.data) < k {
            h.Insert(num)
        } else if num > h.data[0] {
            h.data[0] = num
            h.siftDown(0)
        }
    }
    return h.data
}

斐波那契堆

与二叉堆对比

操作	二叉堆	斐波那契堆
Insert	O(log n)	O(1)
ExtractMin	O(log n)	O(log n) 均摊
DecreaseKey	O(log n)	O(1) 均摊
Merge	O(n)	O(1)
实现难度	简单	复杂

适用场景：Dijkstra 算法中大量 DecreaseKey 操作时，斐波那契堆更优。

核心思想

二叉堆：单一树结构
斐波那契堆：森林 + 惰性合并

- Insert: 直接加到根链表,O(1)
- ExtractMin: 才进行合并整理,O(log n) 均摊
- DecreaseKey: 切下来挂到根链表,O(1) 均摊

各系统中的应用

系统	使用场景	结构
Go timer	定时器管理	四叉堆（runtime/time.go）
Linux CFS	进程调度选择最小 vruntime	红黑树（非堆但思想类似）
Java PriorityQueue	标准库	二叉堆（数组实现）
Python heapq	标准库	最小二叉堆
Redis 有序集合	ZADD/ZPOPMIN	跳表（类似堆的功能）
Dijkstra / Prim	最短路径 / 最小生成树	优先队列（二叉堆或斐波那契堆）
Huffman 编码	构建最优前缀树	最小堆

其他堆变体

变体	特点
d-ary heap	d 叉堆，降低树高，减少 siftDown 次数
Pairing Heap	实现简单，实际性能接近斐波那契堆
Leftist Heap	左倾堆，支持高效 Merge
Binomial Heap	二项堆，Merge O(log n)
Skew Heap	斜堆，自调整堆

实战选型指南

不是所有场景都要用最复杂的堆。实际工程中，选择标准是"够用就好"：

场景	推荐堆类型	原因
通用优先队列	二叉堆	实现简单，性能足够
Dijkstra 最短路径	斐波那契堆 / Pairing Heap	大量 DecreaseKey，需要 O(1)
合并优先队列	Leftist Heap / Binomial Heap	Merge 为高频操作
Go runtime timer	四叉堆（d=4）	降低树高减少 siftDown 次数
Python heapq	二叉堆（数组实现）	极简实现，cache 友好

Go container/heap 实战

Go 标准库不提供现成的堆，而是提供 container/heap 接口——你需要自己的类型实现 5 个方法：

import "container/heap"

// 1. 定义类型
type Item struct {
    Value    string
    Priority int    // 优先级：越小越靠前
    index    int    // 堆中位置（heap 内部维护）
}

// 2. 实现 heap.Interface（5 个方法）
type PriorityQueue []*Item

func (pq PriorityQueue) Len() int           { return len(pq) }
func (pq PriorityQueue) Less(i, j int) bool { return pq[i].Priority < pq[j].Priority }
func (pq PriorityQueue) Swap(i, j int) {
    pq[i], pq[j] = pq[j], pq[i]
    pq[i].index = i
    pq[j].index = j
}
func (pq *PriorityQueue) Push(x any) {
    n := len(*pq)
    item := x.(*Item)
    item.index = n
    *pq = append(*pq, item)
}
func (pq *PriorityQueue) Pop() any {
    old := *pq
    n := len(old)
    item := old[n-1]
    old[n-1] = nil  // 避免内存泄漏
    item.index = -1
    *pq = old[0 : n-1]
    return item
}

// 3. 使用
func main() {
    pq := make(PriorityQueue, 0)
    heap.Init(&pq)

    heap.Push(&pq, &Item{Value: "task1", Priority: 3})
    heap.Push(&pq, &Item{Value: "task2", Priority: 1})
    heap.Push(&pq, &Item{Value: "task3", Priority: 2})

    for pq.Len() > 0 {
        item := heap.Pop(&pq).(*Item)
        fmt.Printf("处理: %s (优先级 %d)\n", item.Value, item.Priority)
        // 输出: task2(1) → task3(2) → task1(3)
    }

    // Update 操作：修改优先级后调用 Fix
    task2.Priority = 0
    heap.Fix(&pq, task2.index)  // 将 task2 提升到堆顶
}

TopK 问题 — 四解法对比

这是面试最高频的堆应用场景。从 1 亿个数中找最大的 K 个，不同方案的抉择：

解法	时间复杂度	空间复杂度	K 很小时	数据流式	实际推荐度
全排序	O(n log n)	O(n)	❌	❌	⭐ 除非需要全局有序
最小堆 (K 个)	O(n log K)	O(K)	✅✅	✅✅	⭐⭐⭐ 最推荐
快速选择	O(n) 平均	O(n)	✅	❌	⭐⭐ Top-K 需要有序时
桶排序/计数	O(n)	O(range)	❌	❌	⭐ 仅整数+范围已知

// TopK 最小堆实现 — 从海量数据流中实时获取最大的 K 个
func topKMax(stream <-chan int, k int) []int {
    // 维护一个大小为 K 的最小堆（堆顶是第 K 大的元素）
    h := &IntMinHeap{}
    heap.Init(h)

    for val := range stream {
        if h.Len() < k {
            heap.Push(h, val)
        } else if val > (*h)[0] {
            // 新元素比第 K 大更大 → 替换堆顶
            (*h)[0] = val
            heap.Fix(h, 0)
        }
        // 否则跳过（比第 K 大还小，不可能进 TopK）
    }

    result := make([]int, h.Len())
    for i := len(result) - 1; i >= 0; i-- {
        result[i] = heap.Pop(h).(int)
    }
    return result  // 从小到大有序
}

工程实践：为什么真实系统会选不同的堆

1. 同样是 O(log n)，堆常常比树更 cache 友好

二叉堆最大的工程优势之一，不是复杂度公式，而是连续数组存储。

结构	内存布局	局部性
二叉堆	连续数组	✅ 好
红黑树/AVL	分散节点 + 指针	❌ 一般偏差
跳表	多层指针	🟡 中等

这意味着在很多内存内场景里：

堆的比较次数不一定更少
但 cache miss 往往更少
CPU 预取更容易生效
常数因子通常更好

所以像定时器、任务调度、TopK 这类“总是取最值”的问题，堆很常见。

2. 为什么很多系统不直接用“理论更强”的斐波那契堆

斐波那契堆的 DecreaseKey 和 Merge 均摊复杂度很好看，但工程上它并不常见，原因很现实：

结构复杂
常数大
指针很多，局部性差
实现、调试、维护成本高

所以真实系统里常见的选择通常是：

二叉堆 / d 叉堆：实现简单、局部性好
Pairing Heap：在需要频繁 DecreaseKey 时作为折中
索引堆 / 懒删除堆：规避标准堆不擅长任意更新的问题

3. `DecreaseKey` 为什么是工程分水岭

普通二叉堆很适合：

Push
Pop
Peek

但如果你需要“把堆中某个任意元素的优先级改掉”，事情就复杂了。因为你必须先知道它在数组里的下标。

这就是为什么工程中常常引入：

index 字段
外部 map[key]index
heap.Fix
或者直接采用“懒删除”策略

例如 Dijkstra、任务调度器、延迟队列里，真正难的往往不是 pop min，而是：任务优先级变化后怎么高效更新。

4. Go runtime 为什么用四叉堆而不是二叉堆

二叉堆并不是唯一答案。像 Go timer 选择四叉堆，本质上是一个工程折中。

d 叉堆的数学属性：

树高: log_d(n)                               ← d 越大，树越矮
每层比较次数: d（找最小/最大孩子）              ← d 越大，单层比较越多

siftDown 的总代价 = 层数 × 每层比较次数
                  = log_d(n) × d
                  = (log₂(n) / log₂(d)) × d

以 n=100 万为例：

d (叉数)	层数 log_d(n)	每层比较	总比较	cache miss (估算)
2 (二叉)	20	2	40	~10 次跨层访问
4 (四叉)	10	4	40	~5 次跨层访问
8 (八叉)	7	8	56	~3 次跨层访问

为什么四叉堆不是八叉堆——理论比较次数相近，实际性能取决于 cache 行为：

每一层 siftDown = 访问一个新的数组位置
同一层的 d 个孩子 → 通常在同一 cache line (64B) 内，访问几乎零延迟
下一层 → 新的数组位置 → 可能 cache miss (~100 CPU cycles)

二叉堆: 20 层, 每层仅 2 次比较 + 可能 cache miss
四叉堆: 10 层, 每层 4 次比较（全在 cache line 内）, 可能 cache miss 减半
八叉堆: 7 层, 每层 8 次比较 → 寄存器比较多了，但跨层访问只少了 3 次 → 边际收益递减
                                      且 8 个孩子可能跨 cache line → 得不偿失

实测结论: 四叉堆比二叉堆快 15-30%，比八叉堆也略快（八叉堆比较太多）

这正是"复杂度相同，但常数和局部性不同"的典型例子。Go 1.14+ 进一步将全局定时器堆拆分为每个 P 独立的四叉堆，消除了锁竞争，使 timer 精度从毫秒级提升到微秒级。

5. 典型故障：堆不是根因，但常是排队与抖动的放大器

mermaid

flowchart LR
    A["任务持续进入"] --> B["优先队列/定时器堆变大"]
    B --> C["每次 push/pop/fix 成本上涨"]
    C --> D["下游处理更慢"]
    D --> E["排队继续加深"]
    E --> F["RT 抖动 / 超时 / 内存上涨"]

例如：

延迟队列堆积过深
定时器太多
scheduler 中优先级更新过于频繁
TopK 窗口太大导致维护成本上升

这些问题常不是“堆写错了”，而是系统负载模型已经不适合当前堆规模和更新方式。

6. 选型判断

场景	更推荐
只要快速取最值	二叉堆 / d 叉堆
频繁 `DecreaseKey`	索引堆 / Pairing Heap / 特化结构
需要有序遍历范围	平衡树 / 跳表
TopK, 流式处理	固定大小最小堆
定时器系统	d 叉堆或红黑树，视更新模式而定

一个很实用的原则：

text

如果你的核心操作是“取最值”，优先想到堆；
如果你的核心操作是“按区间/顺序遍历”，优先想到树或跳表；
如果你的核心操作是“随机查找”，优先想到哈希。

7. 排障时怎么看是不是堆模型不合适

现象	可能问题
RT 随队列深度显著恶化	堆太大或更新太频繁
CPU 热在 `Fix` / `siftDown`	优先级更新模式不合理
内存上涨	堆中对象堆积，消费跟不上
定时器很多且抖动明显	timer 堆规模过大、超时模型过细

合并 K 个有序链表 (LeetCode 23)

合并 K 个升序链表，返回一个升序链表。本题展示了堆和分治两种解法的优劣。

type ListNode struct {
    Val  int
    Next *ListNode
}

// 解法 1: 最小堆 — 时间 O(N log K), 空间 O(K)
func mergeKLists(lists []*ListNode) *ListNode {
    h := &MinHeap{}
    heap.Init(h)

    // 每个链表的头节点入堆
    for _, head := range lists {
        if head != nil {
            heap.Push(h, head)
        }
    }

    dummy := &ListNode{}
    cur := dummy
    for h.Len() > 0 {
        // 取最小值
        node := heap.Pop(h).(*ListNode)
        cur.Next = node
        cur = cur.Next
        // 该链表下一个节点入堆
        if node.Next != nil {
            heap.Push(h, node.Next)
        }
    }
    return dummy.Next
}

type MinHeap []*ListNode

func (h MinHeap) Len() int           { return len(h) }
func (h MinHeap) Less(i, j int) bool  { return h[i].Val < h[j].Val }
func (h MinHeap) Swap(i, j int)       { h[i], h[j] = h[j], h[i] }
func (h *MinHeap) Push(x any)         { *h = append(*h, x.(*ListNode)) }
func (h *MinHeap) Pop() any {
    old := *h
    n := len(old)
    x := old[n-1]
    *h = old[:n-1]
    return x
}

// 解法 2: 分治两两合并 — 时间 O(N log K), 空间 O(log K) 递归栈
func mergeKListsDivideAndConquer(lists []*ListNode) *ListNode {
    if len(lists) == 0 { return nil }
    if len(lists) == 1 { return lists[0] }

    mid := len(lists) / 2
    left  := mergeKListsDivideAndConquer(lists[:mid])
    right := mergeKListsDivideAndConquer(lists[mid:])
    return mergeTwoLists(left, right)
}

// 合并两个有序链表
func mergeTwoLists(l1, l2 *ListNode) *ListNode {
    dummy := &ListNode{}
    cur := dummy
    for l1 != nil && l2 != nil {
        if l1.Val < l2.Val {
            cur.Next = l1
            l1 = l1.Next
        } else {
            cur.Next = l2
            l2 = l2.Next
        }
        cur = cur.Next
    }
    if l1 != nil { cur.Next = l1 }
    if l2 != nil { cur.Next = l2 }
    return dummy.Next
}

解法	时间	空间	适用场景
最小堆	O(N log K)	O(K)	K 很大时堆内存可控，流式输入
分治合并	O(N log K)	O(log K) 栈	K 较小时常数更优，无需堆开销
逐一两两合并	O(N K)	O(1)	K=2 时最简单

面试建议：先写出 mergeTwoLists，然后给出两种 K 路合并方案的复杂度分析，最后根据场景选择实现（分治通常代码更短，堆实现更直观）。

Top-K 问题：三种解法对比

Top-K 是堆的高频应用，但堆不是唯一解法。理解三种方案的原理和适用场景是面试加分点。

方案 1：堆 — 通用最优

// 找到数组中第 K 大的元素 — 小顶堆 O(N log K)
func findKthLargest(nums []int, k int) int {
    h := &IntMinHeap{}
    heap.Init(h)
    for _, num := range nums {
        heap.Push(h, num)
        if h.Len() > k {
            heap.Pop(h) // 堆大小始终为 K，pop 掉最小值
        }
    }
    return heap.Pop(h).(int) // 堆顶就是第 K 大
}

方案 2：快速选择（QuickSelect）— 期望 O(N)

func findKthLargestQuickSelect(nums []int, k int) int {
    target := len(nums) - k // 第 K 大 = 排序后第 N-K 个
    l, r := 0, len(nums)-1
    for l < r {
        pivot := partition(nums, l, r)
        if pivot == target {
            return nums[pivot]
        } else if pivot < target {
            l = pivot + 1
        } else {
            r = pivot - 1
        }
    }
    return nums[l]
}

func partition(nums []int, l, r int) int {
    pivot := nums[r]
    i := l
    for j := l; j < r; j++ {
        if nums[j] < pivot {
            nums[i], nums[j] = nums[j], nums[i]
            i++
        }
    }
    nums[i], nums[r] = nums[r], nums[i]
    return i
}

方案 3：桶排序 — O(N) 但仅限整数小范围

func topKFrequentBucket(nums []int, k int) []int {
    freq := make(map[int]int)
    for _, n := range nums { freq[n]++ }

    // 桶: buckets[i] = 出现 i 次的所有数字
    buckets := make([][]int, len(nums)+1)
    for num, cnt := range freq {
        buckets[cnt] = append(buckets[cnt], num)
    }

    var res []int
    for i := len(buckets) - 1; i >= 0 && len(res) < k; i-- {
        res = append(res, buckets[i]...)
    }
    return res[:k]
}

方案对比

方案	时间复杂度	空间	适用场景
堆	O(N log K)	O(K)	通用最优，K 远小于 N 时最强
快速选择	O(N) 平均, O(N²) 最坏	O(1)	只需第 K 个值，不需要 Top K 列表
桶排序	O(N)	O(N)	数据范围小（如频率、年龄），需返回列表
全排序	O(N log N)	O(1)	N 小或 K 接近 N 时最简单

mermaid

flowchart TB
    A["Top-K 问题"] --> B{"K 远小于 N?"}
    B -->|"是"| C["堆 O(N log K)<br/>内存可控"]
    B -->|"否 (K≈N/2)"| D{"数据范围小?"}
    D -->|"是"| E["桶排序 O(N)"]
    D -->|"否"| F["快速选择 O(N)"]
    F --> G{"最坏 O(N²) 可接受?"}
    G -->|"是"| F
    G -->|"否"| C

面试核心：能讲出三种方案、各自的复杂度、和适用场景差异，比写出代码更重要。

堆与优先队列 ​

二叉堆 ​

性质 ​

数组表示 ​

Go 实现（最小堆） ​

Go 标准库用法 ​

堆排序 ​

TopK 问题 ​

方案对比 ​

斐波那契堆 ​

与二叉堆对比 ​

核心思想 ​

各系统中的应用 ​

其他堆变体 ​

实战选型指南 ​

Go container/heap 实战 ​

TopK 问题 — 四解法对比 ​

工程实践：为什么真实系统会选不同的堆 ​

1. 同样是 O(log n)，堆常常比树更 cache 友好 ​

2. 为什么很多系统不直接用“理论更强”的斐波那契堆 ​

3. DecreaseKey 为什么是工程分水岭 ​

4. Go runtime 为什么用四叉堆而不是二叉堆 ​

5. 典型故障：堆不是根因，但常是排队与抖动的放大器 ​

6. 选型判断 ​

7. 排障时怎么看是不是堆模型不合适 ​

合并 K 个有序链表 (LeetCode 23) ​

Top-K 问题：三种解法对比 ​

方案 1：堆 — 通用最优 ​

方案 2：快速选择（QuickSelect）— 期望 O(N) ​

方案 3：桶排序 — O(N) 但仅限整数小范围 ​

方案对比 ​

💬 文章评论

堆与优先队列

二叉堆

性质

数组表示

Go 实现（最小堆）

Go 标准库用法

堆排序

TopK 问题

方案对比

斐波那契堆

与二叉堆对比

核心思想

各系统中的应用

其他堆变体

实战选型指南

Go container/heap 实战

TopK 问题 — 四解法对比

工程实践：为什么真实系统会选不同的堆

1. 同样是 O(log n)，堆常常比树更 cache 友好

2. 为什么很多系统不直接用“理论更强”的斐波那契堆

3. `DecreaseKey` 为什么是工程分水岭

4. Go runtime 为什么用四叉堆而不是二叉堆

5. 典型故障：堆不是根因，但常是排队与抖动的放大器

6. 选型判断

7. 排障时怎么看是不是堆模型不合适

合并 K 个有序链表 (LeetCode 23)

Top-K 问题：三种解法对比

方案 1：堆 — 通用最优

方案 2：快速选择（QuickSelect）— 期望 O(N)

方案 3：桶排序 — O(N) 但仅限整数小范围

方案对比