本章目標理解為什麼即時插單問題適合用優先佇列,而不是每次全部重排
用 Python 建立可重現的非搶占式工單處理流程
把客服工單、系統告警與臨時支援需求轉成可驗證的等待時間問題
情境與限制情境:你手上正在處理例行工作,但新的客訴、告警或主管交辦會不斷插進來,必須決定下一件先做什麼
輸入:每個任務的名稱、優先等級、到達時間與處理時長
輸出:實際處理順序、每個任務的等待時間,以及整體等待摘要
限制:一次只能處理一件事,任務一旦開始就先做完,不做中途搶占
可重現規則本章程式碼位置:
examples/ch08/本章核心模組:
examples/ch08/triage.py測試:
pytest -q tests/test_ch08_priority_queues_and_triage.py典型用途:客服工單、IT 支援、值班告警、營運插單需求
持續插單的場景為何需要優先佇列¶
很多工作現場不是先把所有事情排好就能照表操課,而是任務會在過程中不斷插進來。這時如果每來一件新任務就重新排序整張清單,成本高,也很難說明目前的處理邏輯。
優先佇列的價值,在於它可以隨時把「目前已到達、且最該先做的事」放到最前面。這種模型很適合客服中心、維運值班與營運支援:高優先任務要先做,但已經開始處理的工作通常不會被硬生生打斷。
為什麼不是每次都重新排序¶
如果任務會持續到達,直覺上你可能想在每個時間點都重新排序一次。但實務上,真正需要的其實只有兩個動作:
把新到的任務放進待處理集合。
從待處理集合裡,拿出目前最重要的那一件。
這正是優先佇列最擅長的事。當待處理任務變多時,heapq 可以讓你比反覆整批排序更穩定地維持決策節奏。
非搶占式規則為什麼重要¶
本章採用的是非搶占式模型,也就是任務一旦開始,就先做完再換下一件。這比理論上的完全即時搶占更貼近日常工作,因為很多任務在處理途中切換,反而會造成額外成本。
所以本章的決策點不是「任何時刻都能打斷」,而是「每當一件事做完時,下一件該輪到誰」。
只知道順序還不夠¶
如果你只拿到處理順序,通常還無法回答管理上真正會被問的問題,例如:
平均來說,工單等了多久才開始處理?
最久的那一件卡了多久?
是不是某幾種高優先插單,讓其他事情普遍變慢?
因此本章除了處理順序,也補上等待時間摘要。這樣你不只知道系統怎麼排,還能用數字說明目前流程的服務品質。
實作範例¶
請參考 examples/ch08/triage.py。請在專案根目錄執行程式,以確保路徑正確。
from examples.ch08.triage import build_triage_order, summarize_waiting_times
tasks = [
{"name": "系統崩潰修復", "priority": 1, "arrived_at": 0, "duration": 30},
{"name": "密碼重置", "priority": 3, "arrived_at": 5, "duration": 5},
{"name": "VIP 請求", "priority": 2, "arrived_at": 10, "duration": 15},
]
order = build_triage_order(tasks)
print("處理順序:", [t["name"] for t in order])
# 處理順序: ['系統崩潰修復', 'VIP 請求', '密碼重置']
summary = summarize_waiting_times(tasks)
print("最久等待任務:", summary["longest_wait_task"])
# 最久等待任務: 密碼重置示範流程包含:
驗證優先等級、到達時間與處理時長是否合法
把已到達任務推入優先佇列
在每個完成時點選出目前最高優先任務
估算每個任務實際等待了多久
彙整平均等待時間、最久等待任務與最大等待時長