Codetopia 的週二晚間，總帶著一股咖啡因與腎上腺素混合的氣味。為了不影響白天的市民服務，系統維運團隊習慣在離峰的深夜執行「夜巡」，盤點系統中延遲的任務。今晚，調度中心的派工調度員 Liam 接下了一個棘手的任務：揪出所有違反「服務等級協議 (SLA)」超過 10 分鐘的拖油瓶工單。所謂的 SLA，就是系統對市民許下的「完工時限承諾」，一旦超時，市民的滿意度就會亮起紅燈。
但問題來了，這些工單根本不是整齊地排在一個地方。它們就像一群調皮的貓，散落在城市系統的各個角落：
① 緊急公文籃 (PriorityQueue)：市長桌上十萬火急的紅色文件夾。
② 待辦事項牆 (Deque)：掛在牆上一長串的便利貼，隨時有人從頭或尾貼上新的。
③ 專案計畫書 (MacroCommand)：一本厚厚的活頁夾，裡面夾著好幾份獨立的施工單，像俄羅斯娃娃。
④ 各區工地白板 (Pools)：北、中、南三個工務所掛在牆上，各自的臨時任務清單。
Liam 的巡檢方式，簡直是一場體力活災難。他得先跑去市長辦公室翻閱紅色文件夾，再跑到牆邊一張張撕便利貼，接著要打開每一本專案計畫書，把裡面的施工單抽出來看，最後還要分別打電話給三個工務所，請他們回報白板上的內容。
更慘的是，只要任何一個地方改變了記錄方式——比如北區工務所突然改用電子郵件——Liam 整套巡檢流程就當場癱瘓。
就在 Liam 瀕臨崩潰之際，巡檢分析員 Harper 帶著一貫從容的微笑飄了過來。「嘿，Liam，你不覺得你更像個快遞員，而不是調度員嗎？」她說，「我們得先把『親自跑腿去拿工單』和『審查工單』這兩件事分開。」
「我想要的，」Harper 敲了敲桌子，「只是一個『萬能助理』。我只要對他說『下一份』，他就會自動從文件籃、便利貼牆、活頁夾或任何地方，把下一張該處理的工單遞到我手上。我不想再管這些東西到底藏在哪了！」
Iterator (迭代器/游標)：提供一種統一的方法來依序存取一個聚合物件（容器）中的各個元素，而又不需暴露該物件的內部表示。
在 Harper 到來前，Liam 的螢幕上正是這段充滿「味道」的程式碼，堪稱義大利麵的食譜：
def night_inspection_v1(prio_q, backlog, macro_cmd_list, pools):
overdue_list = []
# 壞味道 A：遍歷邏輯和商業邏輯混雜
for container in [prio_q, backlog, macro_cmd_list] + pools:
for item in container:
# 壞味道 B：手動展開巢狀結構，未來還有一百層要 if
if isinstance(item, MacroCommand):
for sub_item in item.get_sub_commands():
if sub_item.is_overdue(10):
# 壞味道 C：順序、過濾、輸出格式全寫在一起，怎麼測？
overdue_list.append(f"Macro Sub: {sub_item.id}")
elif item.is_overdue(10):
overdue_list.append(f"Single: {item.id}")
return overdue_list
這段程式碼的問題顯而易見：
脆弱的耦合：巡檢邏輯寫死了容器的種類和順序，任何變動都是一場災難。
職責混亂：一個函式裡，同時決定了巡覽順序、過濾條件和輸出格式，牽一髮而動全身。
重複與遺漏的溫床：多個容器各自維護索引，一旦有工單插隊或移除，很容易就導致漏巡或重複巡檢。（「上次那個 bug 就是這樣來的！」Liam 崩潰地喊道。）
Iterator 模式的核心思想，就是將巡覽的責任從容器本身分離出來，交給一個獨立的「迭代器」物件。容器只需要懂得如何「產生一個迭代器」，呼叫端就可以透過這個統一的介面，對任何類型的容器進行一致的巡覽操作。
關鍵設計：
外部迭代器 (External Iterator)：由呼叫端（例如 Harper 的巡檢程式）主動呼叫 next() 來取得下一個元素。這給了呼叫端完全的控制權，可以隨時暫停、恢復，甚至將多個迭代器合併。
複合/扁平化迭代器 (Composite/Flattening Iterator)：專門用來對付像 MacroCommand 這樣的樹狀結構。它會用深度優先(DFS)或廣度優先(BFS)演算法，將整個樹「攤平」成一個線性的序列，讓呼叫端感覺就像在巡覽一個簡單的清單。
快照 (Snapshot) vs. 即時 (Live) 策略：
快照：在迭代器建立的瞬間，就複製一份當時的資料快照。這保證了巡覽過程中的穩定性和可重現性，不受後續資料變動的影響。非常適合用來產生報表或進行審計。
即時：每次呼叫 next() 都會去存取當下最新的資料。這允許在巡覽過程中看到新加入的工單，適合需要即時反應的場景。（註：要實現真正的「即時」，需要底層資料源支援增量拉取，例如 Deque 或 Channel，僅用 Python list 的迭代器本質上仍是快照視圖。）
合流與排序策略：目前的管線是「先合流、再攤平」，這意味著排序鍵（如優先級）主要應用在根層級的項目，子任務會跟隨其母巨集一起出現。若要實現「所有葉節點的全域排序」，則應調整為「先對各來源攤平、再將所有葉節點合流排序」。本文下方提供 night_inspection_v4_global_leaves() 範例可直接套用。
✅ 跨多種容器巡覽：當你需要用同樣的方式走遍 List、Queue、Tree 等不同結構的資料集合時。
✅ 隱藏集合實作：當你想讓呼叫端專注於「做什麼」而不是「怎麼走」，避免暴露容器的內部細節。
✅ 需要可暫停/恢復的巡覽：當巡覽過程可能被打斷，且需要在中斷點繼續時，外部迭代器是你的好朋友。
✅ 處理巢狀或遞迴結構：使用扁平化迭代器可以優雅地將複雜的樹狀結構線性化。
⛔ 流程骨架固定，只更換實作細節：如果你的演算法流程是固定的，只是某些步驟的「做法」不同，那更適合 Template Method (Day 21)。
⛔ 同一步驟，切換不同演算法：如果只是想在同一個環節切換不同的處理策略（例如排序演算法），Strategy (Day 15) 更為直接。
⛔ 多方物件網狀協調：如果問題的核心是多個物件之間複雜的互動與依賴，需要一個中央協調者來解耦，那請期待 Mediator (Day 19)。
導播，鏡頭拉一下！讓我們從三個不同層次，看看這個「統一游標」是如何運作的。
| 層級 | 對應概念 | Codetopia 詞彙 |
|---|---|---|
| 微觀 (GoF) | IterableWorklist 產生 WorkIterator； CompositeCommand 提供 iter() 攤平成子命令序列。 |
各工單容器提供「巡覽器」； 巨集命令懂得如何「自我攤平」。 |
| 中觀 (EIP/EDA/Actor) | 以「Cursor」訊息語義抽象化資料流 (Pull-based Consumer)； 多來源以 PriorityMergeIterator 合流。 |
Harper 的巡檢器向多個資料源「拉取」工單， 並依優先級穩定排序合併。 |
| 宏觀 (MAS) | InspectorAgent 透過黃頁 (DF) 查詢游標供應能力， 串接夜巡鏈，結果回報給 ReportAgent。 | 巡檢代理查詢哪些單位能提供「工單游標」， 並啟動巡檢任務。 |
Mermaid｜微觀 GoF 結構
（註：createIterator() 是 GoF 的經典命名，在 Python 中通常以 iter() 魔法方法實現。）
Mermaid｜中觀 EIP 資訊流
現在，讓我們看看英雄 Harper 如何運用一系列精巧的迭代器組合，真正實現一個穩定、可回放且職責分離的夜巡腳本。
import time
import heapq
import itertools
from collections import deque
# --- 1. 核心物件與協定 ---
class WorkItem:
def __init__(self, id, priority, content, status="Pending"):
self.id = id
self.priority = priority
self.content = content
self.status = status
self.created_at = time.time() # 用於穩定排序
def is_overdue(self, minutes, now=None):
"""檢查是否逾期，now 參數用於可重現的測試"""
now = now or time.time()
return (now - self.created_at) > minutes * 60
def __repr__(self): return f"WorkItem({self.id}, P{self.priority}, {self.status})"
class Command(WorkItem): pass
class MacroCommand(Command):
def __init__(self, id, priority, content, sub_commands=None):
super().__init__(id, priority, content)
self._sub_commands = sub_commands or []
def add(self, cmd): self._sub_commands.append(cmd)
def __iter__(self):
return iter(self._sub_commands)
# --- 2. 迭代器工具箱 ---
def children_of(node):
"""協定：回傳節點的子節點迭代器。
1) 直接支援 MacroCommand（__iter__ 返回直屬子節點）
2) 若物件實作 children() 亦視為 Composite
3) 其他型別視為葉節點（回空迭代器）
"""
if isinstance(node, MacroCommand):
return iter(node)
if hasattr(node, "children") and callable(getattr(node, "children")):
return iter(node.children())
return iter(())
def is_leaf(node):
"""協定：如果一個節點沒有子節點，它就是葉節點。"""
try:
it = children_of(node)
return next(it, None) is None
except TypeError:
return True
def dedup_key(item):
"""去重鍵：若有 origin 則使用 (origin, id)，否則僅使用 id。"""
item_id = getattr(item, "id", None)
origin = getattr(item, "origin", None)
return (origin, item_id) if origin is not None else item_id
class FlatteningIterator:
"""健壯的扁平化迭代器(DFS)，具備去重與循環偵測"""
def __init__(self, roots):
self._stack = deque([(iter(roots), None)]) # (iterator, owner_oid)
self._seen_keys = set()
self._visiting_oids = set()
def __iter__(self): return self
def __next__(self):
while self._stack:
it, owner_oid = self._stack[-1]
try:
item = next(it)
oid = id(item)
key = dedup_key(item)
if (key is not None and key in self._seen_keys) or oid in self._visiting_oids:
continue
if key is not None:
self._seen_keys.add(key)
kids_iter = children_of(item)
# 避免一次性物化所有子節點
first_kid = next(kids_iter, None)
if first_kid is not None:
self._visiting_oids.add(oid)
full_kids_iter = itertools.chain([first_kid], kids_iter)
self._stack.append((full_kids_iter, oid))
return item
except StopIteration:
self._stack.pop()
if owner_oid is not None:
self._visiting_oids.discard(owner_oid)
raise StopIteration
def priority_merge(sources):
"""合併多個 *已排序* 的來源流；每個來源需以 (priority, created_at) 非遞減輸出。"""
counter = itertools.count()
heap = []
for i, source in enumerate(sources):
try:
it = iter(source)
item = next(it)
heapq.heappush(heap, (item.priority, item.created_at, next(counter), i, item, it))
except StopIteration:
pass
while heap:
_, _, _, i, item, it = heapq.heappop(heap)
yield item
try:
next_item = next(it)
heapq.heappush(heap, (next_item.priority, next_item.created_at, next(counter), i, next_item, it))
except StopIteration:
pass
def priority_merge_all(sources):
"""合併 *未排序* 的來源：一次收集全部元素，再以 (priority, created_at) 全域排序。
註：較吃記憶體，但不要求來源各自保序。
"""
counter = itertools.count()
heap = []
for source in sources:
for item in source:
heapq.heappush(heap, (item.priority, item.created_at, next(counter), item))
while heap:
yield heapq.heappop(heap)[-1]
# --- 3. 呼叫端 (Harper 的巡檢腳本 V4) ---
def night_inspection_v4(worklists):
print("--- Night Inspection v4 (Snapshot, Merged, Flattened & Filtered) ---")
snapshot_time = time.time() # 建立一個固定的時間基準點
# 步驟 1: 以穩定排序合併所有來源
# 假設來源本身已大致排序；若無，請改用下方的 priority_merge_all()
merged_stream = priority_merge(worklists)
# 步驟 2: 將合併後的流攤平 (處理巢狀巨集)
snapshot_roots = list(merged_stream)
flattened_stream = FlatteningIterator(snapshot_roots)
# 步驟 3: 加上商業邏輯過濾 (使用協定判斷葉節點)
overdue_tasks = (
task for task in flattened_stream
if is_leaf(task)
and task.status == "Pending"
and task.is_overdue(0.0001, now=snapshot_time)
)
for task in overdue_tasks:
print(f"OVERDUE Task Found: {task}")
def night_inspection_v4_global_leaves(worklists):
"""備用管線：先攤平各來源並只取『葉節點』，再做全域排序合流（葉節點全域有序）。"""
print("--- Night Inspection v4 (Global Leaf Order) ---")
snapshot_time = time.time()
# 先將每個來源攤平成葉節點流（順序未保證）
leaf_streams = []
for src in worklists:
flat = (node for node in FlatteningIterator(src) if is_leaf(node))
leaf_streams.append(flat)
# 未排序來源 -> 使用 priority_merge_all 做全域排序
globally_sorted = priority_merge_all(leaf_streams)
overdue_tasks = (
task for task in globally_sorted
if task.status == "Pending" and task.is_overdue(0.0001, now=snapshot_time)
)
for task in overdue_tasks:
print(f"OVERDUE Task Found: {task}")
# --- 場景設定 ---
if __name__ == "__main__":
prio_q = [Command("C001", 1, "Fix traffic light"), Command("C003", 1, "Restart server")]
time.sleep(0.01)
backlog = [Command("C002", 2, "Update DB schema")]
nested_macro = MacroCommand("M002", 4, "Nested Batch")
nested_macro.add(Command("C006", 4, "Sub-Sub: Check disk space"))
weekend_batch = MacroCommand("M001", 3, "Weekend Batch")
weekend_batch.add(Command("C004", 3, "Sub: Backup files"))
weekend_batch.add(Command("C005", 3, "Sub: Clear logs"))
weekend_batch.add(nested_macro)
all_worklists = [prio_q, backlog, [weekend_batch]]
night_inspection_v4(all_worklists)
（旁白：這才是真正的「高內聚，低耦合」！每個迭代器各司其職，像一條優雅的生產線，最終的業務邏輯 night_inspection_v4 清晰得像詩一樣。）
動手挑戰：請為 night_inspection_v4 的管線加上一個 WindowIterator，讓它不是一次處理完所有項目，而是每處理 2 筆 overdue task 就暫停一次，回報「已處理一個批次」。你會如何設計這個 WindowIterator？
情境二選一：深夜巡檢時，突然有緊急工單插入佇列。
A. 堅持用 Snapshot 策略：保證本次報表的可重現性與一致性，巡檢結束後再跑一次 Live 模式補上遺漏的。
B. 立刻切換成 Live 策略：接受最終報表順序可能有些微震盪，但能換取最高的即時性。
如果你是 Harper，你會選擇 A 還是 B？請用一句話說明你的理由！
反模式紅旗 🚩
外洩實作：呼叫端還在使用 list[i] 或 stack.pop() 來控制巡覽節奏。
副作用大師：在 next() 方法裡做檔案 IO 或網路請求，導致巡覽順序和可重現性完全失控。
邏輯混雜：把攤平邏輯寫在業務層，導致同樣的判斷邏輯散落在各處。
排序混沌：不定義明確的排序準則，導致每次巡覽輸出的順序都不同，讓測試與審計人員想哭。
Iterator 模式看似簡單，但將其視角拉高，會發現它與更宏觀的架構思想緊密相連：
EIP/EDA (事件驅動)：可以將「游標」視為一種可定址的讀取視窗 (Readable Window)。中介軟體中的 Pull-based Consumer 就是 Iterator 思想的體現，由消費者決定何時「拉取」下一條訊息，從而實現流量控制 (Backpressure)。
Actor 模型：每個資料來源都可以看作一個「可拉取訊息」的 Actor 信箱。Harper 扮演的巡檢器就是一個 Puller Actor，它以自己的節奏去各個信箱拉取信件，而不是被動地被信件淹沒。
今日精華：統一游標巡覽，攤平巢狀多源；可暫停續跑，排序一致可回放。
今夜，Harper 用優雅的 Iterator 模式，將 Liam 從巢狀迴圈的地獄中拯救出來。工單清單終於清晰明瞭。但問題又來了...
明日預告：游標巡覽雖然完成了，但不同維修隊伍在執行這些工單時，發現任務之間有複雜的交互依賴和資源衝突。誰來扮演交通警察，協調這一切呢？敬請期待 Day 19：Mediator (協調中心)！
為了確保在不支援 Mermaid 渲染的環境中也能正常閱讀，以下提供文中圖表的 ASCII 替代版本：
┌─────────────────────┐ ┌─────────────────────┐
│ IterableWorklist │ │ Iterator │
│ <<interface>> │ │ <<interface>> │
├─────────────────────┤ ├─────────────────────┤
│ +createIterator(): │ │ +hasNext(): bool │
│ Iterator │ │ +next(): WorkItem │
└─────────────────────┘ └─────────────────────┘
▲ ▲
│ │
┌─────┴─────┐ │
│ │ │
┌───▼─────┐ ┌───▼─────────┐ │
│Priority │ │MacroCommand │ │
│Queue │ │ │ ┌────▼─────────────┐
├─────────┤ ├─────────────┤ │FlatteningIterator│
│+create │ │+create │ ├──────────────────┤
│Iterator │ │Iterator │ │-stack: Deque │
│(): Iter │ │(): Iter │ │+hasNext(): bool │
└─────────┘ └─────────────┘ │+next(): WorkItem │
└──────────────────┘
▲
│
┌─────────────────┐ │
│ Client │ ......................
├─────────────────┤ uses
│-iterator: Iter │
│+inspectAll() │
└─────────────────┘
Harper Marco Merger Flattener
(Inspector) (Curator) (Priority) (Flattening)
│ │ │ │
│ 請求夜巡游標 │ │ │
├──────────────►│ │ │
│ │ 建立合併 │ │
│ ├───────────►│ │
│ │◄───────────┤ │
│ │ 回傳 │ │
│ │ │ │
│ │ 建立扁平化 │ │
│ ├────────────────────────────►│
│ │◄────────────────────────────┤
│ │ 回傳 │
│◄──────────────┤ │
│ 統一巡覽介面 │ │
│ │ │
│ 巡覽所有工單
├─────────────────────────────────────────────►│
│ next() │
│◄─────────────────────────────────────────────┤
│ 返回下一個葉節點工單 │
│ │
│ (重複直到所有工單巡覽完畢) │
│ │
多源工單容器 合併流 攤平流 過濾結果
┌─────────┐
│優先佇列 │ ──┐
│[C001, │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ C003] │ │ │ │ │ │ │ │
└─────────┘ ├───►│Priority │───►│Flattening │───►│Business │
│ │Merger │ │Iterator │ │Filter │
┌─────────┐ │ │ │ │ │ │ │
│待辦清單 │ ──┤ └─────────────┘ └─────────────┘ └─────────────┘
│[C002] │ │ │ │ │
└─────────┘ │ │ │ │
│ ▼ ▼ ▼
┌─────────┐ │ 按優先級排序 展開巢狀結構 只取逾期葉節點
│巨集命令 │ ──┘ (P1,P2,P3...) (攤平成線性) (狀態=Pending)
│[M001] │
└─────────┘
範例輸出流程：
C001(P1) → C003(P1) → C002(P2) → M001(P3) → C004(P3) → C005(P3) → C006(P4)
↓ 攤平
C001(P1) → C003(P1) → C002(P2) → C004(P3) → C005(P3) → C006(P4)
↓ 過濾(逾期+葉節點)
[逾期的具體工單清單]
FlatteningIterator 內部堆疊變化過程：
初始狀態：
┌─────────────────────┐
│ Stack Top: │
│ [C001,C003,C002,M001]│ ← 根層迭代器
└─────────────────────┘
處理到 M001(巨集命令) 時：
┌─────────────────────┐
│ Stack Top: │
│ [C004,C005,M002] │ ← M001的子命令迭代器
├─────────────────────┤
│ [C001,C003,C002,M001]│ ← 根層迭代器(已完成)
└─────────────────────┘
處理到 M002(嵌套巨集) 時：
┌─────────────────────┐
│ Stack Top: │
│ [C006] │ ← M002的子命令迭代器
├─────────────────────┤
│ [C004,C005,M002] │ ← M001的子命令迭代器
├─────────────────────┤
│ [C001,C003,C002,M001]│ ← 根層迭代器(已完成)
└─────────────────────┘
輸出順序：C001 → C003 → C002 → C004 → C005 → C006
┌─────────────────────┬─────────────────────┐
│ 快照策略 │ 即時策略 │
│ (Snapshot) │ (Live) │
├─────────────────────┼─────────────────────┤
│ T0: 建立快照 │ T0: 建立游標 │
│ [A,B,C] │ → 資料源 │
│ │ │
│ T1: 資料源變更 │ T1: 資料源變更 │
│ [A,B,C,D] (新增) │ [A,B,C,D] (新增) │
│ │ │
│ T2: 巡覽結果 │ T2: 巡覽結果 │
│ 仍然 [A,B,C] │ 包含 [A,B,C,D] │
│ │ │
│ ✓ 穩定可重現 │ ✓ 即時反應 │
│ ✓ 適合報表審計 │ ✓ 適合監控告警 │
│ ✗ 可能錯過新任務 │ ✗ 順序可能變動 │
└─────────────────────┴─────────────────────┘
iThome鐵人賽
看影片追技術