写过一些 Python 服务和脚本后我发现很多人包括早期的我一遇到要并发就凭感觉开线程请求慢了开线程、算得慢了也开线程跑起来才发现IO 任务确实快了可 CPU 密集的任务开了一堆线程反而更慢。问题不在并发本身而在没搞清楚三种模型各自解决什么问题。这篇把threading、multiprocessing、asyncio的边界、适用场景和高频踩坑一次讲清代码都可直接运行。一、前提GIL 决定了线程能做什么、不能做什么CPython 有一把全局解释器锁GIL同一个进程里同一时刻只有一个线程在执行 Python 字节码。这句话是后面所有选型的基础。对CPU 密集型任务大量纯 Python 计算多线程在 CPython 里无法真正并行多个线程轮流拿 GIL还要付切换开销结果往往是开了多线程速度不升反降。对IO 密集型任务网络请求、读写文件、查数据库线程在等待 IO 时会释放 GIL这段等待时间可以让给别的线程所以多线程、协程都能拿到明显的并发收益。想在 CPU 密集场景利用多核正统办法是多进程每个进程有独立解释器和 GIL另外 NumPy 这类 C 扩展在做底层计算时会主动释放 GIL属于特例。所以选型的第一刀永远是先判断任务到底是CPU 密集还是IO 密集。二、三种模型分别是什么1. threading线程配合同步阻塞式 API线程共享同一块内存上手最简单。实际工程里一般不直接Thread().start()而是用线程池import concurrent.futures as cf import time def fetch(url): time.sleep(1) # 模拟一次 1 秒的网络往返阻塞、会释放 GIL return fdone: {url} urls [fhttps://example.com/{i} for i in range(10)] start time.time() with cf.ThreadPoolExecutor(max_workers10) as pool: results list(pool.map(fetch, urls)) print(round(time.time() - start, 2)) # 约 1.0 秒而不是串行的 10 秒适合用的是requests、传统数据库驱动这类同步阻塞库任务以等待 IO 为主。注意有 GIL 不等于不用加锁。GIL 只保证单条字节码的原子性而counter 1在字节码层面是读值→相加→写回三步线程可能在中间被切换counter 0 def inc(): global counter for _ in range(100_000): counter 1 # 非原子操作多线程下最终值会小于预期凡是先读后写的复合操作、对共享可变状态的修改都要显式用threading.Lock保护。2. multiprocessing进程绕开 GIL 吃满多核多进程各自有独立的解释器和内存空间能真正跑在多个 CPU 核心上是 CPU 密集任务的标准答案import concurrent.futures as cf import math, os, time def heavy(n): # 纯 Python 的 CPU 密集计算 return sum(math.factorial(i) % 7919 for i in range(n)) if __name__ __main__: tasks [30_000] * os.cpu_count() start time.time() with cf.ProcessPoolExecutor() as pool: list(pool.map(heavy, tasks)) print(round(time.time() - start, 2), s)同样的heavy换成ThreadPoolExecutor你会发现耗时基本不变甚至更长——这就是 GIL 的直观体现。多进程的代价也要清楚进程启动比线程重任务太小、数量太多时创建和调度开销会盖过收益。进程间内存不共享传给子进程的参数和返回值都要能被pickle序列化跨进程传大对象很慢。跨平台尤其 Windows、macOS 的 spawn 启动方式必须写if __name__ __main__:保护否则子进程递归 import 主模块会反复创建进程甚至报错。3. asyncio单线程事件循环 协程扛海量 IO 连接asyncio不开多线程也不开多进程而是在一个线程里跑一个事件循环靠await在任务之间协作式切换。它特别适合连接数特别多、大部分时间都在等的场景比如爬虫、网关、长连接import asyncio, time async def fetch(i): await asyncio.sleep(1) # 模拟异步 IO等待时把控制权交还事件循环 return i async def main(): return await asyncio.gather(*(fetch(i) for i in range(10))) start time.time() asyncio.run(main()) print(round(time.time() - start, 2)) # 约 1.0 秒协程的切换只发生在await点没有线程切换和 GIL 争用所以单线程就能轻松挂起成百上千个等待中的任务。但它有一个极其关键的前提整条调用链都得是异步的。三、最容易踩的坑在协程里写阻塞代码协程是协作式调度事件循环本身只有一个线程。一旦某个协程里出现同步阻塞调用整个循环都会被卡住所有并发瞬间退化成串行import asyncio, time, requests async def bad_fetch(url): return requests.get(url, timeout5).status_code # 同步阻塞整个循环卡死 async def main(): await asyncio.gather(*(bad_fetch(u) for u in urls)) # 看起来并发实际一个个排队time.sleep、requests.get、同步的数据库驱动、任何纯 Python 的重计算都不能直接放进协程。正确做法有两种用对应的异步库aiohttp替代requests、异步数据库驱动替代同步驱动、asyncio.sleep替代time.sleep老的同步库一时换不掉就用loop.run_in_executor把阻塞函数丢进线程池事件循环继续调度别的协程import asyncio, concurrent.futures as cf import requests def get(url): return requests.get(url, timeout5).status_code async def main(urls): loop asyncio.get_running_loop() with cf.ThreadPoolExecutor(max_workers10) as pool: results await asyncio.gather( *(loop.run_in_executor(pool, get, u) for u in urls) ) return results这也是asyncio 线程池混合模型最常见的用法事件循环负责海量连接的调度少量阻塞调用外包给线程池。四、一张表做选型任务特征推荐方案关键原因大量网络/文件 IO且用的是异步库aiohttp、异步驱动asyncio单线程高并发连接数多时开销最低IO 密集但用的是同步阻塞库requests、传统驱动ThreadPoolExecutor等待 IO 时释放 GIL改造量最小CPU 密集计算、压缩、图像处理、数据加工ProcessPoolExecutor绕开 GIL真正利用多核既要海量 IO 并发又有少量同步阻塞调用asynciorun_in_executor事件循环调度阻塞活外包给线程池需要严格隔离、绕过崩溃影响多进程进程内存独立互不拖垮补充两个工程默认值Python 3.8 里ThreadPoolExecutor的max_workers默认是min(32, os.cpu_count() 4)ProcessPoolExecutor默认是os.cpu_count()。线程/进程不是越多越好还要考虑对方服务的限流、连接池大小和本机内存。五、结论我的判断顺序固定成三步先量再选用计时或 cProfile 确认瓶颈是 CPU 还是 IO别靠猜。IO 看库全链路异步就上asyncio还在用同步阻塞库就先用线程池改造最省事。CPU 上进程纯计算要吃多核就用进程池并记得__main__保护和参数可 pickle。并发不是银弹。很多时候脚本慢根因是外部接口、数据库索引或串行的重复请求而不是并发开得不够多。先定位瓶颈再按上面的边界选模型比一上来就堆线程、堆协程靠谱得多。
写过一些 Python 服务和脚本后我发现很多人包括早期的我一遇到要并发就凭感觉开线程请求慢了开线程、算得慢了也开线程跑起来才发现IO 任务确实快了可 CPU 密集的任务开了一堆线程反而更慢。问题不在并发本身而在没搞清楚三种模型各自解决什么问题。这篇把threading、multiprocessing、asyncio的边界、适用场景和高频踩坑一次讲清代码都可直接运行。一、前提GIL 决定了线程能做什么、不能做什么CPython 有一把全局解释器锁GIL同一个进程里同一时刻只有一个线程在执行 Python 字节码。这句话是后面所有选型的基础。对CPU 密集型任务大量纯 Python 计算多线程在 CPython 里无法真正并行多个线程轮流拿 GIL还要付切换开销结果往往是开了多线程速度不升反降。对IO 密集型任务网络请求、读写文件、查数据库线程在等待 IO 时会释放 GIL这段等待时间可以让给别的线程所以多线程、协程都能拿到明显的并发收益。想在 CPU 密集场景利用多核正统办法是多进程每个进程有独立解释器和 GIL另外 NumPy 这类 C 扩展在做底层计算时会主动释放 GIL属于特例。所以选型的第一刀永远是先判断任务到底是CPU 密集还是IO 密集。二、三种模型分别是什么1. threading线程配合同步阻塞式 API线程共享同一块内存上手最简单。实际工程里一般不直接Thread().start()而是用线程池import concurrent.futures as cf import time def fetch(url): time.sleep(1) # 模拟一次 1 秒的网络往返阻塞、会释放 GIL return fdone: {url} urls [fhttps://example.com/{i} for i in range(10)] start time.time() with cf.ThreadPoolExecutor(max_workers10) as pool: results list(pool.map(fetch, urls)) print(round(time.time() - start, 2)) # 约 1.0 秒而不是串行的 10 秒适合用的是requests、传统数据库驱动这类同步阻塞库任务以等待 IO 为主。注意有 GIL 不等于不用加锁。GIL 只保证单条字节码的原子性而counter 1在字节码层面是读值→相加→写回三步线程可能在中间被切换counter 0 def inc(): global counter for _ in range(100_000): counter 1 # 非原子操作多线程下最终值会小于预期凡是先读后写的复合操作、对共享可变状态的修改都要显式用threading.Lock保护。2. multiprocessing进程绕开 GIL 吃满多核多进程各自有独立的解释器和内存空间能真正跑在多个 CPU 核心上是 CPU 密集任务的标准答案import concurrent.futures as cf import math, os, time def heavy(n): # 纯 Python 的 CPU 密集计算 return sum(math.factorial(i) % 7919 for i in range(n)) if __name__ __main__: tasks [30_000] * os.cpu_count() start time.time() with cf.ProcessPoolExecutor() as pool: list(pool.map(heavy, tasks)) print(round(time.time() - start, 2), s)同样的heavy换成ThreadPoolExecutor你会发现耗时基本不变甚至更长——这就是 GIL 的直观体现。多进程的代价也要清楚进程启动比线程重任务太小、数量太多时创建和调度开销会盖过收益。进程间内存不共享传给子进程的参数和返回值都要能被pickle序列化跨进程传大对象很慢。跨平台尤其 Windows、macOS 的 spawn 启动方式必须写if __name__ __main__:保护否则子进程递归 import 主模块会反复创建进程甚至报错。3. asyncio单线程事件循环 协程扛海量 IO 连接asyncio不开多线程也不开多进程而是在一个线程里跑一个事件循环靠await在任务之间协作式切换。它特别适合连接数特别多、大部分时间都在等的场景比如爬虫、网关、长连接import asyncio, time async def fetch(i): await asyncio.sleep(1) # 模拟异步 IO等待时把控制权交还事件循环 return i async def main(): return await asyncio.gather(*(fetch(i) for i in range(10))) start time.time() asyncio.run(main()) print(round(time.time() - start, 2)) # 约 1.0 秒协程的切换只发生在await点没有线程切换和 GIL 争用所以单线程就能轻松挂起成百上千个等待中的任务。但它有一个极其关键的前提整条调用链都得是异步的。三、最容易踩的坑在协程里写阻塞代码协程是协作式调度事件循环本身只有一个线程。一旦某个协程里出现同步阻塞调用整个循环都会被卡住所有并发瞬间退化成串行import asyncio, time, requests async def bad_fetch(url): return requests.get(url, timeout5).status_code # 同步阻塞整个循环卡死 async def main(): await asyncio.gather(*(bad_fetch(u) for u in urls)) # 看起来并发实际一个个排队time.sleep、requests.get、同步的数据库驱动、任何纯 Python 的重计算都不能直接放进协程。正确做法有两种用对应的异步库aiohttp替代requests、异步数据库驱动替代同步驱动、asyncio.sleep替代time.sleep老的同步库一时换不掉就用loop.run_in_executor把阻塞函数丢进线程池事件循环继续调度别的协程import asyncio, concurrent.futures as cf import requests def get(url): return requests.get(url, timeout5).status_code async def main(urls): loop asyncio.get_running_loop() with cf.ThreadPoolExecutor(max_workers10) as pool: results await asyncio.gather( *(loop.run_in_executor(pool, get, u) for u in urls) ) return results这也是asyncio 线程池混合模型最常见的用法事件循环负责海量连接的调度少量阻塞调用外包给线程池。四、一张表做选型任务特征推荐方案关键原因大量网络/文件 IO且用的是异步库aiohttp、异步驱动asyncio单线程高并发连接数多时开销最低IO 密集但用的是同步阻塞库requests、传统驱动ThreadPoolExecutor等待 IO 时释放 GIL改造量最小CPU 密集计算、压缩、图像处理、数据加工ProcessPoolExecutor绕开 GIL真正利用多核既要海量 IO 并发又有少量同步阻塞调用asynciorun_in_executor事件循环调度阻塞活外包给线程池需要严格隔离、绕过崩溃影响多进程进程内存独立互不拖垮补充两个工程默认值Python 3.8 里ThreadPoolExecutor的max_workers默认是min(32, os.cpu_count() 4)ProcessPoolExecutor默认是os.cpu_count()。线程/进程不是越多越好还要考虑对方服务的限流、连接池大小和本机内存。五、结论我的判断顺序固定成三步先量再选用计时或 cProfile 确认瓶颈是 CPU 还是 IO别靠猜。IO 看库全链路异步就上asyncio还在用同步阻塞库就先用线程池改造最省事。CPU 上进程纯计算要吃多核就用进程池并记得__main__保护和参数可 pickle。并发不是银弹。很多时候脚本慢根因是外部接口、数据库索引或串行的重复请求而不是并发开得不够多。先定位瓶颈再按上面的边界选模型比一上来就堆线程、堆协程靠谱得多。
