Python 并发模型：线程、多进程与 asyncio

本篇目标

面对真实任务时能判断使用线程、多进程还是 asyncio，并看懂超时、取消、结构化并发和并发上限怎样贯穿调用链。

区分并发与并行选择执行模型传播取消避免失控任务

先记住一句话

asyncio 用一个事件循环高效重叠大量异步 I/O，线程适合接入阻塞 I/O 库，多进程适合隔离或并行执行纯 Python CPU 密集任务；三者都必须设置生命周期、上限和失败传播。

并发不等于并行

• 并发：多个任务在一段时间内都有进展，可以通过交替执行实现；
• 并行：多个任务在同一时刻真正使用多个 CPU 核心执行；
• 异步：调用方在等待结果时主动让出执行权，是组织并发的一种方式；
• 进程和线程：进程提供资源与隔离边界，线程负责执行代码；二者都由操作系统管理；
• 协程：运行在线程中的可暂停任务，通常由事件循环管理。

进程、线程和协程怎么区分

可以先记成：进程像独立房间，线程像房间里的工人，协程像工人手中的任务单。每个进程有自己的房间和物品，也就是独立内存；同一进程里的线程共用房间里的物品；一张任务单等待结果时，工人可以先放下它，处理另一张任务单。

Python 程序启动
└─ 进程：拥有独立内存，像一间房
   ├─ 主线程：像房间里的工人
   │  └─ 事件循环：安排多张协程任务单
   │     ├─ 协程 A：等待网络时通过 await 暂停
   │     └─ 协程 B：在线程空闲时继续执行
   └─ 工作线程：与主线程共享进程内存

对比	进程（Process）	线程（Thread）	协程（Coroutine）
它是什么	正在运行的程序实例，也是资源隔离的边界	进程中负责执行代码的单位	线程中可以暂停和恢复的任务
由谁管理	操作系统	操作系统	程序中的事件循环，通常包装成 Task（可调度的任务对象）运行
内存关系	不同进程通常拥有独立内存	同一进程内的线程共享内存	没有独立内存，可访问进程内的共享对象
创建与切换	最重，传递数据通常需要队列或管道	比进程轻，可以直接访问共享对象	最轻，在 await 等位置协作切换
能否多核并行	可以分别使用多个 CPU 核心	默认 CPython 中执行 Python 字节码受 GIL 限制	本身不能带来多核并行
Python 常见用途	纯 Python CPU 密集计算、需要强隔离的任务	网络、文件等阻塞 I/O，或兼容同步库	大量提供异步接口的 I/O 任务
主要风险	通信麻烦，创建和切换成本高	共享数据可能竞争，需要锁或队列保护	阻塞代码会卡住事件循环，未管理的任务可能泄漏

三者不是互斥关系：一个进程至少有一个线程，一个线程又可以交替执行多个协程。asyncio 通常让一个事件循环在一个线程中管理多个协程；协程遇到 await 时主动暂停，让线程继续执行其他协程。它不会创建新线程，也不会自动获得多核并行能力。如果共享对象还会被其他线程使用，仍要确认它能否被多个线程安全地同时访问。

先按任务性质选择

下面不是三个完全独立的场景，而是一个两步判断过程：先判断任务是在等待 I/O，还是持续占用 CPU；如果是在等待 I/O，再看调用的库有没有异步接口。

一次请求中遇到耗时任务
├─ 主要时间在等待网络、文件或数据库（I/O 密集）
│  ├─ 调用的库提供 async / await 接口 → 使用 asyncio
│  └─ 调用的库只有同步阻塞接口       → 使用 to_thread 或受控线程池
└─ 主要时间在执行大量纯 Python 计算（CPU 密集）
   └─ 需要同时利用多个 CPU 核心       → 使用进程池或独立 Worker

因此，asyncio 和线程池都能处理 I/O 等待，区别在于依赖是否提供异步接口；大量纯 Python 计算才主要交给多进程。

场景	首选	原因与代价
FastAPI 调用异步数据库和异步 HTTP 客户端	asyncio	单线程可管理大量等待中的连接
在异步服务中调用阻塞文件或旧 SDK	asyncio.to_thread 或受控线程池	避免阻塞事件循环，但线程数仍要受控
图片预处理、纯 Python 文本计算	进程池或独立 Worker	绕开默认 GIL，但需要序列化参数且进程更重
简单脚本并发下载	线程池或 asyncio	根据依赖是否提供异步接口选择
需要强隔离的高风险任务	独立进程或任务服务	崩溃和内存边界更清楚

不要先决定 “项目必须异步” ，再强行套模型。先确认调用链中的库是否真的是异步 I/O，以及系统瓶颈在等待、CPU、连接数还是下游限额。

asyncio：结构化管理一组任务

TaskGroup 把相关子任务限制在一个明确作用域内：退出作用域前会等待任务结束；其中一个任务失败时，会取消同组剩余任务并传播异常。

# 文件位置：src/note_api/search.py
import asyncio
from collections.abc import Awaitable, Callable

# 类型别名：SearchCall 代表“接收查询字符串，并异步返回字符串列表”的函数。
SearchCall = Callable[[str], Awaitable[list[str]]]

async def search_all(
    query: str,
    vector_search: SearchCall,
    keyword_search: SearchCall,
) -> list[str]:
    # TaskGroup 统一管理两个同属一次检索请求的子任务。
    async with asyncio.TaskGroup() as group:
        vector_task = group.create_task(vector_search(query))
        keyword_task = group.create_task(keyword_search(query))

    # 能走到这里说明两个任务都已完成；结果按业务规则合并去重。
    return list(dict.fromkeys(vector_task.result() + keyword_task.result()))

如果用 asyncio.create_task() 启动后台任务却不保存引用、不等待结果，也没有关闭策略，异常可能丢失，任务还可能在请求结束后继续占用资源。

超时与取消必须向下传播

# 文件位置：src/note_api/model_client.py
import asyncio
from collections.abc import Awaitable, Callable

async def call_with_deadline(
    operation: Callable[[], Awaitable[str]],
    timeout_seconds: float,
) -> str:
    try:
        # timeout 为当前作用域设置截止时间，超时后会取消内部等待。
        async with asyncio.timeout(timeout_seconds):
            return await operation()
    # except 捕获指定异常；as error 把本次异常对象保存到 error。
    except TimeoutError as error:
        # 转成领域可识别的错误，同时保留原始异常链。
        raise RuntimeError("model request timed out") from error

取消是协作式的：协程必须执行到可等待点，才能观察到取消。捕获 asyncio.CancelledError 后通常应完成必要清理并继续抛出，不能静默吞掉，否则上层会误以为任务仍可继续。

并发上限保护下游

# 文件位置：src/note_api/batch.py
import asyncio
from collections.abc import Awaitable, Callable

async def map_limited(
    items: list[str],
    worker: Callable[[str], Awaitable[str]],
    limit: int,
) -> list[str]:
    if limit < 1:
        raise ValueError("limit must be positive")

    semaphore = asyncio.Semaphore(limit)  # 最多允许 limit 个任务进入下游调用。

    async def run_one(item: str) -> str:
        async with semaphore:
            return await worker(item)  # 退出作用域时自动释放并发名额。

    # gather 保持结果顺序；任一异常默认会传播给调用方。
    # * 展开生成器产生的协程，使 gather 把它们作为多个参数并发等待。
    return await asyncio.gather(*(run_one(item) for item in items))

并发上限应结合下游限流、数据库连接池和本机资源设置。一次创建十万个 Task，即使同时只运行十个，也会产生大量任务对象；超大输入还应使用有界队列分批生产。

线程池：兼容阻塞库

# 文件位置：src/note_api/file_reader.py
import asyncio
from pathlib import Path

def read_text_sync(path: Path) -> str:
    # Path.read_text 是阻塞文件 I/O，函数本身保持同步语义。
    return path.read_text(encoding="utf-8")

async def read_text(path: Path) -> str:
    # 把阻塞调用交给线程，事件循环可以继续处理其他任务。
    return await asyncio.to_thread(read_text_sync, path)

取消等待 to_thread 的协程，并不能强行终止已经开始运行的底层线程函数。因此阻塞操作自身仍应支持超时，长时间或不可控任务更适合进程或外部 Worker。

进程池：CPU 密集与隔离

# 文件位置：scripts/count_tokens.py
from concurrent.futures import ProcessPoolExecutor

def count_characters(text: str) -> int:
    # 示例代表纯 CPU 函数；真实项目可能是解析或图像预处理。
    return sum(1 for _ in text)

def main() -> None:
    documents = ["Python", "Go", "Agent"]
    # with 在离开代码块时自动关闭进程池，即使任务执行失败也会清理子进程。
    with ProcessPoolExecutor() as pool:
        # 参数和返回值需要能被进程间序列化。
        counts = list(pool.map(count_characters, documents))
    print(counts)

if __name__ == "__main__":
    # 子进程需要能够安全导入主模块，入口保护不能省略。
    main()

多进程不共享普通 Python 对象，每个进程还会有独立解释器和内存空间。大对象频繁跨进程序列化可能抵消并行收益，模型服务通常采用常驻 Worker，而不是每个请求临时创建进程池。

常见失败模式

现象	常见原因	修复方向
异步接口并发一高就整体卡顿	async def 内调用同步 SDK 或 CPU 重任务	换异步库、受控线程池或任务 Worker
请求取消后下游仍持续计费	没有向下传递取消，或 SDK 不支持取消	传递超时与取消信号，设置客户端超时
内存不断增长	无上限 Task、队列或结果列表	使用 TaskGroup、有界队列和背压
多进程在本地能跑，部署后失败	把不可序列化对象传给 Worker，或入口无保护	只传简单数据，在进程内创建资源
加并发反而更慢	下游连接池或限流成为瓶颈	从端到端指标确定合理并发数

高频面试题与回答

1. 进程、线程和协程有什么区别？参考答案

我把进程记成房间，线程记成工人，协程记成任务单。不同进程各有一份内存，隔离好但更重；同一进程里的线程共享内存，更轻但要保护共享数据；协程运行在线程中，等待 I/O 时通过 await 暂停，让线程处理其他协程。实际选择时，纯 Python 大量计算用多进程，阻塞 I/O 用线程，提供异步接口的大量 I/O 用协程。

2. 线程、多进程和 asyncio 怎样选择？参考答案

我先判断任务是在等，还是在算。大量纯 Python 计算属于“算”，使用多进程或原生计算库；网络、文件等 I/O 属于“等”，依赖提供异步接口就用 asyncio 协程，只有同步阻塞接口就放进受控线程池。最后还要为它们设置超时、取消、并发上限和关闭策略。

3. `async def` 是否一定比同步函数快？参考答案

不是。异步的价值是等待 I/O 时让出执行权，不会让单次 CPU 计算自动变快。如果异步函数内部调用阻塞代码，还会卡住整个事件循环。是否有收益取决于调用链是否异步、并发量和实际瓶颈。

4. 为什么推荐 TaskGroup？参考答案

它让一组子任务的生命周期受词法作用域管理：离开作用域前统一等待，子任务失败时取消同组任务并向上报告。相比随手创建无人管理的后台 Task，更容易保证失败传播和资源清理。

接下来学什么

下一篇学习 Python 工程化、测试与 FastAPI，把语言与并发知识组合成一个可运行、可验证的后端入口。

参考资料

• Python asyncio Task 与 TaskGroup (opens new window)
• Python concurrent.futures (opens new window)
• Python threading (opens new window)
• Python multiprocessing (opens new window)