# Python 中的 asyncio：async/await、TaskGroup、线程、多进程与 GIL

URL: https://skillaria.top/zh/guides/python/asyncio-python

asyncio 是 Python 的标准库，用于编写异步代码：用 async def 定义的协程在 await 处暂停，此时事件循环可以执行其他任务，因此数百个网络请求可以在同一线程中同时进行。等待网络和磁盘时可以使用 asyncio 或线程（threading）；对于重计算，应使用进程（multiprocessing），因为 GIL 使普通 CPython 的线程无法并行计算。

asyncio 是 Python 的标准库，用于编写异步代码。用 `async def` 声明的函数可以在 `await` 处暂停：当一个函数在等待网络响应时，事件循环会运行其他函数。因此，数百个请求可以在同一个线程中同时进行。但 asyncio 只是三种工具之一，另外还有线程（`threading`）和进程（`multiprocessing`）。下面我们来看 asyncio 的工作原理，如何为任务选择合适的工具，以及 GIL 的作用。

## 等待与计算：两类不同的任务

不先理解这个区别，后面的内容就无从谈起。程序变慢通常有两种原因：

- **I/O 密集型（I/O-bound）** — 在等待输入输出：等待 API 的响应、数据库的结果或磁盘读取。此时处理器处于空闲状态。
- **CPU 密集型（CPU-bound）** — 在做计算：处理图像，循环遍历数百万条记录。此时处理器满负荷运转。

对于等待类任务，asyncio 或线程都适用：一个操作在等待时，另一个操作可以运行。对于计算类任务，则需要真正的并行计算，即在多个核心上同时运行——这要用进程。把二者混淆是最主要的错误：在纯 Python 中，线程不会加快计算速度；而为了三个 HTTP 请求去启动进程，则是杀鸡用牛刀。

## asyncio 是什么：协程与事件循环

异步代码建立在两个关键词之上。`async def` 声明一个协程，即可以暂停的函数；调用它时并不会执行函数体，而是创建一个协程对象。函数体内的 `await` 表示：“我要在这里等待，你先去处理其他任务。” 整个过程由事件循环（event loop）控制，通过 `asyncio.run()` 启动。

```
import asyncio

async def fetch_rate(currency):
    await asyncio.sleep(1)          # 模拟对 API 的请求
    return {"USD": 92.5, "EUR": 100.1, "CNY": 12.7}[currency]

async def main():
    rate = await fetch_rate("USD")
    print(rate)

asyncio.run(main())  # 92.5
```

`asyncio.sleep()` 是异步的暂停：它会把控制权交还给事件循环。如果在协程中使用普通的 `time.sleep()`，整个事件循环都会被阻塞，其他任务只能干等。

### 顺序执行还是同时执行

如果在循环中逐个写 `await`，请求会一个接一个地进行——三种货币就要花三秒。要让它们同时进行，需要把它们作为任务启动。

```
import asyncio, time

async def main():
    start = time.perf_counter()
    for cur in ("USD", "EUR", "CNY"):
        await fetch_rate(cur)                       # 依次执行
    print(f"{time.perf_counter() - start:.1f} 秒")   # 3.0 秒

asyncio.run(main())
```

## TaskGroup：一起启动，等待全部完成

现代的做法（Python 3.11+）是使用任务组。在 `async with` 代码块中，每个任务通过 `tg.create_task()` 立即启动，代码块在所有任务结束后才退出。

```
async def main():
    start = time.perf_counter()
    async with asyncio.TaskGroup() as tg:
        tasks = {cur: tg.create_task(fetch_rate(cur)) for cur in ("USD", "EUR", "CNY")}
    rates = {cur: t.result() for cur, t in tasks.items()}
    print(rates, f"{time.perf_counter() - start:.1f} 秒")   # ... 1.0 秒

asyncio.run(main())
```

TaskGroup 的错误处理很严格：如果某个任务失败，其余任务都会被取消，所有错误被收集到 `ExceptionGroup` 中。捕获它们要使用 `except*`：

```
async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fetch_rate("USD"))
            tg.create_task(fetch_rate("XXX"))   # KeyError
    except* KeyError as eg:
        print("未知货币:", eg.exceptions)
```

### asyncio.gather 与 create_task

在 TaskGroup 出现之前，标准做法是 `asyncio.gather()`：它同时运行多个协程，并按相同的顺序返回结果。现有代码中仍然大量使用它。

```
async def main():
    usd, eur, cny = await asyncio.gather(
        fetch_rate("USD"), fetch_rate("EUR"), fetch_rate("CNY")
    )
    # 使用 return_exceptions=True 时，错误会作为列表元素返回，而不会中断全部任务
```

`asyncio.create_task()` 可以在后台启动协程，但不属于任何组。必须保存对任务的引用，否则它可能被垃圾回收器清除，而且它的错误也没有人会看到。因此在新项目中，TaskGroup 更可靠。

### 取消任务与超时

```
async def main():
    task = asyncio.create_task(fetch_rate("USD"))
    task.cancel()                     # 在任务内部会抛出 CancelledError
    try:
        await task
    except asyncio.CancelledError:
        print("已取消")

    try:
        async with asyncio.timeout(0.5):   # Python 3.11+
            await fetch_rate("EUR")
    except TimeoutError:
        print("耗时太长")
```

在较旧的版本中，可以用 `asyncio.wait_for(coro, timeout=0.5)` 代替 `asyncio.timeout()`。另外还有 `asyncio.wait()`，它返回已完成和未完成的任务集合，并且可以只等待第一个完成的任务（`return_when=asyncio.FIRST_COMPLETED`）。

在 Python 3.15 中，正如课程所示，TaskGroup 新增了 `tg.cancel()` 方法：它会取消组内的所有任务。这解决了“谁先返回就用谁”的场景：同时向三个镜像服务发出请求，取第一个响应，然后取消其余请求。以前要么抛出自定义的信号异常，要么手动取消任务。

## 同步原语：Semaphore、Lock、Event、Queue

asyncio 有自己的同步原语，它们通过 `await` 使用。其中最有用的是 `Semaphore`：它限制同时进行的操作数量，避免一次性向别人的 API 发出上千个请求。

```
sem = asyncio.Semaphore(10)

async def limited(url):
    async with sem:              # 同时最多 10 个
        await asyncio.sleep(0.1) # 这里原本是一次请求
        return url
```

- `asyncio.Lock` — 临界区内同一时间只允许一个协程执行。
- `asyncio.Event` — 一个协程通过 `await event.wait()` 等待，另一个通过 `event.set()` 发出信号。
- `asyncio.Queue` — 生产者—消费者队列：`await q.put()`、`await q.get()`、`q.task_done()`。

## 常见陷阱：忘记写 await

最常见的错误是调用协程时忘记写 `await`：

```
async def main():
    rate = fetch_rate("USD")    # 忘记写 await
    print(rate * 2)             # TypeError
# RuntimeWarning: coroutine 'fetch_rate' was never awaited
```

Python 并没有执行这个函数，只是创建了协程对象。如果在日志中看到 “was never awaited”，就去找漏掉的 `await`。第二个常见错误是在协程中调用阻塞函数（`time.sleep`、`requests.get`）。对于 HTTP，请使用异步库（httpx、aiohttp）；无法避免的同步代码则通过 `await asyncio.to_thread(func, arg)` 放到线程中执行。

## Python 中的多线程：threading 模块

线程是同一进程内独立的执行流，线程之间共享内存。对于等待网络的任务，线程非常有效，而且不需要把代码改写成 async。

```
import threading, time

def download(name):
    time.sleep(1)               # 等待网络
    print(name, "完成")

threads = [threading.Thread(target=download, args=(n,)) for n in ("a", "b", "c")]
for t in threads:
    t.start()
for t in threads:
    t.join()                    # 总共约 1 秒
```

共享内存既是优点，也是错误的来源：如果两个线程修改同一个变量，就需要 `threading.Lock`。通常更简单的做法是不手动管理线程，而是使用线程池。

### ThreadPoolExecutor

`concurrent.futures.ThreadPoolExecutor` 创建一个线程池，并把任务分配给线程。`map` 按原始顺序返回结果。

```
from concurrent.futures import ThreadPoolExecutor
import time

def fetch(url):
    time.sleep(1)
    return f"{url}: 200"

urls = [f"https://example.com/{i}" for i in range(5)]
with ThreadPoolExecutor(max_workers=5) as pool:
    for result in pool.map(fetch, urls):
        print(result)           # 总共约 1 秒
```

## 什么是 GIL，为什么线程无法加速计算

GIL（Global Interpreter Lock，全局解释器锁）是 CPython 解释器的全局锁。在任意时刻，只有一个线程能够执行 Python 字节码。线程等待输入输出时会释放 GIL，所以线程对网络任务很有用。但在纯计算中，四个线程并不会比一个线程快：它们只是轮流持有这把锁。

```
from concurrent.futures import ThreadPoolExecutor

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

with ThreadPoolExecutor(4) as pool:
    list(pool.map(heavy, [5_000_000] * 4))   # 大致与顺序执行相同
```

用 C 编写的库（NumPy 及类似库）在执行重型操作时常常会释放 GIL，因此线程对它们能带来加速。

## multiprocessing：真正的并行

每个进程都有自己的解释器和自己的 GIL，因此进程可以在不同的核心上同时计算。代价是独立的内存：进程之间通过序列化（pickle）传递数据，而且启动进程比启动线程开销更大。

```
from concurrent.futures import ProcessPoolExecutor

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

if __name__ == "__main__":      # 在 Windows 和 macOS 上必须写
    with ProcessPoolExecutor() as pool:
        print(list(pool.map(heavy, [5_000_000] * 4)))
```

如果没有 `if __name__ == "__main__"` 这一判断，子进程在导入模块时会不断创建新的进程。还有底层的 `multiprocessing` 模块，提供 `Process`、`Pool`、`Queue`，但 `ProcessPoolExecutor` 更简单，而且与线程池的接口相同。

## 如何选择：asyncio、threading 还是 multiprocessing

| 任务 | 工具 | 原因 |
| --- | --- | --- |

| 下载 200 个页面，大量 API 请求 | asyncio（httpx/aiohttp）或 ThreadPoolExecutor | 等待网络，处理器空闲 |

| 在已经是异步的程序中发出多个请求 | TaskGroup | 代码已经是 async |

| 没有异步版本的同步库 | 线程，`asyncio.to_thread` | 无需改写代码 |

| 纯 Python 的重计算 | ProcessPoolExecutor | 绕过 GIL，利用所有核心 |

## 无 GIL 的 Python 与子解释器：有什么新变化

从 Python 3.13 开始，存在一种没有 GIL 的构建版本（free-threading，PEP 703），在名称中带有字母 t，例如 `python3.15t`。3.13 中它还是实验性的，从 3.14 起（PEP 779）正式支持。在这个版本中，线程可以在不同核心上并行计算。但默认情况下你仍然得到带 GIL 的普通 Python：单线程代码在 free-threaded 构建中会稍慢，而且并非所有库都已经适配。

```
import sys
print(sys._is_gil_enabled())   # 在 free-threaded 构建中为 False
```

实现并行计算的另一条路径出现在 3.14 中（PEP 734）——子解释器，通过 `concurrent.interpreters` 模块使用：同一进程中有多个相互隔离的解释器，每个都有自己的 GIL。最方便的用法是 `InterpreterPoolExecutor`，它的写法与线程池相同。课程中的测试显示，四个重型任务用时 0.39 秒，而单线程需要 1.03 秒。

## 小结

asyncio 是用于等待类任务的工具：用 `async def` 定义协程，用 `await` 暂停，通过 TaskGroup 或 `gather` 同时启动多个任务。线程无需把代码改写成 async 就能解决同样的问题；而对于重计算，需要使用进程，因为 GIL 使普通 CPython 的线程无法并行计算。free-threading 和子解释器正在逐步改变这一局面，但基本原则“等待用 asyncio 或线程，计算用进程”仍然成立。课程中会在实践里讲解这些内容：支出记录器并行获取汇率，只需一秒，而不是三秒。

## FAQ

**能否在普通代码中调用 async 函数？**

可以，使用 asyncio.run(coro()) —— 它会创建事件循环、执行协程，然后关闭循环。在已经运行的事件循环内部不能调用 asyncio.run，那里应该使用 await。

**asyncio 会使用多个线程吗？**

不会。事件循环在一个线程中运行所有协程，只在 await 处切换。因此它的数据竞争比线程少，但一个阻塞操作会使整个程序停止。

**需要用 pip 安装 asyncio 吗？**

不需要，asyncio 已包含在 Python 3 的标准库中，直接写 import asyncio 即可。PyPI 上的 asyncio 包已经过时，无需安装。

**如何在 asyncio 中运行重计算函数而不阻塞事件循环？**

计算任务可以使用 loop.run_in_executor 配合 ProcessPoolExecutor；阻塞的输入输出可以使用 asyncio.to_thread。事件循环会继续处理其他任务。

**协程和任务（Task）有什么区别？**

协程本身什么都不做，直到被 await 才执行。任务则是对协程的包装，并立即被安排到事件循环中，因此它会与其他代码并行运行。


---
Source: Skillaria.Top (https://skillaria.top)