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

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

无 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 或线程，计算用进程”仍然成立。课程中会在实践里讲解这些内容：支出记录器并行获取汇率，只需一秒，而不是三秒。