題幹與適用場景
一個非同步 API 要並行讀取使用者、訂單和推薦。任一關鍵任務失敗時，其他任務要盡快取消；客戶端斷線或總逾時後不能有孤兒協程。請用 asyncio.TaskGroup 設計，說明例外聚合、取消傳播和資源清理。
Python 文件把 TaskGroup 描述為結構化並行工具：作用域離開前等待任務；非取消例外會取消其他任務並以 ExceptionGroup 拋出。重點是精確描述生命週期。
面試官考察點
確認是否理解父子任務、兄弟取消、CancelledError、ExceptionGroup 與 async with join；是否把取消傳給資料庫、HTTP 和檔案操作；是否把響應後工作移至持久佇列。
回答前需要澄清的問題
- 三個讀取是否同等關鍵，推薦能否降級？
- 子任務是否使用連線、游標或暫存檔？
- 總逾時與呼叫方取消如何傳入？
- 例外是否需要分類、重試或轉成使用者訊息？
- 哪些工作必須在 HTTP 回應後繼續？
30 秒回答框架
「在 async with TaskGroup() 內建立三個任務，離開作用域前等待完成。非取消例外會取消兄弟並拋出例外組；用 except* 分類記錄。每個任務在 finally 關閉資源，外層用 asyncio.timeout() 或呼叫方取消控制總預算。持久工作寫入可靠佇列。」
分步驟深入解答
第一步：定義任務樹與結果契約
先定義回傳型別和關鍵性。使用者與訂單是必要資料，推薦可降級；這決定是否取消整組。所有任務屬於同一請求作用域，不要遺忘裸 create_task()。
async with asyncio.TaskGroup() as group:
user_task = group.create_task(load_user(user_id))
order_task = group.create_task(load_orders(user_id))
rec_task = group.create_task(load_recommendations(user_id))離開作用域時 TaskGroup 等待任務完成；不要在區塊外讀未完成任務逃避 join。
第二步：處理失敗與 ExceptionGroup
非 CancelledError 例外會觸發兄弟取消，所有任務結束後拋出 ExceptionGroup。用 except* 處理可預期錯誤，保留未知例外給上層。
try:
async with asyncio.TaskGroup() as group:
user = group.create_task(load_user(user_id))
orders = group.create_task(load_orders(user_id))
except* RetryableDependencyError as errors:
record_dependency_failures(errors.exceptions)
raise ServiceUnavailable from errors不要捕獲 BaseException 後靜默吞掉取消。
第三步：把取消傳給真正的 I/O
TaskGroup 只能取消 Python task；HTTP、資料庫與檔案驅動需要 timeout 或取消參數。每個任務在 finally 關閉連線、釋放 semaphore、刪除暫存檔並停止消費。
async def load_orders(user_id: str):
conn = await pool.acquire()
try:
return await conn.fetch("SELECT ...", user_id, timeout=1.5)
finally:
await pool.release(conn)驅動不能中斷查詢時，使用 statement timeout 或隔離背景工作。
第四步：設定總逾時並區分外部取消
用 asyncio.timeout() 包住整個任務組，三項工作共用預算；不要每項重發完整時間。
try:
async with asyncio.timeout(2.0):
result = await aggregate(user_id)
except TimeoutError:
return degraded_response("deadline")
except asyncio.CancelledError:
raise外部取消要繼續向上傳播，不可轉為成功回應；逾時與使用者取消應記不同原因。
第五步：理解 TaskGroup 與 gather 差異
gather() 預設把第一個例外傳給等待者，但其他任務不一定自動取消；呼叫方返回可能留下孤兒。TaskGroup 把生命週期綁在作用域，失敗時取消兄弟。
gather(return_exceptions=True) 適合明確允許部分失敗，但需逐項檢查，不能取代結構化清理。背景任務必須有界。
第六步：測試例外順序與資源清理
注入不同任務先失敗、同時失敗、呼叫方取消、總逾時、驅動逾時和 finally 失敗。斷言兄弟收到取消、資源歸還、無未完成任務，且例外組能定位子任務。
記錄任務名、請求 ID、耗時、取消原因與下游呼叫，不記隱私。用慢 I/O 重複測試找出競態。
高品質示範回答
「我在 TaskGroup 內建立三個任務，使用者和訂單必需、推薦可降級。作用域保證 join；非取消例外取消兄弟並形成 ExceptionGroup，except* 分類記錄。每個 I/O 在 finally 釋放資源。」
「外層 timeout 提供共享預算，CancelledError 繼續傳播。響應後工作寫入持久佇列。測試同時失敗、取消、逾時和慢 I/O，確認沒有孤兒任務與洩漏。」
常見錯誤
- 裸 create_task 後返回 → 孤兒任務 → 使用 TaskGroup 作用域。
- 吞掉 CancelledError → 父呼叫無法停止 → 清理後重新拋出。
- 每個任務完整逾時 → 總延遲失控 → 共用一個預算。
- 忽略 ExceptionGroup → 丟失平行失敗證據 → 用 except* 分類。
- 只取消 Python task → 資料庫仍運行 → 使用驅動 timeout 或可取消 I/O。
- 用 gather 取代所有結構化並行 → 降級與清理語義不清 → 明確契約。
追問及應對
追問一：TaskGroup 保證底層 I/O 立即停止嗎？
不保證。底層驅動需支援取消或 statement timeout；否則使用受控連線或隔離背景作業。
追問二：推薦失敗為何不取消整組？
依契約決定。可降級時捕獲該類例外回傳空推薦；影響安全或計費時讓例外離開作用域並取消兄弟。
追問三：如何把 ExceptionGroup 轉成 API 回應？
已知依賴錯誤映射成 503 或降級結果並記錄子例外；未知錯誤交給全域處理器，不把堆疊返回客戶端。
追問四：哪些工作應離開 TaskGroup？
響應後郵件、索引和批次寫入持久佇列，由獨立 worker 重試；請求組只承載請求生命週期工作。
追問五：finally 清理也拋例外怎麼辦？
讓清理路徑最小且可觀察，將清理例外作附加上下文，優先保留原始取消或業務例外，避免覆蓋根因。