事件循環是python異步編程的核心機制，負責調度和運行協程。1. asyncio.run() 是啟動事件循環的推薦方式，適用于大多數情況；2. 在需手動獲取事件循環時，應優先使用 asyncio.get_running_loop()；3. 事件循環通過“就緒隊列”管理任務，在 await 遇到 i/o 等待時切換任務以實現并發；4. 使用 create_task() 將協程封裝為任務提交給事件循環執行；5. 避免阻塞主線程，可用 loop.run_in_executor() 處理同步阻塞或cpu密集型任務；6. 多線程中需為每個線程綁定獨立事件循環。正確理解并應用這些機制，有助于編寫高效穩定的異步程序。
python 的異步編程在處理高并發、I/O 密集型任務時非常有用，而 asyncio 是 Python 官方提供的異步框架。理解它的事件循環機制是掌握異步編程的關鍵。
什么是事件循環？
事件循環（Event Loop）是 asyncio 的核心，它負責調度和運行協程。你可以把它想象成一個無限循環，不斷檢查是否有任務需要執行，并在合適的時候調度它們。
在 Python 3.7+ 中，通常使用 asyncio.run() 來啟動主函數，它會自動創建并管理事件循環。但如果你需要更細粒度的控制，比如在 jupyter 或某些嵌入式環境中，可能需要手動獲取或創建事件循環。
立即學習“Python免費學習筆記（深入）”；
例如：
import asyncio async def main(): print("Hello") await asyncio.sleep(1) print("World") asyncio.run(main())
在這個例子中，asyncio.run() 啟動了事件循環，然后把 main() 協程加入其中，事件循環會負責調度它的執行。
如何正確獲取和使用事件循環？
在某些情況下，你可能需要手動獲取當前的事件循環。比如，在 Jupyter Notebook 中直接使用 asyncio.run() 可能會報錯，因為事件循環已經存在。
常見的做法是使用以下方式之一來獲取事件循環：
- 在主線程中：loop = asyncio.get_event_loop()
- 在子線程中：loop = asyncio.new_event_loop()
- 推薦（適用于大多數情況）：loop = asyncio.get_running_loop()
需要注意的是：
- get_event_loop() 已逐漸不推薦使用，尤其是在 Python 3.9+ 中。
- get_running_loop() 更安全，但如果不在事件循環內部調用會拋出異常。
事件循環是如何調度任務的？
事件循環并不是一上來就把所有協程都跑完，而是通過“事件”驅動的方式進行調度。
舉個簡單的例子：
async def task1(): print("Task1 started") await asyncio.sleep(2) print("Task1 done") async def task2(): print("Task2 started") await asyncio.sleep(1) print("Task2 done") async def main(): t1 = asyncio.create_task(task1()) t2 = asyncio.create_task(task2()) await t1 await t2 asyncio.run(main())
在這個例子中，task1 和 task2 被同時提交給事件循環。當其中一個遇到 await asyncio.sleep() 時，事件循環會切換到另一個任務繼續執行，從而實現并發效果。
關鍵點在于：
- 使用 create_task() 把協程封裝成任務并交給事件循環。
- await 某個任務表示等待它完成，但期間可以調度其他任務。
- 事件循環內部維護了一個“就緒隊列”，每次從隊列中取出任務執行。
常見問題與注意事項
不要混用不同的事件循環庫
比如，不要在同一程序中混用 asyncio 和 tornado 的事件循環，除非你清楚自己在做什么。否則容易導致死鎖或調度混亂。
避免阻塞主線程
如果你在事件循環中執行同步阻塞操作（如長時間計算、普通 time.sleep()），整個異步流程都會被卡住。解決辦法是：
- 使用 await asyncio.sleep() 替代 time.sleep()
- 對于 CPU 密集任務，考慮用 loop.run_in_executor() 放到線程或進程中執行
def blocking_function(): time.sleep(5) return "Done" async def main(): loop = asyncio.get_running_loop() result = await loop.run_in_executor(None, blocking_function) print(result) asyncio.run(main())
注意多線程中的事件循環
如果你在一個非主線程中使用事件循環，記得先為該線程綁定一個新的事件循環：
import threading def thread_main(): loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) loop.run_until_complete(main()) threading.Thread(target=thread_main).start()
否則可能會出現找不到事件循環的問題。
基本上就這些。事件循環雖然看起來復雜，但只要理解它是如何調度協程、如何避免阻塞，以及在不同環境下如何正確使用，就能比較順利地寫出高效的異步代碼了。
