当前位置：首页 > news >正文

堆排序工程化适配：从题目代码到生产级优先队列的改造

news 2026/7/8 15:02:21

堆排序工程化适配：从题目代码到生产级优先队列的改造

一、LeetCode 上的堆排序能直接用于生产吗

刷算法题时，堆排序的实现通常是这样的：一个原地排序函数，几十行代码，用数组模拟二叉堆。这种写法能通过所有测试用例，但你如果把它直接放进生产项目的排序模块里，很可能出问题。

问题在哪？不是算法错了。是工程约束不一样。题目的输入是固定长度的整数数组（通常 ≤ 10^5），而生产环境的排序对象可能是流式数据、需要支持动态增删、要求线程安全，甚至要在分布式环境下运作。

这篇文章从一道 LeetCode 堆排序题出发，逐步加入工程化考量，最终得到一个可应用于实际项目的优先队列实现。

二、从算法堆到工程堆：关键差异

flowchart LR subgraph 算法题 A1[原地排序] --> A2[固定长度数组] A2 --> A3[单一数据类型] A3 --> A4[无并发要求] end subgraph 工程场景 B1[动态增删] --> B2[不定长流式数据] B2 --> B3[自定义比较器] B3 --> B4[并发安全] end A1 -.->|需要演进| B1 A2 -.->|需要演进| B2 A3 -.->|需要演进| B3 A4 -.->|需要演进| B4

工程化改造的核心关注点有四个：

类型泛化：不只排序整数，需要支持任意可比较类型。
比较器注入：升序、降序、自定义排序策略由外部传入。
动态容量：不预设固定大小，支持自动扩容。
并发安全：多读多写场景下的数据一致性保证。

三、工程化实现：泛型优先队列

from typing import TypeVar, Generic, Callable, Optional from threading import Lock T = TypeVar("T") class PriorityQueue(Generic[T]): """泛型优先队列：基于最小堆，支持自定义比较器与并发安全 设计要点： 1. 使用 TypeVar 实现泛型，不限定元素类型。 2. 比较器通过构造函数注入，默认使用小于号。 3. 内部用列表存储堆，动态扩容。 4. 通过读写锁保证并发场景下的数据一致性。 """ def __init__( self, comparator: Optional[Callable[[T, T], bool]] = None, initial_capacity: int = 16, ): # 默认比较器：a < b 为最小堆 # 若传入 lambda a, b: a > b，则变为最大堆 self._cmp = comparator if comparator else lambda a, b: a < b self._heap: list[T] = [] # 预分配容量，减少早期 push 的扩容次数 self._heap = [None] * initial_capacity # type: ignore self._size = 0 # 读写锁：允许多个读操作并发，写操作互斥 # 这里用简单的互斥锁演示，实际高并发场景可升级为 RWLock self._lock = Lock() def push(self, item: T) -> None: """插入元素，O(log n)""" with self._lock: # 扩容策略：2 倍扩容，摊销 O(1) if self._size >= len(self._heap): self._heap.extend([None] * len(self._heap)) # type: ignore self._heap[self._size] = item self._sift_up(self._size) self._size += 1 def pop(self) -> Optional[T]: """弹出堆顶元素，O(log n)""" with self._lock: if self._size == 0: return None top = self._heap[0] self._size -= 1 self._heap[0] = self._heap[self._size] self._heap[self._size] = None # 帮助 GC 回收 if self._size > 0: self._sift_down(0) return top def peek(self) -> Optional[T]: """查看堆顶元素但不弹出，O(1)""" with self._lock: return self._heap[0] if self._size > 0 else None def __len__(self) -> int: return self._size def _sift_up(self, idx: int) -> None: """上浮操作：将位置 idx 的元素向上调整到正确位置""" while idx > 0: parent = (idx - 1) >> 1 # 等价于 (idx - 1) // 2，位运算更快 if self._cmp(self._heap[idx], self._heap[parent]): self._heap[idx], self._heap[parent] = ( self._heap[parent], self._heap[idx], ) idx = parent else: break def _sift_down(self, idx: int) -> None: """下沉操作：将位置 idx 的元素向下调整到正确位置""" size = self._size while True: smallest = idx left = (idx << 1) + 1 # 2 * idx + 1 right = left + 1 # 2 * idx + 2 if left < size and self._cmp(self._heap[left], self._heap[smallest]): smallest = left if right < size and self._cmp(self._heap[right], self._heap[smallest]): smallest = right if smallest == idx: break self._heap[idx], self._heap[smallest] = ( self._heap[smallest], self._heap[idx], ) idx = smallest # ---- 使用示例 ---- class Task: """模拟生产环境中的任务对象""" def __init__(self, name: str, priority: int): self.name = name self.priority = priority def __repr__(self): return f"Task({self.name}, pri={self.priority})" if __name__ == "__main__": # 按优先级升序排列的任务队列 pq = PriorityQueue[Task]( comparator=lambda a, b: a.priority < b.priority ) pq.push(Task("日志清理", 3)) pq.push(Task("支付回调", 1)) pq.push(Task("报表生成", 5)) pq.push(Task("告警通知", 2)) while len(pq) > 0: print(pq.pop()) # 输出顺序：支付回调 → 告警通知 → 日志清理 → 报表生成