Python 非同步程式設計探索之旅

第一站：兩種起點 - Threading vs. Asyncio

我們的旅程始於兩個處理併發的經典模型。它們的目標相似，但運作方式卻截然不同。

Thread (執行緒) - 搶佔式多工

由作業系統 (OS) 決定何時切換任務，就像一位專案經理，隨時可能中斷一個員工（執行緒），讓另一個員工接手。程式本身無法控制切換時機。

Asyncio (協程) - 合作式多工

由程式自己透過 await 關鍵字主動放棄控制權。就像一位自律的員工，在等待時會主動去做別的事，效率極高。

第二站：GIL 的高牆 - 為何 Thread 無法真正並行？

我們很快就遇到了一堵名為 GIL (全域直譯器鎖) 的高牆。它規定了在任一時刻，只有一個執行緒能執行 Python 位元組碼，這也引出了兩個關鍵概念的區別：

併發 (Concurrency): 看起來像同時執行。一個處理器在多個任務間快速切換。適用於 I/O 密集型任務（如網路請求），`threading` 和 `asyncio` 都能實現。
並行 (Parallelism): 真正意義上的同時執行。需要多個 CPU 核心。由於 GIL 的存在，`threading` 無法實現，必須使用 `multiprocessing`。

核心結論：在 Python 中，絕對不要用多執行緒來加速 CPU 密集型任務！

第三站：控制流量 - 如何限制併發數量？

當我們想發起大量網路請求時，直接全部啟動會造成問題。我們發現 `threading` 的介面非常簡潔，而 `asyncio` 則需要一個關鍵工具：Semaphore。

# ThreadPoolExecutor 的簡潔寫法
with ThreadPoolExecutor(max_workers=4) as executor:
    futures = [executor.submit(task, arg) for arg in args]

# Asyncio 的標準做法，使用 Semaphore
async def worker(arg, semaphore):
    async with semaphore:
        # ... 執行請求 ...
        await asyncio.sleep(1)

semaphore = asyncio.Semaphore(4)
tasks = [worker(arg, semaphore) for arg in args]
await asyncio.gather(*tasks)

雖然 `asyncio` 的寫法看起來較為複雜，但我們發現可以透過**自己封裝輔助函式**或使用**第三方函式庫**（如 `aiometer`）來達到同樣的簡潔效果，同時保有其高效能與高擴展性的優勢。

第四站：走向通用 - `anyio` 的崛起

在深入研究 `aiometer` 的原始碼時，我們發現它並未使用 `asyncio`，而是用了 `anyio`。這引導我們了解了 Python 非同步生態的一個重要趨勢。

為何使用 `anyio`？

`anyio` 是一個非同步轉接層，目的是解決 `asyncio` 和另一個流行框架 `Trio` 之間的 API 不相容問題。它就像一個萬用轉接頭，讓函式庫作者可以撰寫一套程式碼，同時在兩種後端上運行。

`anyio` 會在**執行時期**動態偵測當前的事件迴圈是 `asyncio` 還是 `Trio`，然後在內部呼叫對應的原生函式。這是一個非常聰明的函式庫設計決策，旨在提升程式碼的相容性與可攜性。

終點站：我能同時使用兩者嗎？

最後，我們探討了一個終極問題：既然有 `asyncio` 和 `Trio`，我們能否在一個程式中同時使用它們？

答案是：不行，至少不應該在同一個執行緒中這樣做。

事件迴圈是其所在執行緒的唯一「主宰者」。在同一個執行緒中運行兩個事件迴圈會導致混亂。正確的做法是：

為你的專案選擇一個主要的非同步框架。
優先使用原生支援該框架的函式庫。
若你身為函式庫作者，請使用 `anyio` 來確保你的程式碼能被更多人使用。