Python 并发编程实操：从 threading 到 asyncio 的选择与避坑

在 Python 开发中，遇到“程序很慢”时，很多人会直接想到“加线程”或者“改异步”。但并发工具选错，反而会让程序更慢、更难调试。本文从任务类型出发，讲清楚 threading、multiprocessing 和 asyncio 各自适合什么场景，并给出可以直接运行的示例和常见的避坑清单。

一、先判断任务类型：I/O 密集还是 CPU 密集

这是选择并发模型的第一步，也是最重要的一步。

I/O 密集型任务的特点是程序大部分时间在等待：等待网络响应、等待数据库返回、等待文件读写。CPU 在等待期间是空闲的，这时让多个任务“轮流等待”，整体吞吐量就能明显提升。

CPU 密集型任务的特点是程序大部分时间在真正计算：数值计算、图像处理、加密解密、大量数据解析。CPU 已经满负荷工作，增加并发并不会凭空多出算力，关键在于把计算分摊到多个 CPU 核心上。

一个简单的判断方法：用系统工具看 CPU 占用率。如果单进程运行时 CPU 占用率很低、程序却很慢，多半是 I/O 密集；如果 CPU 占用率已经接近一个核心的上限，程序仍然不够快，多半是 CPU 密集。先判断类型，再选工具，不要一上来就写并发代码。

二、threading：适合 I/O 密集，写法最直观

threading 的优势是写法直观、与同步代码最接近，适合网络请求、文件操作、数据库访问这类 I/O 密集任务。多个线程共享同一进程的内存，启动成本低，数据传递也方便。

但 threading 有一个必须理解的前提：CPython 解释器有全局解释器锁（GIL），同一时刻只有一个线程能执行 Python 字节码。好在线程在等待 I/O 时会释放 GIL，所以 I/O 密集任务仍然可以从多线程中受益；而纯 Python 计算则基本无法靠多线程加速，这一点后面会再讲。

推荐使用 concurrent.futures.ThreadPoolExecutor，而不是手动创建大量 Thread 对象。线程池可以限制并发数量，避免一次创建上千个线程带来的内存和调度开销。下面是一个用线程池并发执行 I/O 型任务的示例，这里用 time.sleep 模拟等待：

import time
from concurrent.futures import ThreadPoolExecutor

def fetch(n):
    time.sleep(0.5)  # 模拟一次网络请求或数据库查询
    return f"任务 {n} 完成"

start = time.perf_counter()
with ThreadPoolExecutor(max_workers=8) as pool:
    results = list(pool.map(fetch, range(16)))
print(results)
print(f"耗时：{time.perf_counter() - start:.2f} 秒")

16 个任务如果串行执行需要约 8 秒，用 8 个线程并发后大约 1 秒就能完成。ThreadPoolExecutor 在 Python 3.8 及之后版本的默认最大线程数是 min(32, os.cpu_count() + 4)，一般不需要手动设置得很大。

使用多线程时要注意共享状态。多个线程同时修改同一个变量可能产生竞态条件，结果每次运行都不一样。正确做法是用 threading.Lock 保护共享数据，或者尽量让每个任务只处理自己的数据，最后再汇总：

import threading

counter = 0
lock = threading.Lock()

def add():
    global counter
    with lock:
        counter += 1

另外，主线程结束时，非守护线程会阻止进程退出。线程池的 with 语句会在退出时等待任务完成，这是推荐的写法；如果手动创建线程，记得调用 join 等待，或者明确设置 daemon 参数。

三、multiprocessing：适合 CPU 密集，绕开 GIL

CPU 密集任务应该用 multiprocessing。每个进程有独立的解释器和内存空间，不受 GIL 限制，可以真正利用多个 CPU 核心。代价是进程启动成本高，进程之间传递数据需要序列化，共享数据也比线程麻烦。

同样推荐使用 ProcessPoolExecutor，写法与线程池非常接近，便于在两种方案之间切换。下面是一个把计算分块交给多个进程的示例：

from concurrent.futures import ProcessPoolExecutor

def partial_sum(numbers):
    return sum(n * n for n in numbers)

def main():
    data = list(range(1_000_000))
    chunk_size = 250_000
    chunks = [data[i:i + chunk_size] for i in range(0, len(data), chunk_size)]
    with ProcessPoolExecutor(max_workers=4) as pool:
        results = list(pool.map(partial_sum, chunks))
    print(sum(results))

if __name__ == "__main__":
    main()

注意 if name == “main”: 这一行保护。在 Windows 和 macOS 的 spawn 启动方式下，子进程会重新导入主模块，没有这层保护会导致子进程重复创建进程池，甚至无限递归。在 Linux 上默认也可能使用 spawn，养成加保护的习惯没有坏处。

进程池不适合传递很大的数据，也不适合任务本身很轻、数量极多的场景，因为序列化和进程调度的开销可能超过计算本身的收益。判断标准仍然是实测：先用小规模数据对比串行和并行的耗时，再决定是否上规模。

四、asyncio：适合大量 I/O 并发，单线程协作式调度

当 I/O 并发量很大，例如同时维护成百上千个网络连接时，asyncio 往往是更合适的选择。它在单线程内用事件循环调度协程，切换成本低，不需要为每个任务分配一个线程。

asyncio 的核心写法是 async 和 await。await 表示“在这里等待，但把控制权交还给事件循环，让其他任务先跑”。下面是一个并发等待的示例：

import asyncio
import time

async def fetch(n):
    await asyncio.sleep(0.5)  # 模拟一次异步 I/O 等待
    return f"任务 {n} 完成"

async def main():
    start = time.perf_counter()
    results = await asyncio.gather(*(fetch(n) for n in range(16)))
    print(results)
    print(f"耗时：{time.perf_counter() - start:.2f} 秒")

asyncio.run(main())

asyncio 最常见的坑，是在异步函数里调用同步阻塞代码。例如在 async 函数里写 time.sleep(1) 或者同步的 requests 请求，会阻塞整个事件循环，期间所有协程都无法运行，异步的优势就完全丧失了。正确做法是使用异步 I/O 库，或者把同步调用放到线程里执行：

import asyncio

def blocking_work():
    # 这里是同步的耗时操作，例如旧的同步 SDK 调用
    return "完成"

async def main():
    result = await asyncio.to_thread(blocking_work)
    print(result)

asyncio.run(main())

asyncio.to_thread 是在 Python 3.9 引入的，用于把同步函数放到线程中执行，避免阻塞事件循环。

并发量过大时，还要用 asyncio.Semaphore 限制同时进行的任务数，避免一次性发出太多请求把对方服务或者本地连接数打满：

semaphore = asyncio.Semaphore(10)

async def limited_fetch(n):
    async with semaphore:
        return await fetch(n)

五、如何选择：一张决策清单

把前面的结论汇总成可直接使用的决策流程：

1. 任务主要是等待网络、数据库、文件，数量不多（几十个以内）：优先用 threading 或者 ThreadPoolExecutor，改动最小。
2. 任务主要是等待网络、数据库、文件，数量很大（成百上千）：优先用 asyncio，配合异步客户端和信号量控制并发。
3. 任务主要是纯 Python 计算，需要多核加速：用 multiprocessing 或者 ProcessPoolExecutor，并注意数据传递成本。
4. 既有大量 I/O，又有少量计算：以 asyncio 为主，计算部分用 asyncio.to_thread 或者 run_in_executor 交给线程池和进程池。
5. 不确定类型：先写串行版本并测量，再用小规模数据对比两种方案的耗时，不要凭感觉选择。

六、常见坑：这些错误最容易在生产环境暴露

第一，把 GIL 理解成“Python 不能并发”。GIL 限制的是同一时刻执行 Python 字节码的线程数，I/O 等待时会释放锁，C 扩展（例如 NumPy 的部分操作）也可能释放锁。I/O 密集任务用多线程是有效的，纯计算任务才需要多进程。

第二，共享状态没有保护。计数器、列表、字典在多线程下都可能出现竞态。用锁、队列或者“各自计算、最后汇总”的方式处理共享数据。

第三，异常被悄悄吞掉。ThreadPoolExecutor 和 ProcessPoolExecutor 返回的 Future 只有在调用 result() 时才会抛出任务中的异常；asyncio.gather 默认会在第一个异常时传播，但如果设置了 return_exceptions=True，异常会作为结果返回，需要逐个检查。

第四，没有超时和取消。网络请求可能 hang 住，任务可能永远等下去。给关键步骤设置超时，例如 asyncio.wait_for，或者在 Future 上使用 timeout 参数，并记录超时日志，方便排查。

第五，资源没有释放。线程池和进程池要用 with 语句管理，网络连接和文件句柄要及时关闭。程序退出时还在运行的后台线程和未关闭的连接，是线上常见的泄漏来源。

七、结语

Python 并发没有银弹，threading、multiprocessing 和 asyncio 各自解决不同类型的问题。先判断任务是 I/O 密集还是 CPU 密集，再看并发规模，最后用实测数据验证。写并发代码时，把共享状态、异常、超时和资源释放这四件事处理好，程序的可维护性会比单纯追求“并发数”重要得多。