使用 asyncio 進行開發¶

非同步程式設計與經典的“順序”程式設計不同。

本頁列出了常見的錯誤和陷阱，並解釋瞭如何避免它們。

除錯模式¶

預設情況下，asyncio 以生產模式執行。為了方便開發，asyncio 具有一個 除錯模式。

有幾種方法可以啟用 asyncio 除錯模式

將 PYTHONASYNCIODEBUG 環境變數設定為 1。
使用 Python 開發模式。
將 debug=True 傳遞給 asyncio.run()。
呼叫 loop.set_debug()。

除了啟用除錯模式，還請考慮

將 asyncio 記錄器的日誌級別設定為 logging.DEBUG，例如，以下程式碼片段可以在應用程式啟動時執行
```
logging.basicConfig(level=logging.DEBUG)
```
配置 warnings 模組以顯示 ResourceWarning 警告。一種方法是使用 -W default 命令列選項。

當除錯模式啟用時

許多非執行緒安全的 asyncio API（例如 loop.call_soon() 和 loop.call_at() 方法）如果在錯誤的執行緒中呼叫，則會引發異常。
如果 I/O 選擇器執行 I/O 操作花費太長時間，則會記錄其執行時間。
執行時間超過 100 毫秒的回撥將被記錄。 loop.slow_callback_duration 屬性可用於設定被認為是“慢”的最小執行持續時間（以秒為單位）。

併發和多執行緒¶

事件迴圈在一個執行緒中執行（通常是主執行緒），並在其執行緒中執行所有回撥和任務。當一個任務在事件迴圈中執行時，同一執行緒中的其他任務不能執行。當一個任務執行 await 表示式時，正在執行的任務被掛起，事件迴圈執行下一個任務。

要從另一個作業系統執行緒排程回撥，應使用 loop.call_soon_threadsafe() 方法。示例

loop.call_soon_threadsafe(callback, *args)

幾乎所有的 asyncio 物件都不是執行緒安全的，這通常不是問題，除非有程式碼從任務或回撥之外使用它們。如果需要此類程式碼呼叫低階 asyncio API，則應使用 loop.call_soon_threadsafe() 方法，例如

loop.call_soon_threadsafe(fut.cancel)

要從不同的作業系統執行緒排程協程物件，應使用 run_coroutine_threadsafe() 函式。它返回一個 concurrent.futures.Future 來訪問結果

async def coro_func():
     return await asyncio.sleep(1, 42)

# Later in another OS thread:

future = asyncio.run_coroutine_threadsafe(coro_func(), loop)
# Wait for the result:
result = future.result()

為了處理訊號，事件迴圈必須在主執行緒中執行。

可以使用 loop.run_in_executor() 方法與 concurrent.futures.ThreadPoolExecutor 或 InterpreterPoolExecutor 結合使用，以在不同的作業系統執行緒中執行阻塞程式碼，而不會阻塞事件迴圈執行的作業系統執行緒。

目前無法直接從不同的程序（例如使用 multiprocessing 啟動的程序）排程協程或回撥。事件迴圈方法部分列出了可以在不阻塞事件迴圈的情況下從管道讀取和監視檔案描述符的 API。此外，asyncio 的子程序 API 提供了一種從事件迴圈啟動程序並與之通訊的方法。最後，上述 loop.run_in_executor() 方法也可以與 concurrent.futures.ProcessPoolExecutor 結合使用，以在不同的程序中執行程式碼。

執行阻塞程式碼¶

阻塞（CPU密集型）程式碼不應直接呼叫。例如，如果一個函式執行 1 秒的 CPU密集型計算，所有併發的 asyncio 任務和 IO 操作都將延遲 1 秒。

執行器可用於在不同的執行緒中執行任務，包括在不同的直譯器中，甚至在不同的程序中，以避免阻塞事件迴圈所在的作業系統執行緒。有關更多詳細資訊，請參閱 loop.run_in_executor() 方法。

日誌記錄¶

asyncio 使用 logging 模組，所有日誌記錄都透過 "asyncio" 記錄器執行。

預設日誌級別為 logging.INFO，可以輕鬆調整

logging.getLogger("asyncio").setLevel(logging.WARNING)

網路日誌記錄可能會阻塞事件迴圈。建議使用單獨的執行緒處理日誌或使用非阻塞 IO。例如，請參閱處理阻塞處理程式。

檢測從未被等待的協程¶

當呼叫協程函式但未等待時（例如 coro() 而不是 await coro()），或者協程未透過 asyncio.create_task() 排程時，asyncio 將發出 RuntimeWarning

import asyncio

async def test():
    print("never scheduled")

async def main():
    test()

asyncio.run(main())

輸出

test.py:7: RuntimeWarning: coroutine 'test' was never awaited
  test()

除錯模式下的輸出

test.py:7: RuntimeWarning: coroutine 'test' was never awaited
Coroutine created at (most recent call last)
  File "../t.py", line 9, in <module>
    asyncio.run(main(), debug=True)

  < .. >

  File "../t.py", line 7, in main
    test()
  test()

通常的解決方法是等待協程或呼叫 asyncio.create_task() 函式

async def main():
    await test()

檢測從未被檢索的異常¶

如果呼叫了 Future.set_exception() 但從未等待 Future 物件，則該異常將永遠不會傳播到使用者程式碼。在這種情況下，當 Future 物件被垃圾回收時，asyncio 將發出一條日誌訊息。

未處理異常的示例

import asyncio

async def bug():
    raise Exception("not consumed")

async def main():
    asyncio.create_task(bug())

asyncio.run(main())

輸出

Task exception was never retrieved
future: <Task finished coro=<bug() done, defined at test.py:3>
  exception=Exception('not consumed')>

Traceback (most recent call last):
  File "test.py", line 4, in bug
    raise Exception("not consumed")
Exception: not consumed

啟用除錯模式以獲取任務建立位置的追溯

asyncio.run(main(), debug=True)