科技 · 實用指南
在 Jupyter 或 FastAPI 中 asyncio.run() 失敗?修正事件迴圈擁有者,而非掩蓋症狀
這個例外在提示擁有權衝突:目前執行緒已有元件執行事件迴圈。應在該擁有者內 await,或只在真正的同步入口建立一個迴圈。

Python 會在同一執行緒啟動第二個事件迴圈前拋出此錯誤。這是刻意的保護:asyncio.run() 負責迴圈建立、執行、非同步產生器清理、執行器關閉與迴圈關閉;Notebook 核心或 Web 伺服器已承擔這些職責。
安全修復取決於呼叫者。Notebook 儲存格或非同步路由應直接 await 協程;獨立腳本只在程序邊界保留一個 asyncio.run();可重用程式庫應提供非同步 API,並讓同步包裝器在已有迴圈時明確失敗。
修改前先確認迴圈擁有者
記錄執行緒、asyncio.get_running_loop() 的結果和宿主。Notebook 儲存格、非同步處理器、生命週期鉤子與 CLI 入口的契約不同。
不要對未建立的迴圈呼叫 stop()、close()、run_until_complete() 或 asyncio.run();錯誤通常表示邊界放錯。
在 Notebook 與非同步框架內 await
Jupyter 中把 result = asyncio.run(work()) 改成 result = await work()。FastAPI 等框架應由非同步呼叫者在宿主迴圈中 await。
長期 Task 要綁定宿主生命週期並定義啟動、取消與關閉,不能為消除例外而放任背景工作。
獨立腳本只保留一個頂層擁有者
真正入口可保留 asyncio.run(main());需要共用迴圈和內容時,可由一個 asyncio.Runner 依序執行。
不要在輔助函式中散布 runner,內部實作應保持可 await。
分離程式庫的非同步與同步契約
主要操作實作為 async def fetch_async(...)。同步 fetch(...) 只有在目前執行緒無活動迴圈時才呼叫 asyncio.run(),否則提示 await 非同步 API。
在建立協程前檢查迴圈;若已建立未使用協程,應明確關閉,避免第二個警告掩蓋擁有權錯誤。
只在真實執行緒邊界跨執行緒提交
專用迴圈位於另一 OS 執行緒時,asyncio.run_coroutine_threadsafe(coro, loop) 可回傳 concurrent Future;逾時、取消與關閉仍須定義。
在迴圈執行緒裡阻塞 future.result() 可能死結,執行緒身分與迴圈生命週期必須寫入設計。
把迴圈補丁限制為相容層
舊工具若必須用巢狀補丁,應限制範圍和版本並單獨測試,不能成為程式庫的程序級預設行為。
回傳值成功不代表迴圈健康;內容錯誤、待處理工作或後續儲存格失敗仍可能存在。
用測試證明邊界已修復
在活動迴圈中測試非同步 API,在一般執行緒測試同步包裝器;驗證回傳值、例外、取消、contextvars、重複呼叫和 asyncio.all_tasks()。
負面測試應確保活動迴圈內呼叫同步包裝器立即失敗,並明確提示 await 非同步 API。
