Python heapq 實戰：優先佇列、Top-K 與任務排序

一. 前言：不是每次排序，都需要把全部排好

#

很多 Python 程式一開始都會這樣寫：

tasks = sorted(tasks, key=lambda task: task.priority)
next_task = tasks[0]

資料少的時候，沒什麼問題。 但如果你每新增一筆資料就重新排序一次，或是你只想拿前 10 名，卻把 100 萬筆全部排完，那就有點浪費。 這時候標準庫 heapq 就很好用。 heapq 提供的是 heap，也就是堆積資料結構。 它不保證整個 list 完全排序。 它只保證最小的元素永遠在 index 0。 聽起來很小氣，對吧。 但很多工作剛好只需要這件事：

• 每次拿出目前優先度最高的任務
• 從大量資料裡找最大的幾筆或最小的幾筆
• 合併多個已排序序列
• 做簡單的排程模擬
• 寫一個不用每次全排序的待辦清單 今天拍拍君要帶你從 heappush()、heappop() 開始，做出一個穩定的 priority queue。 我們也會處理幾個真實世界會踩到的細節：同優先度怎麼排序、任務怎麼刪除、Top-K 什麼時候該用 heap，什麼時候直接 sorted() 比較簡單。 Heap 不是魔法。 它只是很務實地說：我不幫你整理全房間，我只把下一個最該拿的東西放在門口。

二. 安裝：標準庫內建，先準備練習專案

#

heapq 是 Python 標準庫，不需要安裝。 建立一個練習資料夾即可：

uv init heapq-lab
cd heapq-lab

不用 uv 也沒關係。 今天的範例都只靠標準庫，可以直接用 python basic_heap.py 跑。 heapq 的 API 很小，但觀念很值得練；一旦理解 heap invariant，之後看到 priority queue、scheduler、Dijkstra shortest path、event simulation，都會比較不慌。

三. 第一個 heap：把 list 變成「最小值在前面」的結構

#

先建立 basic_heap.py。 heapify() 會把 list 原地整理成 heap：

import heapq

numbers = [8, 3, 5, 1, 9, 2]
heapq.heapify(numbers)

print(numbers)
print(numbers[0])

你可能會看到：

[1, 3, 2, 8, 9, 5]
1

重點來了：這個 list 不是完全排序。 Heap 只保證最小值在 numbers[0]，剩下的順序是內部維持效率用的布局。 如果你真的要完整排序，請用 sorted()；如果你要一次一個拿出最小值，才用 heappop()：

while numbers:
    print(heapq.heappop(numbers))

輸出會是：

1
2
3
5
8
9

這就是 heap 的用法核心：維持一個可以快速拿出最小值的集合。

四. heappush 與 heappop：動態加入、動態取出

#

很多情境不是先拿到全部資料再排序。 任務可能會一個一個進來。 事件可能會在程式跑的過程中產生。 這時候可以用 heappush()：

import heapq

heap: list[int] = []

heapq.heappush(heap, 5)
heapq.heappush(heap, 1)
heapq.heappush(heap, 8)
heapq.heappush(heap, 3)

print(heap)
print(heapq.heappop(heap))
print(heapq.heappop(heap))
print(heap)

輸出：

[1, 3, 8, 5]
1
3
[5, 8]

heappush() 會把新元素放進 heap，同時維持 heap invariant。 heappop() 會拿出最小值，也會重新整理 heap。 時間複雜度大致是：

五. Priority Queue：用 tuple 排出任務優先度

#

heapq 比較元素時，會照 Python 的一般排序規則。 最常見做法是把任務包成 tuple，讓 priority 放在第一個欄位：

import heapq

tasks: list[tuple[int, str]] = []

heapq.heappush(tasks, (3, "整理 README"))
heapq.heappush(tasks, (1, "修掉登入 bug"))
heapq.heappush(tasks, (2, "補測試"))

while tasks:
    priority, title = heapq.heappop(tasks)
    print(priority, title)

輸出：

1 修掉登入 bug
2 補測試
3 整理 README

tuple 排序會先比第一個欄位。 第一個欄位相同，再比第二個欄位。 這很適合建立「數字越小越優先」的 priority queue。

Python 3.14 的原生 max-heap

#

Python 3.14 起，heapq 新增一組 _max API，不必再靠負數 key 才能把最大值放在最前面：

import heapq

scores = [18, 92, 41, 73]
heapq.heapify_max(scores)

print(heapq.heappop_max(scores))
heapq.heappush_max(scores, 88)
print(scores[0])

輸出是 92 與 88。如果專案還要支援 Python 3.13 以下，再用「推入負數、取出後轉回正數」的相容寫法。不要在同一個 heap 混用 min-heap 與 max-heap API；兩者維持的是相反 invariant。

六. 同優先度的坑：不要讓 task 物件自己互相比大小

#

tuple 很方便，但會有一個常見坑。 如果兩個 priority 都是 1，Python 會接著比較 tuple 的第二個欄位。 若第二個欄位是一般 dataclass 或自訂物件，可能會出現 TypeError: '<' not supported...。 解法是加一個永遠遞增的 counter 當 tie-breaker。

from __future__ import annotations

from dataclasses import dataclass
from itertools import count
import heapq

@dataclass(frozen=True)
class Task:
    title: str
    owner: str

counter = count()
heap: list[tuple[int, int, Task]] = []

heapq.heappush(heap, (1, next(counter), Task("補測試", "pypy")))
heapq.heappush(heap, (1, next(counter), Task("修 bug", "pypy")))

while heap:
    priority, order, task = heapq.heappop(heap)
    print(priority, order, task.title)

這樣 tuple 的排序規則會變成：

1. 先比 priority
2. priority 一樣時，比加入順序
3. 永遠不需要直接比較 Task 這個技巧很重要。 它可以讓 priority queue 穩定，也就是同優先度任務會照加入順序取出。 使用者通常不喜歡同樣優先度的東西突然亂跳。 程式也一樣。

七. 做一個可重用的 PriorityQueue

#

接著把剛才的想法包起來。 實務上可以先寫一個小 wrapper：

from itertools import count
import heapq

class PriorityQueue:
    def __init__(self) -> None:
        self._heap: list[tuple[int, int, object]] = []
        self._counter = count()

    def push(self, value: object, priority: int) -> None:
        heapq.heappush(self._heap, (priority, next(self._counter), value))

    def pop(self) -> object:
        if not self._heap:
            raise IndexError("pop from empty priority queue")
        return heapq.heappop(self._heap)[2]

    def __len__(self) -> int:
        return len(self._heap)

使用方式：

queue = PriorityQueue()

queue.push("整理 README", priority=3)
queue.push("修掉登入 bug", priority=1)
queue.push("補測試", priority=2)

while len(queue) > 0:
    print(queue.pop())

這個包裝有幾個好處。 呼叫端不用知道 tuple 長什麼樣子，tie-breaker 也被藏在內部。 未來如果要加刪除、更新優先度、統計資訊，也有地方可以放。 不要過度包裝很重要，但 priority queue 的 tuple 很容易被寫散，這種小 wrapper 通常值得。

八. 更新與刪除任務：用 lazy deletion 比直接改 heap 乾淨

#

Heap 有一個麻煩點：它很擅長拿最小值，但不擅長從中間刪任意元素。 你可以硬找出元素、刪掉、再 heapify()，但那會變成 O(n)，而且程式很容易亂。 常見做法是 lazy deletion：先把舊任務標記成 removed，等它未來被 pop 到最前面時再跳過。

REMOVED = object()

def remove(entry: list[object]) -> None:
    entry[-1] = REMOVED

def pop_valid(heap: list[list[object]]) -> object:
    while heap:
        priority, order, task = heapq.heappop(heap)
        if task is not REMOVED:
            return task
    raise IndexError("pop from empty priority queue")

更新優先度時，建立一筆新 entry 推進 heap，再把舊 entry 標記成 removed。 它的代價是 heap 裡會短暫留一些垃圾 entry。 如果任務更新非常頻繁，可以定期重建 heap；對一般小工具來說，lazy deletion 已經夠乾淨。

九. Top-K：只保留目前最好的 K 筆

#

另一個 heapq 常見場景是 Top-K。 假設我們有很多筆分數，只想找最高的 5 筆。 最直覺的寫法是：

top5 = sorted(scores, reverse=True)[:5]

這很好讀，資料量不大時拍拍君完全支持。 但如果資料很多，而且 K 很小，可以用 heap 只保留 K 筆：

import heapq

def top_k(scores: list[int], k: int) -> list[int]:
    heap: list[int] = []

    for score in scores:
        if len(heap) < k:
            heapq.heappush(heap, score)
        elif score > heap[0]:
            heapq.heapreplace(heap, score)

    return sorted(heap, reverse=True)

這裡的 heap 保存目前最高的 K 筆。 heap[0] 是這 K 筆裡最小的那一筆；如果新分數比它還高，就用 heapreplace() 把目前最小的踢掉。 複雜度是 O(n log k)。 當 K 遠小於 n 時，很划算。 不過標準庫已經有現成 helper：

import heapq

top5 = heapq.nlargest(5, scores)
smallest5 = heapq.nsmallest(5, scores)

實務上請優先用 nlargest() 和 nsmallest()。 自己寫版本是為了理解。 不是每次都要把輪子重新削一遍。

十. 用 key 做 Top-K：找出最慢的 API request

#

nlargest() 支援 key 參數，跟 sorted() 很像。 例如找出最慢的 API request：

import heapq

logs = [
    {"path": "/api/search", "duration_ms": 182},
    {"path": "/api/report", "duration_ms": 1240},
    {"path": "/api/export", "duration_ms": 2010},
]

slowest = heapq.nlargest(2, logs, key=lambda log: log["duration_ms"])

for log in slowest:
    print(log["duration_ms"], log["path"])

輸出：

2010 /api/export
1240 /api/report

這種寫法很適合資料分析前處理、log 掃描、簡單監控工具。 如果你只想看前幾名，nlargest() 比完整排序更貼近需求；如果你要的是排序後的完整頁面、分頁結果、穩定報表，那 sorted() 比較清楚。

十一. heapq.merge：合併多個已排序序列

#

heapq.merge() 可以把多個已排序 iterable 合併成一個排序結果。 它很適合合併多個 log stream：

import heapq

service_a = [
    ("10:00:01", "api started"),
    ("10:00:08", "request /api/search"),
    ("10:00:12", "request /api/export"),
]

service_b = [
    ("10:00:03", "worker started"),
    ("10:00:05", "job accepted"),
    ("10:00:20", "job finished"),
]

for timestamp, message in heapq.merge(service_a, service_b):
    print(timestamp, message)

輸出：

10:00:01 api started
10:00:03 worker started
10:00:05 job accepted
10:00:08 request /api/search
10:00:12 request /api/export
10:00:20 job finished

注意：輸入本身必須已排序。 merge() 不會先幫你把每個來源排序。 它的優點是 lazy，會一邊需要一邊產生結果，不必先把所有資料放進一個大 list。 如果你只有兩個小 list，sorted(service_a + service_b) 反而更直覺；但當資料開始變大，heapq.merge() 就很香。

十二. 測試 Priority Queue：順序、穩定性、空佇列

#

Priority queue 最怕的是邊界條件。 至少要測三件事：不同 priority 是否照順序 pop、同 priority 是否保留加入順序、空佇列是否丟出你預期的錯誤。 這些測試看起來小，但很有用；priority queue 的 bug 常常不是「完全不能跑」，而是同優先度時順序亂掉、空佇列時錯誤訊息怪怪的、或更新優先度後舊任務又冒出來。 資料結構越小，越值得用測試把行為釘住。

十三. 常見選擇：heapq、deque、queue.PriorityQueue 怎麼分？

#

Python 裡有幾個名字看起來很像的工具。 它們用途不太一樣。

十四. 什麼時候不要用 heap？

#

heapq 很好，但不是萬用。 以下情況，拍拍君通常不會先拿 heap：

1. 你需要完整排序結果
2. 你常常要依照任意欄位搜尋或刪除
3. 你的資料量很小，sorted() 已經夠清楚
4. 你需要多執行緒阻塞 queue
5. 你需要資料庫等級的查詢、索引與持久化 例如待辦事項 App 如果只有 30 筆任務，直接 sorted(tasks, key=lambda task: (task.priority, task.created_at)) 可讀性可能更好。 但如果你每秒收到上千個事件，每次只處理下一個，heap 就合理很多。 選工具的順序可以這樣想：先寫出清楚版本，確認瓶頸真的在排序或取下一個元素，再換成 heap。 這樣你的程式會從需求長出來，而不是從演算法題目長出來。

十五. 小結：heap 是「下一個是誰」的資料結構

#

今天我們整理了 heapq 的核心用法：

• heapify() 可以把 list 原地整理成 heap
• heappush() 和 heappop() 適合動態加入與取出最小值
• Python 的 heapq 是 min-heap，max-heap 常用負數 key 模擬
• Python 3.14 起可直接使用 heapify_max()、heappush_max() 與 heappop_max()
• Priority queue 常用 (priority, order, value) 避免同優先度比較物件
• 更新與刪除任務時，可以用 lazy deletion
• nlargest() / nsmallest() 很適合 Top-K
• heapq.merge() 可以 lazy 合併已排序序列 拍拍君覺得 heapq 最值得記住的，不是 API 名字。 而是它的思考方式。 當你不需要完整排序，只需要穩定地知道「下一個最重要的是誰」，heap 就是很漂亮的工具。 它沒有很花俏。 但在任務排序、log 分析、Top-K、事件模擬裡，它會默默省掉很多不必要的排序工作。 小小一個標準庫模組，該出手時還是很有力。拍拍。

延伸閱讀

#

• Python 官方文件：heapq
• Python collections 模組：讓你的資料結構更強大
• Python APScheduler 實戰：讓程式定時執行背景工作
• Python graphlib 實戰：拓撲排序、依賴圖與工作流程排程
• Python pytest fixtures 實戰：測試資料、依賴注入與乾淨重用