使用 asyncio 開發¶

非同步程式設計 (asynchronous programming) 與傳統的"順序"程式設計 (sequential programming) 不同。

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

除錯模式¶

在預設情況下 asyncio 以正式生產模式 (production mode) 執行。為了讓開發更輕鬆，asyncio 還有一種除錯模式 (debug mode)。

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

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

除了啟用除錯模式外，還要考慮：

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

啟用除錯模式時：

asyncio 會檢查未被等待的協程並記錄他們；這會減輕"被遺忘的等待 (forgotten await)" 問題。
許多非執行緒安全 (non-threadsafe) 的 asyncio APIs（例如 loop.call_soon() 和 loop.call_at() 方法），如果從錯誤的執行緒呼叫就會引發例外。
如果執行一個 I/O 操作花費的時間太長，則將 I/O 選擇器 (selector) 的執行時間記錄到日誌中。
執行時間超過 100 毫秒的回呼 (callback) 將會被記錄於日誌。屬性 loop.slow_callback_duration 可用於設定以秒為單位的最小執行持續時間，超過這個值執行時間就會被視為"緩慢"。

並行性和多執行緒 (Concurrency and Multithreading)¶

事件迴圈在執行緒中運行（通常是主執行緒），並在其執行緒中執行所有回呼和 Tasks（任務）。當一個 Task 在事件迴圈中運行時，沒有其他 Task 可以在同一個執行緒中運行。當一個 Task 執行一個 await 運算式時，正在執行的 Task 會被暫停，而事件迴圈會執行下一個 Task。

要從不同的 OS 執行緒為一個 callback 排程，應該使用 loop.call_soon_threadsafe() 方法。例如：

loop.call_soon_threadsafe(callback, *args)

幾乎所有 asyncio 物件都不支援執行緒安全 (thread safe)，這通常不是問題，除非在 Task 或回呼函式之外有程式需要和它們一起運作。如果需要這樣的程式來呼叫低階 asyncio API，應該使用 loop.call_soon_threadsafe() 方法，例如：

loop.call_soon_threadsafe(fut.cancel)

要從不同的 OS 執行緒為一個協程物件排程，應該使用 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()

為了能夠處理訊號，事件迴圈必須於主執行緒中運行。

The loop.run_in_executor() method can be used with a concurrent.futures.ThreadPoolExecutor or InterpreterPoolExecutor to execute blocking code in a different OS thread without blocking the OS thread that the event loop runs in.

目前沒有什麼辦法能直接從另一個行程（例如透過 multiprocessing 啟動的程序）來為協程或回呼排程。事件迴圈方法小節列出了可以從 pipes（管道）讀取並監視 file descriptor（檔案描述器）而不會阻塞事件迴圈的 API。此外，asyncio 的子行程 API 提供了一種啟動行程並從事件迴圈與其通訊的辦法。最後，之前提到的 loop.run_in_executor() 方法也可和 concurrent.futures.ProcessPoolExecutor 搭配使用，以在另一個行程中執行程式。

執行阻塞的程式¶

不應該直接呼叫阻塞（CPU 密集型）程式。例如一個執行 1 秒 CPU 密集型計算的函式，那麼所有並行非同步 Tasks 和 IO 操作都會被延遲 1 秒。

An executor can be used to run a task in a different thread, including in a different interpreter, or even in a different process to avoid blocking the OS thread with the event loop. See the loop.run_in_executor() method for more details.

日誌記錄¶

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

日誌級別被預設為 logging.INFO，它可以很容易地被調整：

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

網路日誌記錄可能會阻塞事件迴圈。建議使用獨立的執行緒來處理日誌或使用非阻塞 IO，範例請參見 Dealing with handlers that block。

偵測從未被等待的 (never-awaited) 協程¶

當協程函式被呼叫而不是被等待時（即執行 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()

偵測從未被獲取的 (never-retrieved) 例外¶

如果呼叫 Future.set_exception()，但 Future 物件從未被等待，例外將無法被傳播 (propagate) 到使用者程式。在這種情況下，當 Future 物件被垃圾回收 (garbage collected) 時，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

啟用除錯模式以取得任務建立處的追蹤資訊 (traceback)：

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