1.1 基礎概念與事件迴圈

2026-01-20

本章介紹 asyncio 的核心概念。理解這些概念是掌握異步程式設計的基礎。

先備知識

入門系列 3.7 並行處理
了解 I/O 密集任務的特性

本章目標

學完本章後，你將能夠：

理解並發、並行、異步的區別
解釋事件迴圈的工作原理
寫出第一個異步程式
判斷何時使用 asyncio

【原理層】並發、並行與異步

三個容易混淆的概念

在開始之前，我們需要釐清三個經常被混用的概念：

 1並發（Concurrency）：
 2  同時「處理」多件事情（不一定同時執行）
 3  重點是結構和設計
 4
 5並行（Parallelism）：
 6  同時「執行」多件事情
 7  需要多核 CPU 或多台機器
 8
 9異步（Asynchronous）：
10  不等待結果就繼續執行
11  一種實現並發的方式

用一個生活化的例子：

 1同步做早餐：
 2  1. 煮咖啡（等 5 分鐘）
 3  2. 烤土司（等 3 分鐘）
 4  3. 煎蛋（等 4 分鐘）
 5  總共：12 分鐘
 6
 7異步做早餐（一個人）：
 8  1. 啟動咖啡機
 9  2. 放土司進烤箱
10  3. 開始煎蛋
11  4. 等待全部完成
12  總共：約 5 分鐘（最長任務的時間）
13
14並行做早餐（三個人）：
15  三人同時分別處理
16  總共：約 5 分鐘

asyncio 是並發，不是並行

這是最重要的概念：asyncio 在單執行緒中實現並發。

 1import asyncio
 2import time
 3
 4async def task(name, delay):
 5    print(f"{name} 開始")
 6    await asyncio.sleep(delay)  # 非阻塞等待
 7    print(f"{name} 完成")
 8    return name
 9
10async def main():
11    start = time.perf_counter()
12
13    # 同時啟動三個任務
14    results = await asyncio.gather(
15        task("A", 2),
16        task("B", 1),
17        task("C", 1.5)
18    )
19
20    elapsed = time.perf_counter() - start
21    print(f"總耗時：{elapsed:.2f}s")  # 約 2 秒，不是 4.5 秒
22
23asyncio.run(main())

輸出：

1A 開始
2B 開始
3C 開始
4B 完成
5C 完成
6A 完成
7總耗時：2.00s

三個任務「並發」執行，但都在同一個執行緒中。

為什麼不用多執行緒？

你可能會問：用 threading 也能達到類似效果，為什麼要用 asyncio？

面向	threading	asyncio
記憶體開銷	每執行緒約 8MB stack	每協程約幾 KB
切換成本	OS 排程，較高	用戶空間切換，極低
並發數量	數百到數千	數萬到數十萬
執行緒安全	需要鎖保護	單執行緒，天生安全
除錯難度	競爭條件難以重現	確定性較高

當你需要處理大量 I/O 並發（如 Web 伺服器、爬蟲）時，asyncio 通常是更好的選擇。

【設計層】事件迴圈

什麼是事件迴圈？

事件迴圈（Event Loop）是 asyncio 的核心，它負責：

排程和執行協程
處理 I/O 事件
管理回呼函式

 1事件迴圈的運作：
 2
 3┌─────────────────────────────────────┐
 4│           事件迴圈                    │
 5│  ┌─────────────────────────────┐    │
 6│  │    就緒佇列（Ready Queue）    │    │
 7│  │  [協程A] [協程B] [協程C]      │    │
 8│  └─────────────────────────────┘    │
 9│              ↓                       │
10│         執行就緒任務                  │
11│              ↓                       │
12│         遇到 await                   │
13│              ↓                       │
14│  ┌─────────────────────────────┐    │
15│  │   等待佇列（Waiting Queue）   │    │
16│  │  [等待 I/O] [等待計時器]      │    │
17│  └─────────────────────────────┘    │
18│              ↓                       │
19│         I/O 完成                     │
20│              ↓                       │
21│         放回就緒佇列                  │
22└─────────────────────────────────────┘

協作式多任務

asyncio 使用「協作式多任務」（Cooperative Multitasking）：

任務主動讓出控制權（透過 await）
事件迴圈選擇下一個就緒任務執行
沒有強制搶佔

 1async def cooperative_task(name):
 2    for i in range(3):
 3        print(f"{name}: 步驟 {i}")
 4        await asyncio.sleep(0)  # 主動讓出控制權
 5    print(f"{name}: 完成")
 6
 7async def main():
 8    await asyncio.gather(
 9        cooperative_task("A"),
10        cooperative_task("B")
11    )
12
13asyncio.run(main())

輸出：

1A: 步驟 0
2B: 步驟 0
3A: 步驟 1
4B: 步驟 1
5A: 步驟 2
6B: 步驟 2
7A: 完成
8B: 完成

注意任務是交替執行的，每次 await 都是一個切換點。

asyncio.run() 的背後

asyncio.run() 是 Python 3.7+ 推薦的入口點：

1# 這是簡化的偽代碼
2def run(coro):
3    loop = asyncio.new_event_loop()
4    try:
5        asyncio.set_event_loop(loop)
6        return loop.run_until_complete(coro)
7    finally:
8        loop.close()

它做了三件事：

建立新的事件迴圈
執行協程直到完成
關閉事件迴圈

【實作層】第一個異步程式

async 和 await

 1import asyncio
 2
 3# async def 定義協程函式
 4async def greet(name, delay):
 5    print(f"開始問候 {name}")
 6    await asyncio.sleep(delay)  # await 等待可等待物件
 7    print(f"你好，{name}！")
 8    return f"問候 {name} 完成"
 9
10async def main():
11    # 方法 1：依序執行
12    result1 = await greet("Alice", 1)
13    result2 = await greet("Bob", 1)
14    # 總共約 2 秒
15
16    # 方法 2：並發執行
17    results = await asyncio.gather(
18        greet("Charlie", 1),
19        greet("David", 1)
20    )
21    # 總共約 1 秒
22
23asyncio.run(main())

協程 vs 協程函式

這是一個常見的混淆點：

 1async def my_coro():
 2    return 42
 3
 4# my_coro 是協程函式（coroutine function）
 5print(type(my_coro))  # <class 'function'>
 6
 7# my_coro() 是協程物件（coroutine object）
 8coro = my_coro()
 9print(type(coro))  # <class 'coroutine'>
10
11# 協程物件需要被執行
12result = asyncio.run(coro)
13print(result)  # 42

常見錯誤：忘記 await

 1async def fetch_data():
 2    await asyncio.sleep(1)
 3    return {"data": "value"}
 4
 5async def main():
 6    # 錯誤：沒有 await
 7    result = fetch_data()  # 這只是建立協程物件，沒有執行
 8    print(result)  # <coroutine object fetch_data at 0x...>
 9
10    # 正確：使用 await
11    result = await fetch_data()
12    print(result)  # {'data': 'value'}
13
14asyncio.run(main())

Python 會發出警告：

1RuntimeWarning: coroutine 'fetch_data' was never awaited

偵錯技巧

 1import asyncio
 2
 3async def main():
 4    # 取得當前事件迴圈
 5    loop = asyncio.get_running_loop()
 6    print(f"事件迴圈：{loop}")
 7
 8    # 檢查是否在事件迴圈中
 9    print(f"正在執行：{loop.is_running()}")
10
11asyncio.run(main())

【選擇指南】何時使用 asyncio

適合 asyncio 的場景

Web 伺服器：處理大量並發請求
API 客戶端：批次呼叫多個 API
網路爬蟲：同時抓取多個網頁
即時應用：WebSocket、聊天室
資料庫操作：批次查詢

 1# 範例：並發下載多個網頁
 2import asyncio
 3import aiohttp
 4
 5async def fetch(session, url):
 6    async with session.get(url) as response:
 7        return await response.text()
 8
 9async def main():
10    urls = [
11        "https://python.org",
12        "https://docs.python.org",
13        "https://pypi.org"
14    ]
15
16    async with aiohttp.ClientSession() as session:
17        tasks = [fetch(session, url) for url in urls]
18        results = await asyncio.gather(*tasks)
19
20    for url, html in zip(urls, results):
21        print(f"{url}: {len(html)} bytes")
22
23asyncio.run(main())

不適合 asyncio 的場景

CPU 密集任務：asyncio 無法繞過 GIL
簡單腳本：增加複雜度沒有好處
依賴同步函式庫：需要額外處理

決策流程

 1任務類型？
 2    │
 3    ├─→ CPU 密集
 4    │       └─→ multiprocessing 或 Free-threading
 5    │
 6    └─→ I/O 密集
 7            │
 8            ├─→ 並發量 < 100
 9            │       └─→ threading 可能夠用
10            │
11            └─→ 並發量 > 100
12                    │
13                    ├─→ 需要共享複雜狀態
14                    │       └─→ threading + Lock
15                    │
16                    └─→ 任務相對獨立
17                            └─→ asyncio（推薦）

思考題

為什麼說 asyncio 是「協作式」而不是「搶佔式」？這對程式設計有什麼影響？
如果一個協程中有 CPU 密集的計算而沒有 await，會發生什麼事？
asyncio.sleep(0) 的作用是什麼？

實作練習

寫一個程式，同時「下載」5 個檔案（用 asyncio.sleep() 模擬下載時間），並顯示總耗時
修改上面的程式，讓它顯示每個檔案完成的順序
實作一個簡單的計時器，每秒印出一次時間，持續 5 秒

延伸閱讀

下一章：協程與 Task 管理

#python #python-advanced #asyncio #event-loop #coroutine