从零实现建堆、推入与弹出
为推入实现向上调整，为弹出实现向下调整，然后使用 Floyd 算法以 O(n) 的时间从无序数组构建堆。
从零实现建堆、推入与弹出 是 CoddyKit 上的免费 Coding Interview Prep 课时。 这是第 2 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 Coding Interview Prep 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 Coding Interview Prep 课程共包含 4 节课。
构建 MinHeap 类
从零实现堆有助于掌握底层机制，偶尔也会出现在高级职位的面试中。MinHeap 类封装一个数组，并提供 push、pop、peek 和 size 操作。它在内部通过 push 后调用上滤、pop 后调用下滤来维护堆性质。理解这一实现后，Python 的 heapq 模块就会变得完全透明。
class MinHeap:
def __init__(self):
self._data = []
def push(self, val):
self._data.append(val)
self._sift_up(len(self._data) - 1)
def pop(self):
if len(self._data) == 1:
return self._data.pop()
min_val = self._data[0]
self._data[0] = self._data.pop() # move last to root
self._sift_down(0)
return min_val
def peek(self):
return self._data[0] if self._data else None
def size(self):
return len(self._data)
def _parent(self, i): return (i - 1) // 2
def _left(self, i): return 2 * i + 1
def _right(self, i): return 2 * i + 2
print('MinHeap class skeleton defined')实现上滤
上滤会将节点与其父节点进行比较，只要堆性质（最小堆中父节点 <= 子节点）被违反，就不断向上交换。关键在于，新插入的元素位于末尾，并会向上移动到正确位置。while 循环最多运行 floor(log n) 次，即树的高度。每一步都将 i = parent 赋给 i，以继续向上移动。
class MinHeap:
def __init__(self):
self._data = []
def _parent(self, i): return (i - 1) // 2
def _left(self, i): return 2 * i + 1
def _right(self, i): return 2 * i + 2
def _sift_up(self, i):
while i > 0:
p = self._parent(i)
if self._data[p] > self._data[i]: # parent > child: swap
self._data[p], self._data[i] = self._data[i], self._data[p]
i = p
else:
break # heap property satisfied
def push(self, val):
self._data.append(val)
self._sift_up(len(self._data) - 1)
h = MinHeap()
for v in [5, 3, 8, 1, 4]:
h.push(v)
print(h._data) # valid min-heap实现下滤
下滤会反复将节点与其最小的子节点交换，使节点向下移动（对于最小堆），直到两个子节点都不再更小，或节点到达叶子。始终比较两个子节点，并与较小的那个交换，以维护堆性质。请记得在比较值之前，检查子节点索引是否在有效范围内。
def _sift_down(data, i):
n = len(data)
while True:
smallest = i
l = 2 * i + 1
r = 2 * i + 2
if l < n and data[l] < data[smallest]:
smallest = l
if r < n and data[r] < data[smallest]:
smallest = r
if smallest == i:
break # already the smallest among i, l, r
data[i], data[smallest] = data[smallest], data[i]
i = smallest
# Test: put a large value at root and sift down
heap = [10, 1, 2, 3, 4, 5, 6]
print('Before sift-down:', heap)
_sift_down(heap, 0)
print('After sift-down:', heap) # 1 should reach top, 10 sink完善 MinHeap 的弹出操作
pop 操作会移除并返回根节点（对于最小堆来说是最小值）。为了保持完全二叉树的形状，将最后一个元素移到根节点位置，然后对它执行下滤。这样可以避免在数组中产生空缺，并保持表示有效。边界情况：如果只剩一个元素，直接弹出并返回它，不执行下滤。
class MinHeap:
def __init__(self):
self._data = []
def push(self, val):
self._data.append(val)
i = len(self._data) - 1
while i > 0:
p = (i - 1) // 2
if self._data[p] > self._data[i]:
self._data[p], self._data[i] = self._data[i], self._data[p]
i = p
else: break
def pop(self):
if not self._data: return None
if len(self._data) == 1: return self._data.pop()
result = self._data[0]
self._data[0] = self._data.pop() # last -> root
i, n = 0, len(self._data)
while True:
s, l, r = i, 2*i+1, 2*i+2
if l < n and self._data[l] < self._data[s]: s = l
if r < n and self._data[r] < self._data[s]: s = r
if s == i: break
self._data[i], self._data[s] = self._data[s], self._data[i]
i = s
return result
h = MinHeap()
for v in [5, 3, 8, 1, 4, 2]: h.push(v)
print([h.pop() for _ in range(6)]) # [1,2,3,4,5,8] sorted弗洛伊德的堆化算法
弗洛伊德算法通过对每个非叶子节点调用下滤，在 O(n) 时间内从无序数组构建最小堆：从最后一个内部节点（n//2 - 1）开始，逐步向根节点移动。叶子节点已经天然是有效的单元素堆。O(n) 的时间界来自以下事实：大多数节点靠近树底，只需要向下调整很短的距离。
def heapify(arr):
n = len(arr)
# Start from last non-leaf: index n//2 - 1
# Work backward to root (index 0)
for i in range(n // 2 - 1, -1, -1):
# Sift down node at index i
j = i
while True:
s = j
l, r = 2*j+1, 2*j+2
if l < n and arr[l] < arr[s]: s = l
if r < n and arr[r] < arr[s]: s = r
if s == j: break
arr[j], arr[s] = arr[s], arr[j]
j = s
return arr
arr = [9, 7, 5, 3, 1, 8, 2, 4, 6]
print('Before:', arr)
heapify(arr)
print('After (min-heap):', arr) # arr[0] should be 1为什么弗洛伊德算法是 O(n)
O(n) 的证明如下：树中高度为 k 的节点有 n/2^(k+1) 个。高度为 k 的每个节点在下滤过程中最多进行 k 次交换。总工作量 = 对所有高度 k 求和：n/2^(k+1) * k。这个几何级数收敛到 O(n)。相比之下，逐个插入的朴素方法中，每次 push 的复杂度为 O(log n)，因此 n 次 push 的总成本为 O(n log n)。对于批量构建，弗洛伊德算法严格更优。
import time
import random
# Compare: O(n) heapify vs O(n log n) one-by-one
n = 100000
data = list(range(n, 0, -1)) # reverse sorted = worst case for push
# Method 1: Floyd's O(n)
data1 = data[:]
start = time.time()
for i in range(n // 2 - 1, -1, -1):
j = i
while True:
s = j; l, r = 2*j+1, 2*j+2
if l < n and data1[l] < data1[s]: s = l
if r < n and data1[r] < data1[s]: s = r
if s == j: break
data1[j], data1[s] = data1[s], data1[j]; j = s
print(f'Floyd heapify: {time.time()-start:.4f}s')
# Method 2: One-by-one insertion
import heapq
start = time.time()
heap = []
for x in data: heapq.heappush(heap, x)
print(f'Push one-by-one: {time.time()-start:.4f}s')向现有集合中推入堆元素
Python 的 heapq.heappushpop 和 heapq.heapreplace 是高效的组合操作。heappushpop(heap, item) 会推入新项目并立即弹出最小项目，比两次分别调用更高效。heapreplace(heap, item) 会在一次操作中弹出最小项目并推入新项目（为保证正确性，新项目必须 >= 旧的最小值）。这些操作适用于前 k 个元素的流式算法。
import heapq
heap = [1, 3, 5, 7, 9]
heapq.heapify(heap)
# heappushpop: push 2, then pop minimum
# More efficient than push + pop separately
result = heapq.heappushpop(heap, 2)
print('heappushpop(2):', result, '| heap:', heap)
# heapreplace: pop minimum, then push new item
# New item does NOT need to be larger (different from heappushpop)
result2 = heapq.heapreplace(heap, 4)
print('heapreplace(4):', result2, '| heap:', heap)
# Use case: maintaining a fixed-size top-k heap
# heappushpop is the standard pattern从零实现 MaxHeap
MaxHeap会反转比较规则：父节点必须大于或等于所有后代节点。只需反转上滤和下滤中的比较即可。另一种方法是使用取反包装类，或像 Python 的 heapq 一样对整数取相反数。从零实现这一结构可以说明，最小堆和最大堆是完全相同的结构，唯一变化是比较运算符。
class MaxHeap:
def __init__(self):
self._data = []
def push(self, val):
self._data.append(val)
i = len(self._data) - 1
while i > 0:
p = (i - 1) // 2
if self._data[p] < self._data[i]: # FLIP: parent < child = violation
self._data[p], self._data[i] = self._data[i], self._data[p]
i = p
else: break
def pop(self):
if not self._data: return None
if len(self._data) == 1: return self._data.pop()
result = self._data[0]
self._data[0] = self._data.pop()
i, n = 0, len(self._data)
while True:
g = i; l, r = 2*i+1, 2*i+2
if l < n and self._data[l] > self._data[g]: g = l # FLIP
if r < n and self._data[r] > self._data[g]: g = r # FLIP
if g == i: break
self._data[i], self._data[g] = self._data[g], self._data[i]; i = g
return result
h = MaxHeap()
for v in [5, 3, 8, 1, 4, 2]: h.push(v)
print([h.pop() for _ in range(6)]) # [8,5,4,3,2,1]从堆中删除任意元素
从堆中删除任意元素（而非根节点）的时间复杂度为 O(log n)，但需要知道该元素的索引。用最后一个元素替换目标元素，移除最后一个元素，然后对替换元素执行上滤或下滤（只有一个方向会违反堆性质）。这种技术用于带延迟删除的迪杰斯特拉算法，以及支持减键操作的优先队列。
def delete_at_index(heap, i):
n = len(heap)
heap[i] = heap[n - 1]
heap.pop()
if i >= len(heap):
return # deleted the last element
# Try sift-up first
p = (i - 1) // 2
if i > 0 and heap[i] < heap[p]:
while i > 0:
p = (i - 1) // 2
if heap[p] > heap[i]:
heap[p], heap[i] = heap[i], heap[p]; i = p
else: break
else: # sift down
j = i; n2 = len(heap)
while True:
s = j; l, r = 2*j+1, 2*j+2
if l < n2 and heap[l] < heap[s]: s = l
if r < n2 and heap[r] < heap[s]: s = r
if s == j: break
heap[j], heap[s] = heap[s], heap[j]; j = s
heap = [1, 3, 2, 7, 4, 5, 6]
print('Before:', heap)
delete_at_index(heap, 2) # delete element at index 2 (value=2)
print('After:', heap) # 2 removed, heap still valid频率最高的 K 个元素中的堆
频率最高的 K 个元素（LeetCode #347）使用大小为 k 的最小堆。维护一个最小堆，其中每个条目都是 (frequency, element)。处理每个不重复的元素：如果堆中的元素少于 k 个，则执行 push；否则，如果新元素的频率高于堆中的最小频率，则执行 pop 和 push。最终的堆包含频率最高的 k 个元素，时间复杂度为 O(n log k)。
import heapq
from collections import Counter
def top_k_frequent(nums, k):
count = Counter(nums)
# Min-heap of (frequency, num)
heap = []
for num, freq in count.items():
heapq.heappush(heap, (freq, num))
if len(heap) > k:
heapq.heappop(heap) # remove least frequent
return [num for freq, num in heap]
print(top_k_frequent([1,1,1,2,2,3], 2)) # [1, 2]
print(top_k_frequent([4,4,4,3,3,2,1], 2)) # [4, 3]堆在调度中的应用
除了竞赛编程之外，堆还为现实世界中的调度系统提供支持。操作系统的任务调度器使用优先队列（堆），始终运行优先级最高的就绪进程。事件驱动的模拟会使用按事件时间排序的最小堆，按照时间顺序处理事件。网络数据包调度器则根据服务质量类别为流量分配优先级。理解堆可以帮助您建立对所有这些系统的直观模型，在围绕排队和调度的系统设计面试中也经常会自然地用到这些知识。
import heapq
# Simple event-driven simulation using a heap
events = [] # (time, event_description)
def schedule(time, event):
heapq.heappush(events, (time, event))
def process_next():
time, event = heapq.heappop(events)
print(f't={time}: {event}')
return time, event
# Schedule events out of order:
schedule(10, 'Send email')
schedule(3, 'Open app')
schedule(7, 'Process request')
schedule(1, 'Start server')
# Process in time order:
while events:
process_next()
# Output: t=1, t=3, t=7, t=10 -- always in time order快速检查
测试您对本课中数据结构与算法——编程面试准备相关概念的理解。
课程回顾
在本课中，您学习了：从零实现带有向上调整和向下调整的MinHeap 和 MaxHeap，Floyd 的 O(n) heapify 算法，以及它为何优于逐个插入的 O(n log n) 方法；还学习了包括前 k 个高频元素和按索引删除在内的实际应用。接下来，我们将探索 Python 的 heapq 模块和最大堆技巧。
常见问题解答
「从零实现建堆、推入与弹出」课时是免费的吗？
是的 — 「从零实现建堆、推入与弹出」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 Coding Interview Prep 课程的其余内容，请升级到 CoddyKit PRO。 Coding Interview Prep 课程共包含 4 节课。
「从零实现建堆、推入与弹出」这节课中我会学到什么？
为推入实现向上调整，为弹出实现向下调整，然后使用 Floyd 算法以 O(n) 的时间从无序数组构建堆。 你通过在浏览器中直接运行的动手代码来练习 Coding Interview Prep，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 Coding Interview Prep 需要有经验吗？
无需任何先前经验。CoddyKit 上的 Coding Interview Prep 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 2 节课，共 4 节。
「从零实现建堆、推入与弹出」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 Coding Interview Prep 课中编写并运行代码吗？
能。每节 Coding Interview Prep 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。
此课程中的所有课时
- 堆性质与数组表示
- 从零实现建堆、推入与弹出
- Python heapq 与大根堆技巧
- 数据流中位数与 K 路合并