Python asyncio：從生活直覺一路接到 Event Loop 與 OS 實作

1. 全篇閱讀架構：先聽懂，再接上工程術語

只用比喻會產生一個問題：你覺得自己「好像懂了」，但看到文件裡的 Task、Future、selector、callback、event-loop implementation 時仍然接不上。反過來，只堆工程名詞又會讓初學者不知道每個元件到底在解決什麼。

1先建立四個概念所在的層級

Coroutine、Task、Thread、Process 不是同一層的四個替代選項。 它們分別回答不同問題：

2從比喻正式轉成工程模型

3先看整體資料流

2. Runtime 到底是什麼？它不是單一元件，而是「執行中的支援環境」

1生活直覺：程式碼像劇本，runtime 像整套演出系統

一份劇本只描述角色要說什麼、何時進場，但劇本自己不會演出。 真正演出時還需要舞台、演員、燈光、道具、場務、時間控制與錯誤處理。

把它換成程式：

2先分清楚：language、implementation、runtime

Compile time 與 runtime 的差別

Python複製# 這段程式可以先被讀取、解析並編譯成 bytecode
def divide(a, b):
    return a / b
# 直到程式真的執行到這一行，才在 runtime 發生 ZeroDivisionError
divide(10, 0)

Compile time 關注把原始碼解析、檢查並轉成可執行形式； runtime 關注程式真正跑起來後發生的 function calls、物件建立、I/O、例外、排程與資源管理。

3CPython runtime 在這張架構圖的哪裡？

它不是夾在 Thread 與 event loop 中間的一個盒子，而是遍布整個 Python Process 的執行環境。 Process 提供 OS 隔離；CPython runtime 在該 Process 裡管理 Python 世界； Threads 再進入某個 interpreter context 執行 Python code。

CPython runtime 主要提供的能力包括：

• 執行 Python bytecode 與 function calls。
• 建立、保存與操作 Python objects。
• 管理 call frames、tracebacks 與 exceptions。
• 管理記憶體，例如 reference counting 與 garbage collection。
• 維護 modules、imports、builtins 與 interpreter state。
• 協調 Thread 進入 Python interpreter 的執行狀態；具體行為取決於 CPython build 與版本。

4為什麼有時又聽到 asyncio runtime？

「asyncio runtime」通常是工程師使用的方便總稱，不是一個獨立的 OS 元件。 它指的是 asyncio 程式執行時一起合作的機制，例如：

這些東西仍然是由 CPython runtime 執行的 Python／native objects，event loop 也仍然跑在某一條 Thread 上。 因此兩者不是競爭關係：

asyncio、event loop 與 runtime 的關係

5ASGI server runtime 又是什麼？

當你執行 Uvicorn 時，還會多一層 server runtime。它負責把網路世界與 FastAPI application 接起來：

1. 建立 server Process、event loop 與 listening socket。
2. 接受 TCP connections，解析 HTTP／WebSocket protocol。
3. 依照 ASGI 規格建立 scope、receive、send。
4. 把 FastAPI application 作為 async callable 呼叫並等待。
5. 處理 startup／shutdown lifespan、signals、connections 與 worker lifecycle。

圖中的層不是單純「上層只能呼叫下一層」的硬切割，而是責任分工。 例如 Uvicorn 使用 event loop，event loop 呼叫 OS I/O，FastAPI coroutine 則由 Task 推進。

用一條啟動命令走完整個 runtime 鏈

Bash複製uvicorn main:app --loop uvloop --workers 4

1. OS 建立四個 worker Processes。
2. 每個 Process 啟動一份 CPython runtime／interpreter state。
3. CPython 執行 Uvicorn server code。
4. Uvicorn 在每個 worker 選擇 uvloop 作為 event-loop implementation。
5. asyncio 的 Task／Future／awaitable 模型建立在該 loop 上運作。
6. Uvicorn 將 HTTP／WebSocket events 轉成 ASGI messages。
7. FastAPI endpoint coroutine 被 request Task 推進。
8. uvloop／libuv 再透過 OS I/O 機制等待 sockets 與 timers。

6同一句 runtime，在不同上下文可能指不同東西

7最終定位：它和 Process、Thread、Event Loop 的關係

3. Event Loop 完整拆解：它是什麼、做什麼、有哪些實作？

1生活直覺：event loop 像學校總務台

想像總務台同時處理很多事情：冷氣報修、包裹到校、教室鑰匙申請、影印完成通知。 總務人員不會在冷氣維修員抵達前一直站在門口等；他會先記下這件事，去處理其他目前能做的工作。 當電話、廣播或系統通知「維修員到了」，再接續原本的流程。

總務台反覆做的事是：

1. 查看目前有哪些事情已經可以處理。
2. 選一件事情，執行它的下一小段。
3. 若需要等待外部結果，就登記「結果到了再通知我」。
4. 繼續處理其他 ready 的事情。
5. 沒有事情可做時，睡到下一個通知或計時器到期。

2High-level 工程定義

它的主要用途可以拆成四類：

Event loop 不負責什麼？

• 它不會讓純 Python CPU 計算自動變成多核心平行。
• 它不會把任意 blocking function 自動變成 non-blocking。
• 它不是 Task，也不是 Thread；它是執行於 Thread 上的排程與 I/O 協調機制。
• 它不能讓資料庫或第三方 API 本身更早完成，只能更有效率地管理等待。

3一輪 event loop 在做什麼？

以下是適合初學者使用的簡化模型：

1. 搬移到期 timers：把已到時間的 scheduled callbacks 放入 ready queue。
2. 執行 ready work：逐一執行 ready callbacks，或讓 Task 推進 coroutine 一小段。
3. Task 遇到等待：若 coroutine await 的 Future 尚未完成，Task 暫停，並在 Future 上登記喚醒 callback。
4. 計算可睡多久：依最接近的 timer、是否已有 ready work 等條件決定 I/O poll timeout。
5. 等待 OS event：讓 selector／proactor／libuv 等底層機制等待 I/O 或 completion。
6. 接收通知：將完成的 I/O、timer、thread-pool 結果轉成 callbacks，放回 ready queue。
7. 下一輪：只要還有活動中的工作，loop 就繼續重複。

工程上不同 event-loop implementation 的輪次細節並不完全相同；上圖是共通心智模型， 不是要求 application developer 依賴的固定內部 API。

4I/O 等待有哪些底層做法？

「等待網路」不是只有一種實作。常見模型可以先分成 readiness 與 completion 兩大類， 再加上無法直接非同步化時使用的 thread-pool fallback。

5那 asyncio 到底是哪一種？

asyncio 不是單一演算法，也不只是一個 event loop class。 它是 Python 標準函式庫中的非同步程式設計套件，包含：

• 高階使用方式：asyncio.run()、TaskGroup、Queue、Lock、streams。
• 工作抽象：Task、Future、callback、transport／protocol。
• event-loop API 與內建 implementations。
• 與 thread／process executor、subprocess、networking 的整合。

CPython 內建 asyncio 一般選哪一種？

「asyncio 選哪一種」必須連同 Python 版本、作業系統、server 啟動模式與 custom loop 設定一起回答， 不能只說 asyncio 永遠等於 epoll。

6uvloop 是另一種嗎？

因此同一份 application code 通常不必改寫：

Python複製@app.get("/users/{user_id}")
async def get_user(user_id: str):
    return await repository.get_user(user_id)

上層仍使用 asyncio 的 Task、Future 與 awaitable 模型，只是下方換了 loop implementation：

libuv 又是什麼？

libuv 是跨平台 asynchronous I/O library。它提供 event loop、network I/O、timer、process、DNS、file-system 等能力，並在各平台選用合適機制：例如 Linux epoll、macOS/BSD kqueue、Windows IOCP。 對某些無法直接以 non-blocking OS API 處理的操作，libuv 會使用自己的 thread pool。

Uvicorn 如何選擇？

Bash複製# 預設：uvloop 已安裝且平台支援時優先使用，否則退回 asyncio
uvicorn main:app --loop auto

# 明確使用 CPython 內建 asyncio implementation
uvicorn main:app --loop asyncio

# 明確要求 uvloop
uvicorn main:app --loop uvloop

Uvicorn 目前的 --loop auto 會優先選 uvloop，若不可用則使用 Python 內建 asyncio loop。 uvloop 不支援 Windows 或 PyPy；Uvicorn 在 Windows 的具體 built-in loop 選擇還會受到單程序、reload、 multi-worker 模式影響。

uvloop 為什麼可能更快？

它主要降低 event-loop 自己的管理成本：I/O polling、callback dispatch、timer、socket handling、 Future／Task 喚醒等。如果 workload 有大量短小 network events，這些微小成本累積後可能很可觀。

4. FastAPI Worker：一個 Process、主要 Thread、Event Loop 怎麼接起來？

1生活直覺與正式術語

把一個 Uvicorn worker 想成一間獨立分店。分店內有一位主要櫃台人員，使用訂單系統協調很多顧客； 某些舊設備只能讓人站在旁邊等，這時才叫支援人員處理。客流再大時，不是讓同一位櫃台人員長出四雙手， 而是多開幾間分店。

2High-level 架構

3為什麼一條主要 Thread 可以管理很多 request？

因為 request 大部分時間常在等 DB、Redis、第三方 API 或 client socket。 Event loop 會在一個 Task 等待時執行其他 ready Task；等待本身交給 OS I/O 機制管理。

Python複製@app.get("/user")
async def get_user():
    # 等待期間目前 Task 暫停，event loop 可推進其他 Tasks
    return await repository.get_user()

這是 concurrency，不是同一 Thread 的 CPU parallelism。同一瞬間主要 Thread 仍只執行一段 Python code。

4Thread pool 在哪裡出現？

FastAPI 的普通 def path operation 會由外部 thread pool 執行：

Python複製@app.get("/legacy")
def legacy_endpoint():
    return blocking_database.query()

async endpoint 也可以明確 offload blocking I/O：

Python複製@app.get("/legacy-async")
async def legacy_async_endpoint():
    return await asyncio.to_thread(blocking_sdk.fetch)

5多核心與多 workers

Bash複製uvicorn main:app --workers 4 --loop auto

這會建立多個 worker Processes；每個 worker 各自有自己的 Main Thread、event loop、memory、global variables、 DB connection pool 與 HTTP client pool。OS 可以把不同 Processes 排到不同 CPU cores 上。

uvloop 放在這個架構的哪裡？

uvloop 只替換每個 worker 裡的 event-loop implementation，不會改變 worker 數量、Thread 數量或 endpoint 語法。

5. Coroutine：可以暫停並保留現場的工作流程

1生活直覺與術語橋接

普通函式像一段必須一路跑完的關卡；coroutine 像可以存檔的關卡。 它暫停時會保留「走到哪裡」與「身上有哪些資料」，等等待條件完成後再從原處繼續。

2High-level 定義

Python複製async def fetch_user():
    user = await database.get_user()
    return user
coro = fetch_user()
print(coro) # coroutine object，尚未完整執行

3底層關係：誰讓它前進？

Coroutine 本身不是排程器。通常由 Task 包裝並驅動它；Task 讓 coroutine 執行， 直到 coroutine 完成、拋錯，或 await 一個尚未完成的 awaitable。

await x 的正式意思不是「一定切換 Task」，而是「等待一個 awaitable 的結果」。 如果結果已經完成，可能立即繼續；如果尚未完成，目前 coroutine 才會 suspend。

4實際使用與錯誤

Python複製async def main():
    user = await fetch_user() # 正確：等待並取得結果

Python複製async def main():
    user = fetch_user() # 錯誤心智模型：這是 coroutine object
    print(user)

6. Coroutine 與 Task 的關係：為什麼有 Coroutine 之後還需要 Task？

1先把四個容易混在一起的東西拆開

Python複製async def fetch_user(user_id: str) -> dict:
    user = await database.get_user(user_id)
    return user
coroutine_object = fetch_user("u-123")
task = asyncio.create_task(coroutine_object)

2Coroutine 到底保存了什麼？

Coroutine object 不是一個抽象的「工作名稱」，而是某一次函式呼叫的執行現場。 它概念上保存：

• 目前停在哪個 instruction／source position。
• 區域變數目前的值。
• 目前的 coroutine frame。
• 正在 await 哪個 awaitable。
• 恢復時要接收正常結果，還是被注入 exception。
• 最後 return 的值，或未處理的 exception。

Python複製import inspect
async def example() -> int:
    value = 10
    await asyncio.sleep(1)
    return value + 1
coro = example()
print(inspect.getcoroutinestate(coro))
# CORO_CREATED：已建立，但尚未開始執行

Coroutine 的典型狀態可以理解成：

這些狀態描述的是「函式執行進度」。但是它沒有完整回答： 何時開始、由誰恢復、是否獨立取消、結果由誰保存、錯誤由誰負責收走。

3Task 補上的不是函式內容，而是執行管理

Task 是 Future-like object。它包住一個 coroutine，並加入一個獨立 async 工作需要的管理能力：

4直接 await coroutine：它通常不會變成新的 Task

Python複製import asyncio
async def child() -> int:
    print("child task:", asyncio.current_task().get_name())
    await asyncio.sleep(1)
    return 42
async def parent() -> None:
    print("parent task:", asyncio.current_task().get_name())
    result = await child()
    print(result)
asyncio.run(parent())

概念上的結構不是「parent Task 裡再自動產生 child Task」，而是：

因此 parent() 和 child() 中的 asyncio.current_task() 會看到同一個 Task。 它們只是同一個 Task 裡的不同 coroutine frames。

5create_task：明確建立新的 scheduling identity

Python複製import asyncio
async def child() -> int:
    task = asyncio.current_task()
    print("child task:", task.get_name())
    await asyncio.sleep(1)
    return 42
async def parent() -> None:
    parent_task = asyncio.current_task()
    print("parent task:", parent_task.get_name())
    child_task = asyncio.create_task(
        child(),
        name="child-task",
    )
    print("parent continues before child result")
    result = await child_task
    print(result)
asyncio.run(parent())

現在結構變成：

create_task() 不是讓 coroutine「比較 async」； 它是建立新的排程與生命週期邊界，使 child 可以和 parent 交錯前進。

Python複製# 多數情況下沒有必要
task = asyncio.create_task(fetch_user())
user = await task
# 更直接
user = await fetch_user()

6兩條時間線：直接 await 與建立 Task 到底差在哪裡？

情況 A：直接 await

Python複製async def main():
    result_a = await fetch_a()
    result_b = await fetch_b()

A 和 B 都在同一個 Main Task 裡；必須等 A 的 await chain 完成， 程式才會執行到建立 B coroutine 的那一行。

情況 B：建立兩個 Tasks

Python複製async def main():
    task_a = asyncio.create_task(fetch_a())
    task_b = asyncio.create_task(fetch_b())
    result_a = await task_a
    result_b = await task_b

A、B 現在是兩個 event loop 可以分別恢復的工作單位。 這才是建立 Task 所帶來的 concurrency。

7底層：Task 怎麼「驅動」Coroutine？

Coroutine 具有類似 generator 的恢復介面。概念上可以被：

• send(...)：送入值並繼續執行。
• throw(...)：在暫停點注入 exception。
• close()：要求 coroutine 清理並結束。

Application code 幾乎不應該直接操作這些方法；Task 會替你處理。 以下是高度簡化、只用來建立心智模型的偽程式碼：

Python複製class ConceptualTask:
    def __init__(self, coroutine, loop):
        self.coroutine = coroutine
        self.loop = loop
        self.result_value = None
        self.exception_value = None
        self.done = False
        loop.call_soon(self.step)
    def step(self, value=None, exception=None):
        try:
            if exception is None:
                awaited = self.coroutine.send(value)
            else:
                awaited = self.coroutine.throw(exception)
        except StopIteration as completed:
            self.result_value = completed.value
            self.done = True
        except BaseException as error:
            self.exception_value = error
            self.done = True
        else:
            awaited.add_done_callback(self.wake_up)
    def wake_up(self, completed_future):
        try:
            value = completed_future.result()
        except BaseException as error:
            self.loop.call_soon(self.step, None, error)
        else:
            self.loop.call_soon(self.step, value, None)

真正的流程可壓縮成：

這就是為什麼需要 Task：Coroutine 知道自己如何暫停和恢復， 但 Task 負責把它接上 event loop 和 Future completion。

8一個 Task 可以驅動整條巢狀 Coroutine Chain

Python複製async def repository_get_user(user_id: str) -> dict:
    return await database.get_user(user_id)
async def service_get_user(user_id: str) -> dict:
    user = await repository_get_user(user_id)
    return normalize(user)
async def endpoint(user_id: str) -> dict:
    return await service_get_user(user_id)

這通常不是三個 Tasks，而是：

Text複製Request Task
└── endpoint coroutine
    └── service_get_user coroutine
        └── repository_get_user coroutine
            └── database Future

這是 Task 和 Coroutine 分離最重要的好處之一： 普通 async function composition 不需要為每一層 service／repository 建立新的排程單位。

9為什麼不能讓每個 Coroutine 自動具有 Task 功能？

技術上可以設計另一種語言，讓 async function 一呼叫就自動啟動。 Python 沒這樣做，是為了保留以下重要性質。

理由 A：分離工作內容與排程政策

這種分離稱為 separation of computation from scheduling policy： 同一個 coroutine function 可以依情境被循序 await，也可以成為 concurrent Task。

理由 B：避免隱藏 concurrency 與副作用

Python複製# Python 現在的行為：只建立 coroutine object
operation = charge_credit_card(order)
# 明確決定何時執行
receipt = await operation
# 或明確建立獨立工作
task = asyncio.create_task(operation)

如果 coroutine 一建立就自動成為 Task，那麼僅僅呼叫函式可能已經開始扣款、寫 DB、 發送 HTTP request 或產生 exception。程式的 concurrency 邊界會變得不透明。

理由 C：大部分巢狀 async calls 不需要獨立 Task

Endpoint → Service → Repository 通常是一條相依工作鏈。 每一層自動建立 Task 只會增加物件、排程、取消與除錯成本，卻不產生有用 concurrency。

理由 D：建立 Coroutine 時可能還沒有 running event loop

Python複製async def main():
    ...
coro = main()
# 此刻只建立 coroutine object，可能還沒有 running loop
asyncio.run(coro)
# run() 建立 runner／loop，並負責執行入口 awaitable

Task 必須由某個 event loop 執行；asyncio.create_task() 也要求目前 Thread 已有 running loop。Coroutine object 則可以先存在，再交給 runner。

理由 E：Task 是 Future-like 結果容器

Task 繼承 asyncio Future 的大部分 API，並以 wrapped coroutine 的 return／raise 決定自己的 result／exception。這讓它可以被多個地方等待。

Python複製task = asyncio.create_task(fetch_user())
consumer_a = await task
consumer_b = await task
assert consumer_a is consumer_b

相反地，同一個 coroutine object 不能執行兩次：

Python複製coro = fetch_user()
first = await coro
second = await coro
# RuntimeError: cannot reuse already awaited coroutine

Coroutine 是一次性的執行現場；Task 是一次執行加上一個可重複讀取的完成結果。

理由 F：取消與錯誤需要明確生命週期邊界

直接 await child coroutine 時，child 是 caller Task 的 await chain； 建立 child Task 後，它有自己的 cancellation state 和 exception channel。

理由 G：Structured concurrency 需要「子工作物件」

Python複製async with asyncio.TaskGroup() as group:
    user_task = group.create_task(fetch_user())
    config_task = group.create_task(fetch_config())
# 離開區塊時，兩個 Tasks 都已結束
user = user_task.result()
config = config_task.result()

TaskGroup 可以持有子 Tasks、等待它們、在失敗時取消 sibling Tasks， 並建立清楚的 parent／child lifetime。Raw coroutine object 本身不提供完整的 ownership model。

理由 H：同一份 Coroutine 模型可搭配不同 event-loop implementations

Coroutine 是 Python language/runtime 層級的 awaitable； Task 則由 asyncio／event loop implementation 建立與管理。 這讓內建 asyncio loop 和 uvloop 都能提供相容的 Task scheduling。

10Result、Exception、Cancellation 的傳遞差異

直接 await：像普通函式呼叫鏈

Python複製async def child():
    raise ValueError("child failed")
async def parent():
    await child()

錯誤沿同一個 Task 的 await chain 往上傳：

Text複製child coroutine
    ↓ ValueError
parent coroutine
    ↓ ValueError
parent Task

獨立 Task：錯誤先存入 Task

Python複製async def parent():
    task = asyncio.create_task(child())
    # child 的錯誤先讓 child Task 進入 done 狀態
    await asyncio.sleep(1)
    # 讀取 child Task 結果時，錯誤重新拋出
    await task

如果沒有任何人 await Task 或讀取 task.exception()， 可能出現 Task exception was never retrieved。

Cancellation

Python複製task = asyncio.create_task(worker())
task.cancel()
try:
    await task
except asyncio.CancelledError:
    print("worker was cancelled")

task.cancel() 會安排在 wrapped coroutine 的下一個適當執行機會注入 CancelledError。Coroutine 可以在 finally 清理資源， 通常應讓取消繼續傳播。

11Task 自己 await Task，會發生什麼？

Python複製async def parent():
    child_task = asyncio.create_task(child())
    result = await child_task
    return result

Parent Task await Child Task 時：

1. Parent Task 暫停。
2. Child Task 繼續由 event loop 獨立排程。
3. Child Task 完成後，自己的 Future-like 狀態變成 done。
4. 完成 callback 讓 Parent Task 重新進入 ready queue。
5. Parent Task 恢復，取得 Child Task 保存的 result 或 exception。

12Task 的建立時機：不是保證下一行之前一定沒執行

一般心智模型可以說 create_task() 把 coroutine 排定為「很快執行」。 傳統預設通常會在目前工作下一次把控制權交回 event loop 後開始。

但較新的 Python 支援 eager task execution／eager_start： 在啟用時，Task 可能在建立當下就同步推進 coroutine，直到第一次真正 blocking。 因此 production code 不應依賴「create_task 後，child 絕對還沒開始」這種假設。

Python複製task = asyncio.create_task(operation())
# 不要把正確性建立在 operation 此刻一定尚未開始
update_shared_state()

13什麼時候直接 await？什麼時候建立 Task？

快速決策流程

14常見誤解修正

15最後用一句工程術語收尾

本章官方依據

• Python：Coroutines and Tasks
• Python：Futures
• Python Data Model：Coroutine Objects

7. Task：event loop 管理 coroutine 的工作單位

1生活直覺與工程術語

Coroutine 像一份辦案流程；Task 是把某一份流程正式立案後的案件。 立案後系統可以知道案件是否待處理、正在等待、完成、失敗或被取消。

2High-level 定義

Python複製task = asyncio.create_task(fetch_user(), name="fetch-user")
user = await task

3底層運作

1. Task 被建立並排入 event loop 的 ready work。
2. Event loop 執行 Task；Task 推進 coroutine。
3. Coroutine await 尚未完成的 Future，Task 暫停。
4. Task 在該 Future 上登記 wake-up callback。
5. Future 完成後 callback 讓 Task 重新進入 ready queue。
6. Coroutine return 時，Task 儲存 result；拋錯時儲存 exception。

4為什麼連續 await 與 create_task 不同？

Python複製# 有先後相依：A 完成後才會開始 B
result_a = await fetch_a()
result_b = await fetch_b()

Python複製# 兩份工作先各自立案，由 event loop 交錯推進
task_a = asyncio.create_task(fetch_a())
task_b = asyncio.create_task(fetch_b())
result_a = await task_a
result_b = await task_b

5實務管理：優先使用 TaskGroup

Python複製async with asyncio.TaskGroup() as group:
    user_task = group.create_task(fetch_user())
    config_task = group.create_task(fetch_config())
user = user_task.result()
config = config_task.result()

TaskGroup 建立明確的生命週期邊界：離開區塊前會等待子 Tasks；其中一個發生未處理錯誤時， 會取消其他 sibling Tasks 並彙整錯誤。這是 structured concurrency。

8. Thread：Process 內真正執行指令的一條路徑

1生活直覺與術語橋接

多條 Threads 像同一間廚房裡的多位廚師：大家能直接拿同一台冰箱裡的材料，溝通很快； 但也可能同時修改同一鍋湯，造成資料競爭。

2High-level 定義

3與 Task 的核心差異

4GIL 與真正平行

在一般預設 CPython 中，GIL 通常限制同一 interpreter 內同時執行 Python bytecode 的 Threads 數量， 因此純 Python CPU-heavy 工作不會因增加 Threads 而線性加速。Thread 仍適合 blocking I/O， 某些釋放 GIL 的 native libraries 也可能平行執行。

5實際使用

Python複製result = await asyncio.to_thread(
blocking_sdk.fetch,
request_id,
)

這不是把 blocking function 變成真正 async API，而是把等待成本移到 thread pool， 讓 event-loop Thread 可以繼續服務其他 Tasks。

9. Process：具有獨立虛擬記憶體與執行資源的程式個體

1生活直覺與術語橋接

不同 Processes 像不同分店。每間店有自己的庫存與工作台；一間店修改庫存，不會直接改到另一間。 要交換物品必須透過物流或正式通訊管道。

2High-level 定義

3為什麼適合 CPU-heavy？

多個 Processes 可以由 OS 排到不同 CPU cores 上，並各自執行 Python interpreter， 因此可用於純 Python CPU-heavy parallelism。代價是啟動、記憶體與跨 Process 傳輸較昂貴。

Python複製from concurrent.futures import ProcessPoolExecutor
def cpu_heavy(number: int) -> int:
    return sum(value * value for value in range(number))
with ProcessPoolExecutor() as pool:
    results = list(pool.map(cpu_heavy, [20_000_000] * 4))

4FastAPI workers 也是 Processes

uvicorn --workers 4 建立的是四個 server Processes，不是四個 Tasks。 它們各自載入 application，也各自擁有 event loop 與 connection pools。

5選擇限制

• 資料傳輸可能需要 serialization，巨大資料的 IPC 成本可能抵銷平行收益。
• 每個 Process 都會增加記憶體與連線數。
• 某個 Process 內的普通 global variable 不會自動與其他 Processes 同步。
• 影片轉碼、模型推論等工作也可能已有自己的 native threads／GPU，需要避免過度並行。

10. 完整比較表：把五層理解壓縮成一張工程決策表

11. 排程與切換方式：從直覺切換到 cooperative／preemptive scheduling

Task / Coroutine：合作式切換

正在執行的 Task 必須走到一個尚未完成的 await， 才會把控制權交給 event loop。程式設計者相對容易知道可能在哪些位置切換。

Thread / Process：搶佔式切換

作業系統可以在時間片用完、Thread 阻塞或其他排程條件出現時切換。 從 application 角度看，切換點比較難預測。

12. 程式範例：從需求判斷應使用哪一層

以下每個例子都依固定順序閱讀：先判斷工作是在等待還是在計算， 再決定使用 coroutine／Task 表達流程、Thread 橋接 blocking API，或 Process 取得多核心平行。

需求：同時查詢 100 個遠端 API

需求：將四支影片做 CPU-heavy 編碼

Coroutine 與 Task 的最小實驗

Python複製import asyncio
async def job(name: str, delay: float) -> str:
    print(f"{name}: start")
    await asyncio.sleep(delay)
    print(f"{name}: done")
    return name
async def sequential() -> None:
    # 第一個完成後才開始第二個
    await job("A", 2)
    await job("B", 1)
async def concurrent() -> None:
    # 先建立兩個獨立排程單位
    task_a = asyncio.create_task(job("A", 2))
    task_b = asyncio.create_task(job("B", 1))
    await task_a
    await task_b
asyncio.run(concurrent())

asyncio 與 blocking library 的橋接

Python複製import asyncio
import requests
def blocking_fetch(url: str) -> int:
    response = requests.get(url, timeout=10)
    return len(response.content)
async def main() -> None:
    # blocking_fetch 在 ThreadPool 的 Thread 執行，
    # event loop Thread 不會被 requests.get 卡住。
    size = await asyncio.to_thread(
        blocking_fetch,
        "https://example.com",
    )
    print(size)
asyncio.run(main())

asyncio 協調 ProcessPool

Python複製import asyncio
from concurrent.futures import ProcessPoolExecutor
def cpu_heavy(number: int) -> int:
    return sum(value * value for value in range(number))
async def main() -> None:
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        results = await asyncio.gather(
            loop.run_in_executor(pool, cpu_heavy, 20_000_000),
            loop.run_in_executor(pool, cpu_heavy, 20_000_000),
        )
    print(results)
if __name__ == "__main__":
    asyncio.run(main())

13. 實際開發時怎麼選？先判斷瓶頸，再選抽象層

快速選擇表

• FastAPI 同時呼叫三個獨立 API：TaskGroup。
• 處理 10,000 個 WebSocket 連線：asyncio Tasks。
• 只有同步版本的第三方 SDK：asyncio.to_thread() 或 ThreadPool。
• 大量圖片縮放：ProcessPool 或外部 worker。
• 影片轉檔：通常交給 FFmpeg subprocess／worker，不在 event loop 直接算。
• 單純先後相依的 async 操作：直接依序 await，不必 create_task。

15. 官方來源

教材以官方文件與專案文件為主要依據；比喻與分層順序是為初學者重新組織， 正式行為仍以實際 Python／Uvicorn／uvloop 版本文件為準。

• Python：Execution model（interpreter 與 Python runtime）
• Python：Python Runtime Services
• ASGI：Main specification（scope、receive、send）
• ASGI：Lifespan Protocol（startup／shutdown runtime context）
• Python：asyncio — Asynchronous I/O
• Python：Coroutines and Tasks
• Python：Event loop APIs 與 implementations
• Python：asyncio Platform Support（Windows／macOS）
• Python：selectors — High-level I/O multiplexing
• Python：threading — Thread-based parallelism
• Python：multiprocessing — Process-based parallelism
• Python：ThreadPoolExecutor／ProcessPoolExecutor
• FastAPI：Concurrency and async / await
• FastAPI：Server Workers
• Uvicorn：Event Loop 選擇與 custom loop
• Uvicorn：Settings（--loop、--workers）
• uvloop 官方專案：asyncio drop-in replacement、Cython、libuv
• libuv：Design overview 與 I/O loop
• libuv：Thread pool work scheduling