具代表性的面試主題

程式設計面試:如何用 asyncio.timeout_at 與 reschedule 管理動態截止時間?

程式題困難
Offer.cc 編輯團隊發佈 更新

題幹

一個非同步請求只有讀到上游中繼資料後才知道剩餘預算。請使用 asyncio.timeout 或 timeout_at 設計動態截止時間,並解釋 reschedule、expired、CancelledError、TimeoutError、巢狀上下文與 wait_for 的差異。

題幹與適用場景

一個非同步請求先讀取路由中繼資料,之後才知道下游呼叫的剩餘預算。實作要求:統一使用事件迴圈單調時鐘,允許先建立無截止時間的上下文、隨後設定絕對截止時間;逾時後只能在上下文外處理 TimeoutError,呼叫方主動取消仍要繼續傳播。

Python 文件說明 asyncio.timeout() 返回可重排的非同步上下文管理器,timeout_at() 接受事件迴圈時鐘上的絕對時間。上下文內的取消會在邊界外轉換為 TimeoutError

面試官考察點

考察候選人能否區分相對延遲與絕對截止時間、業務逾時與外部取消;是否理解 reschedule()expired()、巢狀上下文和 wait_for() 的取消語意,並能給出共享預算與資源清理方案。

回答前需要釐清的問題

  • 預算由哪個元件擁有,是否已消耗讀取中繼資料的時間?
  • 下游客戶端、資料庫和佇列是否支援逾時或取消?
  • 子呼叫需要共享一個絕對 deadline,還是各自擁有獨立預算?
  • 逾時是可降級結果、重試,還是直接返回錯誤?
  • 執行環境最低 Python 版本是否包含 asyncio.timeout

30 秒回答框架

「我用事件迴圈的 loop.time() 計算絕對 deadline。先用 asyncio.timeout(None) 建立作用域,讀到中繼資料後呼叫 cm.reschedule(deadline);上下文內發生取消,離開時會轉成 TimeoutError,所以只在外層捕獲。外部 CancelledError 不轉換,清理後繼續拋出。下游 I/O 同樣接收剩餘預算,巢狀呼叫不能重新發放完整逾時。需要觀察 cm.expired() 區分到期和其他退出。」

分步深入解答

第一步:用單調時鐘表達絕對截止時間

不要用牆上時鐘計算剩餘時間,因為系統校時會跳變。用 loop.time() 取得單調值,統一把 deadline 傳給所有下游呼叫。

python
loop = asyncio.get_running_loop()
deadline = loop.time() + 2.0
async with asyncio.timeout_at(deadline):
    await call_dependency(deadline)

第二步:未知預算時先建立可重排上下文

請求開始時還不知道預算,可用 asyncio.timeout(None) as cm。讀取中繼資料後計算絕對 deadline,再呼叫 cm.reschedule(deadline);不要銷毀上下文重建,避免中間工作失去統一邊界。

第三步:把剩餘預算傳到下游

每個客戶端根據 deadline - loop.time() 計算剩餘秒數,並把非正值視為立即失敗。這樣路由、資料庫和 HTTP 呼叫共享一個預算,不會層層重新計時。

第四步:理解取消到逾時的轉換

逾時上下文在內部取消目前任務,並在上下文退出時將該取消轉換為 TimeoutError。因此 TimeoutError 只能在 async with 外捕獲;在內部捕獲它會失效。

python
try:
    async with asyncio.timeout(1.0):
        await slow_call()
except TimeoutError:
    return degraded_result()

第五步:保留外部取消

請求被客戶端斷開或父任務取消時,CancelledError 不是業務逾時。清理連線、鎖和暫存檔後繼續拋出,不能把使用者取消包裝成降級成功。

第六步:比較 timeout、timeoutat 與 waitfor

timeout(delay) 使用相對秒數且支援重排;timeout_at(when) 直接使用絕對單調時間,適合跨多層傳遞 deadline。wait_for(aw, timeout) 針對單一 awaitable,逾時會取消該 awaitable,且可能等待其完成取消,組合多個呼叫時更容易形成預算漂移。

第七步:處理巢狀和過期狀態

逾時上下文可以巢狀,內層應使用不晚於外層的 deadline。退出後透過 cm.expired() 判斷是否真的到期;普通例外、外部取消和業務返回不應被誤報為 timeout。

第八步:驗證競態與清理

測試中繼資料延遲、deadline 已過、內層先逾時、外部取消與逾時同時發生、下游忽略取消、reschedule(None)、重複重排和清理失敗。記錄 deadline、剩餘預算、取消原因和下游耗時,不記錄敏感請求內容。

高品質示範回答

「我用 loop.time() 產生絕對 deadline,並把它傳給所有下游。預算未知時先建立 timeout(None),中繼資料到達後 reschedule;內部取消由上下文在退出時轉成外部 TimeoutError,所以只在上下文外捕獲。外部 CancelledError 保持傳播。巢狀呼叫共享最早 deadline,客戶端和資料庫也接收剩餘預算。透過競態測試確認沒有任務和資源洩漏。」

常見錯誤

  • time.time() 計算 deadline → 校時導致預算跳變 → 使用 loop.time()
  • 在 timeout 上下文內捕獲 TimeoutError → 捕獲不到正確例外 → 在上下文外處理。
  • 把外部取消當成逾時 → 請求取消被錯誤降級 → 區分 CancelledError 與 TimeoutError。
  • 每層重新給完整 timeout → 總延遲超過契約 → 傳遞絕對 deadline。
  • 忽略 wait_for 的取消等待 → 實際耗時超過數字 → 評估取消收斂和驅動行為。
  • 只測成功和單一逾時 → 競態資源洩漏 → 覆蓋同時取消、重排和清理失敗。

追問及應對

追問一:為什麼 timeout_at 比逐層 timeout 穩定?

所有層都對同一個絕對時刻負責,不會因每層重新計算相對秒數而累積誤差或超支。

追問二:reschedule 到過去的 deadline 會怎樣?

上下文會在事件迴圈下一次機會觸發逾時;呼叫方應在重排前檢查剩餘預算,避免繼續發起不可取消的 I/O。

追問三:如何讓資料庫也遵守 deadline?

將剩餘秒數傳給驅動的 statement timeout 或取消 API;Python task 被取消不會自動停止伺服器端查詢。

追問四:外部取消與 timeout 同時到達要記錄哪個?

保留取消計數和父任務原因,優先向上傳播外部取消;把上下文是否 expired 作為診斷欄位,不要把所有情況歸為業務逾時。

追問五:何時使用 wait_for?

單一 awaitable 需要簡單相對逾時時可用;跨多個子呼叫、需要動態截止時間或共享預算時,優先使用 timeout/timeout_at。

公開來源

同類題目

相關面試工具

用 Screenshot 處理演算法題

截圖題目後,依序看約束、解法、程式碼、邊界條件和複雜度。

查看工具