iT邦幫忙

2025 iThome 鐵人賽

DAY 16
0

在前 15 天,我們已經打好了「能跑、能測、能控管」的工程基礎:環境一致性、依賴管理、測試藍圖、日誌、錯誤處理、序列化格式…專案算是能穩穩上線了。

但實務開發裡,還有一個經常被忽略的維度:效能與併發。尤其是 I/O 密集的應用(API 服務、資料抓取、排程任務),如果還在用傳統同步方式,效能會被白白綁死。

今天,我們就來談 Python 世界的非同步基礎,從 asyncioanyio,帶你理解為什麼「寫得動」和「寫得對」差這麼遠。


為什麼要非同步?

先把誤解拆掉:

  • 非同步 不是 多核心並行(那是 multiprocessing)。
  • 非同步 不是 平行線程(那是 threading)。

它的目標很單純:用一個執行緒,切換不同 I/O 任務,讓 CPU 不要傻等。

典型場景:

  • 呼叫外部 API,要等 200ms 回應。
  • 查詢資料庫,I/O 卡了 300ms。

如果同步執行,這 500ms CPU 幾乎閒著發呆。用非同步,這段時間可以讓別的請求先跑,效能直接翻倍甚至數倍。


asyncio:官方解法

asyncio 是 Python 3.5+ 就內建的非同步框架,也是大多數現代框架(FastAPI、httpx)的基礎。

最小範例:

import asyncio
import httpx

async def fetch(url: str) -> str:
    async with httpx.AsyncClient() as client:
        resp = await client.get(url)
        return resp.text[:50]

async def main():
    urls = ["https://example.com"] * 5
    tasks = [fetch(u) for u in urls]
    results = await asyncio.gather(*tasks)
    print(results)

asyncio.run(main())

關鍵心法:

  1. async def 宣告非同步函式。
  2. await 暫停等待,不會卡住 event loop。
  3. gather 一次跑多個 coroutine。

👉 適合場景:高併發 I/O(API gateway、爬蟲、聊天伺服器)。


anyio:抽象層 + 心靈救贖

雖然 asyncio 是官方標配,但它的 API 偏低階,寫起來容易踩坑。這時候就輪到 anyio 出場。

anyio 提供一層抽象,能同時支援 asyncio / trio,並加上更友善的 API,讓開發者少掉許多 boilerplate。

範例:並行執行多個 task

import anyio
import httpx

async def fetch(url: str) -> str:
    async with httpx.AsyncClient() as client:
        resp = await client.get(url)
        return resp.text[:30]

async def main():
    async with anyio.create_task_group() as tg:
        for i in range(5):
            tg.start_soon(fetch, "https://example.com")

anyio.run(main)

心法:

  • TaskGroup:用來管理並行任務,類似「非同步版 with context」。
  • 自動錯誤傳播:子任務錯誤會回傳,不會默默吞掉。
  • 跨生態:FastAPI、Starlette、Trio 都能用,降低綁定。

👉 適合場景:需要乾淨錯誤處理、複雜任務調度的專案。


async/await 的工程化注意事項

  1. 不要混雜同步阻塞

    在 async 函式裡呼叫同步的 requests.get(),等於整個 loop 都被卡死。解法是換成非同步版(httpx、aiosqlite 等),或把同步呼叫丟進 anyio.to_thread.run_sync

  2. 測試要 async-friendly

    pytest 提供 pytest-asyncio plugin,讓你可以直接寫:

    @pytest.mark.asyncio
    async def test_fetch():
        result = await fetch("https://example.com")
        assert "Example" in result
    
    
  3. 結合 Nox / Hatch

    在 Day 8 建立的一鍵化工作流,可以加一個 async 測試環境:

    [tool.hatch.envs.async]
    dependencies = ["pytest", "pytest-asyncio", "httpx", "anyio"]
    [tool.hatch.envs.async.scripts]
    test = "pytest -q tests/async"
    
    
  4. 例外處理要設計

    延續 Day 14 的心法,非同步程式更容易遇到外部失敗(timeout、連線斷掉)。記得搭配 tenacity 或 TaskGroup 的錯誤傳播機制,避免整個系統直接崩。


專案整合:從同步 → 非同步

延續 Day 4 的目錄結構,可以加一個 async 模組:

my_project/
├─ src/my_project/
│   ├─ __init__.py
│   ├─ core.py
│   ├─ async_ops.py      # 放置非同步 I/O
│   └─ adapters/
│       └─ web.py        # FastAPI / httpx 用非同步
└─ tests/
    └─ async/
        └─ test_async_ops.py

這樣就能把「效能關鍵」的 I/O 操作集中管理,未來要擴展也比較安全。


結語

非同步不是銀彈,但在 I/O 密集場景,它就是讓 Python 撐住高併發的核心解法。

  • asyncio:官方內建,足以應付 80% 的場景。
  • anyio:更乾淨的抽象,幫你少掉許多坑。

工程化的 Python,不只要「能跑」,還要「跑得有效率」。把非同步納進工具箱,才算真的踏進現代後端的世界 🚀。

明天 Day 17,我們會進一步探討效能觀測:cProfilepy-spyline-profiler,讓你不只是盲目加 async,而是能用數據說話。


上一篇
Day 15 -序列化與設定格式:orjson、YAML、TOML 實務
系列文
30 天 Python 專案工坊:環境、結構、測試到部署全打通16
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言