Python 非同步 vs 執行緒：何時使用 asyncio

為什麼並行在 Python 中很重要

你有一個從多個 API 獲取資料的 Python 腳本。它能運作，但很慢——每個請求都要等待前一個完成。你聽過 asyncio 和執行緒，但該用哪一個？選擇錯誤的模型可能導致程式碼不必要地複雜或極度緩慢。

本指南將釐清困惑。我們會比較 Python 的非同步和執行緒模型，確切說明何時該使用哪一種，並提供你可以套用到自己專案的實用範例。

了解 Python 的並行環境

Python 提供三種主要的並行模型：

• 執行緒 (Threads) – 在單一程序內有多個執行緒，由作業系統管理。對於 CPU 密集工作會受到全域解釋器鎖 (GIL) 的限制。
• 非同步 (asyncio) – 單執行緒事件迴圈，在 await 點之間切換任務。非常適合 I/O 密集操作。
• 多程序 (Multiprocessing) – 獨立的程序，每個都有自己的 Python 解釋器和記憶體空間。繞過 GIL 以實現真正的平行處理。

GIL 是一個互斥鎖，防止多個執行緒同時執行 Python 位元組碼。這表示執行緒無法加速 CPU 密集任務，但確實有助於 I/O 密集任務，因為在 I/O 操作期間 GIL 會被釋放。

何時使用 asyncio

當你的程式大部分時間都在等待外部事件（網路請求、檔案 I/O、資料庫查詢或計時器）時，asyncio 表現出色。因為它使用單一執行緒，你可以避免執行緒建立和上下文切換的開銷。事件迴圈能有效管理數千個並行連線。

在以下情況使用 asyncio：

• 你正在建立處理許多同時連線的網頁伺服器或客戶端（例如 FastAPI、aiohttp）。
• 你的任務是 I/O 密集，且需要高並行（數百或數千個任務）。
• 你想要對任務排程和取消進行細粒度控制。
• 你已經在使用非同步框架或函式庫。

以下是一個簡單範例，同時獲取多個 URL：

import asyncio
import aiohttp

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    urls = [
        'https://example.com',
        'https://example.org',
        'https://example.net',
    ]
    tasks = [fetch(url) for url in urls]
    results = await asyncio.gather(*tasks)
    print(f'Fetched {len(results)} pages')

asyncio.run(main())

何時使用執行緒

當你有阻塞式 I/O 操作且沒有非同步對應版本時，或者當你使用不支援非同步的舊程式碼時，執行緒是個好選擇。對於少量任務，它們也更容易理解。

在以下情況使用執行緒：

• 你需要平行執行阻塞式 I/O，例如讀取多個檔案或進行同步 API 呼叫。
• 你正在整合不支援非同步的函式庫。
• 你有適量的並行任務（數十個，而非數千個）。
• 你想保持程式碼同步但仍重疊 I/O。

使用 concurrent.futures.ThreadPoolExecutor 的範例：

from concurrent.futures import ThreadPoolExecutor
import requests

def fetch(url):
    response = requests.get(url)
    return response.text

urls = [
    'https://example.com',
    'https://example.org',
    'https://example.net',
]

with ThreadPoolExecutor(max_workers=10) as executor:
    results = list(executor.map(fetch, urls))

print(f'Fetched {len(results)} pages')

比較 asyncio 和執行緒

以下是並排比較，幫助你決定：

那 CPU 密集任務呢？

由於 GIL，asyncio 和執行緒都無法幫助你在 Python 中平行處理 CPU 密集工作。對於 CPU 密集任務——例如數值計算、影像處理或資料壓縮——請使用 multiprocessing 或 concurrent.futures.ProcessPoolExecutor。

範例：

from concurrent.futures import ProcessPoolExecutor

def cpu_heavy(n):
    return sum(i * i for i in range(n))

if __name__ == '__main__':
    with ProcessPoolExecutor() as executor:
        results = list(executor.map(cpu_heavy, [10**6, 10**6, 10**6]))
    print(results)

最佳實踐和陷阱

• 不要將阻塞呼叫與 asyncio 混合使用。 如果你在非同步任務內呼叫阻塞函式，它將阻塞整個事件迴圈。使用 asyncio.to_thread() 在執行緒池中執行阻塞程式碼。
• 限制執行緒池大小。 建立太多執行緒會導致上下文切換開銷。使用 ThreadPoolExecutor(max_workers=N)。
• 正確處理例外。 在 asyncio 中，任務中未處理的例外可能被忽略。使用 asyncio.gather(..., return_exceptions=True) 或新增回呼。
• 新的 I/O 密集專案優先使用 asyncio。 生態系統成熟，且 FastAPI 和 aiohttp 等框架讓它變得容易。
• 舊有或簡單腳本使用執行緒。 如果你只需要平行處理幾個阻塞呼叫，執行緒侵入性較小。

常見問題

我可以同時使用 asyncio 和執行緒嗎？

可以。你可以使用 asyncio.to_thread() 或 loop.run_in_executor() 在執行緒池中執行阻塞程式碼。當你需要從非同步程式碼呼叫同步函式庫而不阻塞事件迴圈時，這很有用。

asyncio 能與 GIL 一起運作嗎？

是的，asyncio 在單一執行緒中運行，受 GIL 約束。然而，因為它是為 I/O 密集任務設計的，在 I/O 操作期間 GIL 會被釋放，允許其他任務運行。對於 CPU 密集工作，asyncio 不會提供加速。

哪個更快：asyncio 還是執行緒？

對於有許多並行操作的 I/O 密集任務，asyncio 通常更快且更具可擴展性，因為它避免了執行緒開銷。對於少量阻塞任務，執行緒可能更簡單且效能相似。兩者都無法幫助 CPU 密集任務。

做出正確選擇

首先確定你的瓶頸是 I/O 還是 CPU。如果是 I/O 且需要高並行，請使用 asyncio。如果是 I/O 但處理阻塞函式庫或較簡單的腳本，執行緒是可靠的選擇。對於 CPU 密集工作，請使用多程序。

記住，你通常可以結合這些模型——例如，使用 asyncio 進行網路操作，並使用執行緒池進行檔案 I/O。關鍵是了解權衡並選擇適合你特定問題的工具。

當你需要快速格式化或驗證並行 API 呼叫返回的 JSON 資料時，試試我們的 JSON Formatter，輕鬆美化和除錯。