Async、线程还是进程

目标：能够根据工作负载是在等待还是计算、依赖是否支持 await、数据是否可序列化，选择顺序、AsyncIO、线程池或进程池。

“并发更快”不是规则。先问时间花在哪里：

async-native I/O：使用 AsyncIO，等待时让出事件循环；
只提供同步 API 的阻塞 I/O：使用线程或 asyncio.to_thread()，避免阻塞 loop；
纯 Python CPU 密集任务：线程通常受 GIL 限制，可考虑进程；
很小的任务：顺序执行可能最快，避免调度和序列化开销。

有些 C 扩展计算会释放 GIL，所以“CPU 一律进程”也不准确。选择需要真实工作负载测量。

完整示例：把阻塞 I/O 移出事件循环

保存为 execution_lab.py：

from __future__ import annotations

import asyncio
import time
from time import perf_counter


def blocking_call(name: str) -> str:
    time.sleep(0.03)
    return name.upper()


async def main() -> None:
    start = perf_counter()
    sequential = [blocking_call("a"), blocking_call("b")]
    sequential_time = perf_counter() - start

    start = perf_counter()
    concurrent = await asyncio.gather(
        asyncio.to_thread(blocking_call, "a"),
        asyncio.to_thread(blocking_call, "b"),
    )
    threaded_time = perf_counter() - start

    assert sequential == concurrent == ["A", "B"]
    assert threaded_time < sequential_time
    print("blocking waits overlapped in threads")


if __name__ == "__main__":
    asyncio.run(main())

运行：

uv run python execution_lab.py

预期输出：

blocking waits overlapped in threads

这个实验只证明 sleep 等待可以在线程中重叠，不证明你的数据库驱动或 SDK 是 thread-safe，也不决定生产 worker 数。

进程池的额外契约

ProcessPoolExecutor 使用独立进程，可以绕过单个解释器的 GIL，但有代价：启动进程、序列化参数和结果、独立内存。提交的函数和数据必须可 pickle，worker 还需要能导入 __main__ 模块，因此入口保护很重要：

from concurrent.futures import ProcessPoolExecutor


def cpu_work(limit: int) -> int:
    return sum(number * number for number in range(limit))


def main() -> None:
    with ProcessPoolExecutor() as pool:
        print(list(pool.map(cpu_work, [100_000, 100_000])))


if __name__ == "__main__":
    main()

不要把 worker 函数定义为 main 内部的局部函数；它通常无法 pickle。Python 3.14 改变了 POSIX 上 ProcessPoolExecutor 的默认启动方式，不再默认 fork，这也是课程保持显式入口保护的原因。

引导实验：同时检查响应性

在 event loop 中并发运行 heartbeat 和直接 blocking_call，你会看到 heartbeat 停顿。改成 await asyncio.to_thread(blocking_call, "x") 后 heartbeat 能继续。总耗时之外，响应性也是选择执行模型的证据。

独立练习

为三类工作写选择表并各给一个理由：async HTTP client、同步文件压缩库、纯 Python 大素数判断。然后对 CPU 示例分别顺序、线程、进程运行五次，报告中位数和机器信息。

参考判断

async HTTP client：AsyncIO；库本身提供 await，能高效重叠网络等待。
同步文件压缩：先查实现是否释放 GIL；可能适合线程，不能仅凭“文件”二字决定。
纯 Python 大素数判断：任务足够大且可分割时考虑进程；小输入先顺序。

基准必须同时断言结果一致。若进程版本对小任务更慢，这是合理结果，不是实验失败。

常见失败