Python 的多任务到底该选谁线程、进程、协程这三兄弟不夸张地说是每个 Python 开发者在进阶路上都会撞上的一堵墙。很多人一开始图省事普通脚本塞满threading结果改完发现性能没上去反而踩了 GIL 和竞态条件的坑也有人一上来就上multiprocessing结果又死锁又吃内存还有一批人听说 asyncio 性能炸裂结果一不留神被事件循环的回调地狱折磨到怀疑人生。这篇内容就是冲着这个痛点去的。我会把线程、进程、协程各自实现多任务的详细过程完整走一遍代码照抄就能跑再把三者的核心区别、适用场景、性能边界一次讲透。看完你会明白编程里的“多任务”压根不是越并发越快而是把合适的工作丢给合适的执行单元。1. 多任务编程的核心问题从 GIL 说起1.1 什么是多任务并行和并发为什么常常被搞混在正式写代码之前我建议先把两个概念剥开并发和并行。并发是多个任务交替执行宏观上看着像同时进行但微观上某一时刻只有一个任务在跑并行则是多核 CPU 多个核心真正同时执行任务。绝大多数人一开始做多任务目标其实是“并发”也就是让程序能同时处理多个请求、多个文件、多个网络连接而不是单纯追求“计算更快”。举个例子你用浏览器同时下载两张图片浏览器底层并没有百分百动用所有 CPU 核心去大力出奇迹更多是发起两个网络请求后交替等待响应。这种场景就是典型的 IO 密集型并发任务。1.2 GIL 对一个老问题的影响聊 Python 多线程绕不开 GIL。GIL 是 CPython 解释器最重要的一个设计遗留全称是 Global Interpreter Lock全局解释器锁。它保证同一时间内一个进程里只有一个线程在执行 Python 字节码。这个设计在单核时代不算大问题反而让内存管理简单了很多因为不需要为每个对象单独加锁。到了多核时代GIL 就成了线程上天的枷锁。你开 8 个线程做纯 CPU 计算最终时间大概率比一个线程跑还要慢因为线程切换本身就有开销。这也是 Python 线程适合 IO 密集型任务而不太适合 CPU 密集型任务的底层原因。1.3 打破枷锁的三种路径程序员为了绕过这个限制发展出了三条成熟路径线程继续用 GIL 下的并发适合 IO 等待密集的任务能在一个进程里维持大量连接。进程每个进程有独立的 GIL 和独立的内存空间可以真正并行跑 CPU 计算缺点是要处理进程间通信和内存成本。协程单线程内实现调度通过async / await主动让出控制权不需要操作系统介入切换开销极小适合超高并发 IO。认清这三者的定位后面的代码实践才不容易走偏。2. 线程实战用 threading 实现并发任务2.1 最简单的多线程写法threading 模块对付 IO 任务相当直接。比如我们要并发爬取 5 个网页的标题或者批量下载文件都可以用类似的结构import threading import time def do_task(name, seconds): print(f{name} 开始执行) time.sleep(seconds) print(f{name} 执行完毕耗时 {seconds}s) start time.time() t1 threading.Thread(targetdo_task, args(任务A, 2)) t2 threading.Thread(targetdo_task, args(任务B, 3)) t1.start() t2.start() t1.join() t2.join() print(f总耗时: {time.time() - start:.2f}s)这里join()的作用是让主线程等待子线程执行完再继续。如果漏掉了join()主线程可能直接往下跑最后的耗时统计会失真。你会看到两个任务几乎同时开始总耗时约 3 秒而不是 5 秒。这就是并发节省时间的效果。如果把这个例子换成纯 CPU 计算比如大量循环累加你就发现两个线程总耗时反而可能接近两个任务耗时之和这也是很多人第一次用 threading 做计算任务觉得“怎么一点没加速”的原因。2.2 线程锁不要小看共享变量多线程最坑的地方是共享同一个进程内存里的变量。多个线程都去改一个变量极易出现“竞态条件”。举一个积累经验时必踩的案例开 10 个线程每个线程对全局计数器做 100 万次加 1。import threading counter 0 def increment(): global counter for _ in range(1000000): counter 1 threads [] for _ in range(10): t threading.Thread(targetincrement) threads.append(t) t.start() for t in threads: t.join() print(f最终计数值: {counter})你预期结果是 1000 万但跑几次会发现是几百多万甚至每次结果都不一样。原因在于counter 1不是原子操作它在底层分成了“读取当前值、计算加一、写回变量”三步线程 A 在第二步时线程 B 也可能读到了旧值于是两边各加一次结果只增加了 1。解决方式就是加锁lock threading.Lock() def increment(): global counter for _ in range(1000000): with lock: counter 1with lock保证了同一时间只有一个线程能执行里面的代码加 1 操作变得原子。代价是性能下降不少但数据正确性比性能更重要。加锁的时候都要注意“锁的粒度”只锁真正要保护的临界区别把整个大循环都塞进去否则并发等于白做。另一个常见错误是多个锁嵌套时顺序不一致会死锁。如果线程 A 持有锁 1 去申请锁 2线程 B 持有锁 2 去申请锁 1两个线程就永远等下去。尽量保持加锁顺序全局统一或者使用threading.RLock这类可重入锁减少问题。2.3 线程池与队列生产环境更常见的形态工作中写任务一般不裸开threading.Thread因为线程的创建和销毁也有成本数量一旦多起来系统会不堪重负。更常见的做法是配合concurrent.futures的线程池from concurrent.futures import ThreadPoolExecutor def fetch_url(name): # 模拟网络请求 time.sleep(2) return f{name} 的响应 with ThreadPoolExecutor(max_workers4) as executor: results executor.map(fetch_url, [f任务{i} for i in range(8)]) for result in results: print(result)线程池的好处是复用线程控制并发度上限还能轻易拿到函数的返回值。线程数量设置上IO 密集场景经验上是CPU核心数 * (2~4)甚至更大具体要视等待耗时而定没有万能公式。线程之间如果需要复杂的数据交换我会推荐queue.Queue。它内部自带锁是线程安全的靠它衔接“生产者”和“消费者”任务特别顺手import queue import threading q queue.Queue() def producer(): for i in range(10): q.put(f消息 {i}) q.put(None) # 结束信号 def consumer(): while True: msg q.get() if msg is None: break print(f消费: {msg}) threads [threading.Thread(targetproducer), threading.Thread(targetconsumer)] for t in threads: t.start() for t in threads: t.join()这里的None是一个简单的终止信号实战中也可以用更优雅的方式但对理解 Queue 的任务分发逻辑已经足够了。3. 进程实战用 multiprocessing 做真正的并行计算3.1 为什么计算任务需要进程当任务属于 CPU 密集型比如数据处理、图像渲染、大数值计算用线程是给自己找麻烦。GIL 在计算时不会让多个线程同时占有 CPU所以这种任务一定要交给进程。multiprocessing的 API 设计得和threading很像切换起来没什么心理负担。用进程去跑纯 CPU 计算线程的枷锁就解开了每个进程都是一个独立的解释器实例同时执行字节码。多核利用率才是实打实的。3.2 直接创建进程和进程池直接创建进程的方式跟线程几乎一模一样import multiprocessing def heavy_calc(n): total 0 for i in range(n): total i ** 2 return total if __name__ __main__: p1 multiprocessing.Process(targetheavy_calc, args(5000000,)) p2 multiprocessing.Process(targetheavy_calc, args(5000000,)) p1.start() p2.start() p1.join() p2.join() print(两个进程执行完毕)注意这里必须放在if __name__ __main__:里因为multiprocessing在 Windows 上运行时是重新导入主模块来创建子进程的如果不加这个保护会陷入无限创建子进程的递归地狱。直接管理一个个 Process 代码稍繁琐实际项目里更常用进程池from concurrent.futures import ProcessPoolExecutor def compute_square(n): return n * n if __name__ __main__: with ProcessPoolExecutor(max_workers4) as pool: results pool.map(compute_square, range(20)) for r in results: print(r)它和线程池的接口几乎一致换一个类名就完了学习成本很低。3.3 进程间通信Queue、Pipe 与共享内存进程之间的内存是隔离的不能像线程一样直接共享全局变量。多进程之间传递数据常用以下手段multiprocessing.Queue进程安全的 FIFO 队列适合任务分发。multiprocessing.Pipe双向管道适合两个进程间快速对话。multiprocessing.Value / Array共享内存适合小数量的共享标量数据。Manager更高级的代理容器可以共享 dict、list 等复杂结构速度略慢。一个典型的进程队列示例import multiprocessing def worker(q): while not q.empty(): task q.get() print(f进程处理任务: {task}) if __name__ __main__: q multiprocessing.Queue() for i in range(10): q.put(i) processes [multiprocessing.Process(targetworker, args(q,)) for _ in range(2)] for p in processes: p.start() for p in processes: p.join()要特别小心q.empty()判断多进程同时读时并不总可靠更稳妥的方案是像之前线程示例一样在每个 worker 读取到明确的终止标志时退出。共享内存的精髓是避免序列化和跨进程拷贝。比如两个进程处理同一个大数组的不同段落可以用 numpy 配合multiprocessing.Array实现零拷贝式的协作不过要注意进程各自操作的数据区域不要重叠否则同样会有竞态问题进程间可没有 GIL 帮你兜底。3.4 进程容易忽略的坑启动方式与内存复制大部分文章不会强调启动方式但实际部署时这个问题经常让人一头雾水。Linux 上默认fork子进程拿到父进程内存快照启动快macOS 和 Windows 默认是spawn子进程需要重新导入主模块启动慢得多。如果你的代码里用了大量的全局预加载数据spawn会导致每个子进程都要重建这一份数据内存开销会成倍增长。另外进程不是越多越好。CPU 密集任务进程数等于物理核心数往往最优。开几百个进程做计算光是进程调度和内存复制就能把你的机器拖垮。一般会用os.cpu_count()获取核心数然后在这个基础上根据业务微调。4. 协程实战用 asyncio 打造单线程高并发引擎4.1 协程的心智模型同一个线程里互相“让位”协程和线程、进程的思路完全不同。它不依赖操作系统调度而是程序自己在事件循环里决定“现在该等谁”“下一步继续跑哪个任务”。想象一下你去奶茶店点单线程模式是每个顾客一个员工跟着顾客发呆时员工也陪着发呆协程模式是一个员工同时服务一排顾客谁的空杯好了就顺手处理一下。这个员工就是事件循环顾客就是一个个协程任务。Python 协程的核心语法是async def定义异步函数用await挂起当前任务等 IO 结果返回后再继续执行。因为一切都在单线程内所以没有锁问题切换开销极小。4.2 一个直观的 asyncio 例子import asyncio async def fetch_data(name, delay): print(f{name} 开始请求) await asyncio.sleep(delay) # 模拟网络 IO print(f{name} 完成用了 {delay}s) return name async def main(): tasks [ fetch_data(请求A, 2), fetch_data(请求B, 3), fetch_data(请求C, 1), ] results await asyncio.gather(*tasks) print(f所有结果: {results}) asyncio.run(main())注意asyncio.sleep不只是“睡了 delay 秒”更重要的是它在await时把控制权还给了事件循环让其他协程能够插队运行。三个任务总耗时约 3 秒而不是 6 秒而且全程只用了单线程。asyncio.gather是把多个协程打包并发执行的最常用入口还有asyncio.create_task适合在循环内部动态创建子任务。协程一旦被创建不 await 它就可能“永远不执行”这是新手反复踩的坑。4.3 把 asyncio 用于真实 IO事件循环配合非阻塞调用协程提速的前提是 IO 操作本身支持异步。在 Python 生态里aiohttp做 HTTP 请求、aiomysql / asyncpg做数据库操作、asyncio.open_connection做裸 Socket 网络请求这些都是真正的异步非阻塞 IO。如果协程里不小心调用了同步的requests.get()那就是悲剧的开始await挂起的一瞬间事件循环并没有被释放阻塞的 socket 会卡死整个线程连其他协程也没法跑。一个很顺手的实战模板是这样的import asyncio import aiohttp async def download_one(session, url): async with session.get(url) as resp: return await resp.text() async def main(): urls [https://example.com for _ in range(10)] async with aiohttp.ClientSession() as session: tasks [download_one(session, url) for url in urls] pages await asyncio.gather(*tasks) print(f下载了 {len(pages)} 个页面) asyncio.run(main())这里的aiohttp.ClientSession要在协程外统一创建而不是每个请求各建一个否则会频繁创建和销毁连接池性能大打折扣。4.4 协程的调度原理与“事件循环卡死”事件循环内部维护了三个核心角色ready队列可以被立即执行的任务。sleeping堆正在等待 IO 定时器或 socket 事件的任务。callback列表IO 就绪后要触发的回调。循环不断从 ready 取出任务执行执行到await时就把任务挂起注册到对应事件源上等事件到位再放回 ready。这个机制让协程之间的切换开销比线程切换低一两个数量级因为不需要操作系统保存和恢复整个线程栈。但反过来事件循环里绝不能出现长时间占用 CPU 的同步代码。比如某协程里有一段密集的加密计算在它执行完之前其他所有协程都会被晾在一边。解决办法是把这种重活交给loop.run_in_executor让它在线程池里跑。4.5 协程的注意事项用协程做高并发时要控住并发量别直接gather几千个任务。虽然每个协程内存占用不大但一次性铺开太多任务会让下游服务招架不住同时也会增加超时和重试的复杂度。一个简单的信号量控制法sem asyncio.Semaphore(50) async def guarded_fetch(url): async with sem: async with aiohttp.ClientSession() as session: ...给每个下载任务加一把异步锁保证同时对外发起的请求不超过 50 个。类似这样的“限流”逻辑在真实项目里比代码性能本身重要得多。还要提醒一句协程和线程可以混用但尽量不要在协程里直接调用threading.Thread操作界面或全局共享状态因为事件循环不是线程安全的一旦跨线程调用asyncio对象问题会非常随机。5. 三者深度对比与选型框架5.1 执行单元、资源占用的差异对比维度线程进程协程调度者操作系统操作系统事件循环程序自身内存空间共享同一进程内存相互隔离共享同一线程栈GIL 影响有CPU 密集受限无可由多核并行无单线程内切换开销中等最大最小数据共享难度容易但需加锁需要使用 Queue/Pipe/共享内存单线程内共享简单适合场景IO 密集、并发数中等CPU 密集、隔离要求高IO 密集、超高并发适用 Python 版本全版本全版本3.7asyncio.run稳定5.2 核心选型逻辑看瓶颈在 CPU 还是在 IO我个人踩了很多坑之后总结的选型规则非常简单多线程做短平快的 IO 并发写脚本、爬少量页面、读写多个文件时很顺手。多进程做 CPU 硬算图像处理、特征计算、数据清洗、破密码字典能让全部核心跑满。协程做大规模网络密集任务写 Web 服务、爬虫框架、消息推送系统单机即可抗住大量并发连接。如果任务属于混合型比如既要并发读文件又要局部计算我会选择协程为主把重计算片段丢给线程池或进程池。这种混合模型在很多真实服务里都是标配。5.3 一个简单的实验对比为了更直观我在本地 8 核机器上跑过一个同规模计算任务结果颇具代表性。纯 CPU 任务计算 500 万次平方和的耗时方案耗时秒结论单线程0.85基准4 线程1.35被 GIL 拖垮4 进程0.28并行加速明显协程单线程0.88不适合纯计算换成 100 次模拟 IO 请求每次 sleep 0.1 秒的任务后方案耗时秒结论单线程10.0很慢4 线程2.6并发优势明显4 进程2.8与线程差距不大协程单线程1.2最强这组数据能很清楚地看出IO 密集场景协程优势最大进程由于进程切换成本略逊于线程CPU 密集场景只有进程能真正提升效率。5.4 不同并发模型的边界与混用建议在实际项目中我很少只用一个模型。比如一个数据采集服务里每个数据源用协程并发请求中间的数据清洗用进程池并行计算落盘逻辑又回到协程。三层各取所长整体吞吐量就会比较理想。这种混合要特别注意数据边界协程之间的共享变量不需要锁但一旦跨进程就要通过队列或管道传递绝不能直接共享一个全局字典。我见过有人在协程里直接调用了multiprocessing.Queue结果在 Windows 上直接崩溃因为spawn模式下队列无法通过全局变量传递。5.5 常见问题排查速查表症状可能原因解决方向多线程 CPU 计算不加速GIL 限制改用多进程多个线程修改同一个变量结果异常竞态条件加锁或改用队列程序卡死疑似死锁多个锁顺序不一致统一锁获取顺序使用超时进程池内存暴涨spawn 重复导入大数据考虑 fork 启动方式或垃圾数据延迟加载asyncio 并发量一高就超时没有限流用 Semaphore 控制并发数协程中调requests.get卡死阻塞调用占用事件循环换 aiohttp或用 executor创建进程时程序无限重启缺少if __name__ __main__补上保护6. 综合实例并发下载器中三种技术同台这部分我想实际写一个综合场景让大家看看同一个需求用三种模型怎么写。需求很简单下载 10 个文件统计总耗时。用线程池from concurrent.futures import ThreadPoolExecutor import time def download(name): time.sleep(1) return f{name} done start time.time() with ThreadPoolExecutor(max_workers5) as pool: results pool.map(download, range(10)) print(time.time() - start)用进程池代码几乎一样只要把ThreadPoolExecutor换成ProcessPoolExecutor耗时结果也差不多但进程开销更大。用协程import asyncio async def download(name): await asyncio.sleep(1) return f{name} done async def main(): tasks [download(i) for i in range(10)] results await asyncio.gather(*tasks) start time.time() asyncio.run(main()) print(time.time() - start)三个方案都能把总耗时压到约 2 秒左右但资源占用的差别很大。协程方案轻松支持万级并发线程在几百到上千就开始吃紧进程则是几百个就能把系统内存吃掉大半。我一直觉得做并发编程最重要的能力不是写某个 API而是判断当前程序的瓶颈在哪。一个比较接地气的方法先用最简单的方式写出单线程版本跑一遍看慢不慢再判断是 IO 等待多还是 CPU 消耗多。只有先看清瓶颈才能选对工具而不是机械地套模板。上面这些例子拿过去直接跑一遍比单纯背文档要管用得多。尤其建议大家把线程、进程、协程三份代码同时放在一个脚本里调大任务量对比那些抽象的概念会在数据上变得非常直观。