Python 并发编程实战:线程、进程与协程的选择策略

Python 的并发编程有三条路：threading（多线程）、multiprocessing（多进程）、asyncio（异步协程）。看起来都能"同时做多件事"，但它们的适用场景差异极大。选错模型，代码写完了才发现性能不升反降——这类案例在 code review 里太常见了。

本文从 GIL 出发，逐一拆解三种并发模型的底层机制、适用边界和代码示例，最后给出一张明确的选型决策表。

一、GIL：绕不开的话题

GIL（Global Interpreter Lock，全局解释器锁）是 CPython 解释器的一个设计决策：同一时刻，只有一个线程能执行 Python 字节码。

这意味着什么？即使你开了 8 个线程跑在 8 核 CPU 上，Python 代码的执行仍然是串行的。多线程在 CPU 密集型任务上不会加速，反而因为线程切换的开销而更慢。

import threadingimport timedef cpu_bound(n):    total = 0    for i in range(n):        total += i ** 2    return totalstart = time.perf_counter()cpu_bound(10_000_000)print(f"单线程: {time.perf_counter() - start:.3f}s")start = time.perf_counter()threads = [threading.Thread(target=cpu_bound, args=(2_500_000,)) for _ in range(4)]for t in threads:    t.start()for t in threads:    t.join()print(f"4线程: {time.perf_counter() - start:.3f}s")   # 大概率比单线程更慢

但 GIL 有一个关键例外：当线程在等待 IO（网络请求、磁盘读写、数据库查询）时，GIL 会被释放，其他线程可以继续执行。这正是多线程在 IO 密集型场景下仍然有效的根本原因。

二、threading：IO 密集场景的首选

2.1 基本用法

import threadingimport requestsfrom concurrent.futures import ThreadPoolExecutorurls = ["https://httpbin.org/delay/1"] * 10def fetch(url):    resp = requests.get(url, timeout=10)    return resp.status_codewith ThreadPoolExecutor(max_workers=5) as executor:    results = list(executor.map(fetch, urls))print(f"完成 {len(results)} 个请求")

2.2 线程安全与锁

多线程共享内存空间，写操作必须加锁：

counter = 0lock = threading.Lock()def increment():    global counter    for _ in range(100_000):        with lock:            # 等同于 lock.acquire() / lock.release()            counter += 1

counter += 1 看起来是原子操作，但 Python 字节码层面它分成多条指令：读取 → 加 1 → 写入。不加锁时线程 A 读到旧值、线程 B 也读到旧值，各自加 1 写回，结果少算一次。

另一种选择是 queue.Queue——线程安全的队列，天然适合生产者-消费者模式：

from queue import Queueimport threadingdef producer(q):    for i in range(100):        q.put(i)def consumer(q):    while True:        item = q.get()        if item is None:            break        process(item)

2.3 适用场景

• 网络爬虫 / API 批量调用
• 文件批量读写
• 数据库并发查询
• GUI 程序中让后台任务不阻塞界面

三、multiprocessing：CPU 密集场景的解法

换用进程绕过了 GIL：每个进程有独立的 Python 解释器和内存空间，真正利用多核 CPU。

from concurrent.futures import ProcessPoolExecutorimport timedef cpu_bound(n):    total = 0    for i in range(n):        total += i ** 2    return totalstart = time.perf_counter()with ProcessPoolExecutor(max_workers=4) as executor:    results = list(executor.map(cpu_bound, [2_500_000] * 4))print(f"4进程: {time.perf_counter() - start:.3f}s")   # 接近线性加速

代价也很明显：

• 创建进程的开销远大于线程。
• 进程间内存不共享，数据传递需要序列化（pickle），大对象传输成本高。
• 在 Windows 上子进程会重新导入主模块，需要用 if __name__ == "__main__": 保护入口代码。

3.1 进程间通信

from multiprocessing import Process, Queuedef worker(q):    q.put({"result": 42})if __name__ == "__main__":    q = Queue()    p = Process(target=worker, args=(q,))    p.start()    result = q.get()   # {"result": 42}    p.join()

对于大量数据的共享，multiprocessing.shared_memory（Python 3.8+）提供了更高效的方式，避免 pickle 序列化的开销。

四、asyncio：高并发 IO 的现代方案

asyncio 的核心思想是协作式多任务：单个线程内，通过 await 显式交出控制权，让事件循环在同一线程中调度多个协程。

import asyncioimport aiohttpimport timeasync def fetch(session, url):    async with session.get(url) as resp:        return await resp.text()async def main():    urls = ["https://httpbin.org/delay/1"] * 50    async with aiohttp.ClientSession() as session:        tasks = [fetch(session, url) for url in urls]        results = await asyncio.gather(*tasks)    return resultsstart = time.perf_counter()asyncio.run(main())print(f"50个请求耗时: {time.perf_counter() - start:.3f}s")   # 约 1-2 秒

50 个各延迟 1 秒的请求，同步顺序执行需要 50 秒，asyncio 并发只需约 1-2 秒（取决于带宽和连接池大小）。这是 asyncio 的核心优势：以极低的线程开销处理海量 IO 并发。

4.1 async/await 的使用边界

async def fetch_data():    data = await db.query("SELECT ...")    return dataasyncio.run(fetch_data())def main():    data = await fetch_data()   # SyntaxError

4.2 同步代码与异步代码的桥接

实际项目中不可能所有代码都是 async。在异步上下文中调用同步函数：

import asynciofrom concurrent.futures import ThreadPoolExecutorexecutor = ThreadPoolExecutor(max_workers=4)async def main():    loop = asyncio.get_running_loop()    result = await loop.run_in_executor(executor, blocking_io_function)

反过来，在同步代码中调用异步函数用 asyncio.run()，但注意不要嵌套——asyncio.run() 不能在一个已有事件循环的线程中再次调用。

五、并发模型选型决策

六、一个混合场景的实战示例

假设一个任务：从多个 API 拉数据（IO 密集），对每条数据做 CPU 密集的解析：

import asynciofrom concurrent.futures import ProcessPoolExecutordef heavy_parse(raw_text: str) -> dict:    """CPU 密集的解析逻辑。"""    return parsedasync def fetch_and_parse(session, url, executor):    async with session.get(url) as resp:        raw = await resp.text()    loop = asyncio.get_running_loop()    return await loop.run_in_executor(executor, heavy_parse, raw)async def main(urls):    executor = ProcessPoolExecutor(max_workers=4)    async with aiohttp.ClientSession() as session:        tasks = [fetch_and_parse(session, url, executor) for url in urls]        results = await asyncio.gather(*tasks)    executor.shutdown()    return results

这个模式综合了 asyncio 的高并发 IO 能力和 multiprocessing 的 CPU 并行能力，适合数据管道、ETL、批量 API 处理等场景。

最后

Python 并发选型的关键是看清楚任务的性质：瓶颈在 IO 还是 CPU？并发规模是百级还是万级？现有代码是同步还是异步？

几条原则：

• 先 profile，再优化。用 cProfile 确认瓶颈到底在哪里，不要凭直觉选并发模型。
• 能用 concurrent.futures 统一接口就别裸写 Thread/Process——代码量少 50%，出错率更低。
• asyncio
   能力很强，但它是一种"传染性"的编程模型——一旦选了 async，调用链上下游都得 async。评估改造成本后再决定。