如果要在编程世界里找一对“宿敌”Python 的多线程和多进程绝对榜上有名。熟悉漫威和 DC 的朋友都知道浩克和超人分别是两家漫画公司“力量天花板”级别的角色真要分出胜负得看战场规则。如果这场战斗发生在 Python 里多进程就是浩克——力大砖飞、一力降十会多线程则更像超人——看起来全能却在“氪石”——也就是 GIL 的束缚下时常有力使不出。这篇文章不会纠结两个漫画角色谁更能打而是借“浩克 VS 超人”这个对比思路把 Python 并发编程中最容易搞混的多线程与多进程讲透。你会看到它们各自的原理、代码表现、适用场景、坑点以及一份可以直接运行的实战对比案例。不管你是刚接触 Python 并发的新手还是已经在项目里被 GIL、进程池、线程安全折磨过的开发者这篇文章都能帮你建立一套清晰的选择逻辑。全文内容比较长建议先收藏。我们直接从概念开始。1. 并发编程的核心概念线程、进程与 GIL1.1 什么是进程什么是线程在操作系统层面进程是程序的一次运行实例。每个进程拥有独立的内存空间、文件描述符和系统资源。进程之间的数据默认不共享如果想要通信必须借助专门的机制比如队列、管道、共享内存。线程是进程内部的一条执行路径。同一个进程里的多个线程共享该进程的内存空间因此线程之间访问全局变量是“天然可见”的。但也正是因为共享才引入了锁、竞态条件、死锁等一系列问题。用日常生活来打比方一个厨房进程里可以有多个厨师线程同时炒菜大家共用灶台、水槽和冰箱而不同的厨房多个进程之间锅碗瓢盆完全隔离要借个酱油还得通过传菜窗口进程间通信。1.2 为什么 Python 多线程常常“不生效”这是 Python 并发学习中最容易踩坑的地方。CPython 解释器也就是最常见的 Python 官方版本中有一个全局解释器锁简称GIL。它的作用是保证同一时刻只有一个线程在解释器中执行 Python 字节码。也就是说在单个进程中Python 多线程并不能真正利用多核 CPU 来并行执行计算任务。CPU 密集型任务使用多线程反而可能因为线程切换和锁竞争而变慢。但在IO 密集型任务中比如网络请求、文件读写、数据库查询线程大部分时间都在等待 IO 返回GIL 并不会成为瓶颈多线程通过快速切换仍然能显著提升吞吐量。1.3 多进程为什么能绕开 GIL多进程的方案是干脆多开几个解释器每个进程拥有自己独立的 GIL。这样一来多个进程可以真正并行运行在不同 CPU 核心上不受单一 GIL 的限制。代价也很明显进程的内存开销比线程大得多创建和销毁进程的成本更高进程间通信也比线程间共享变量更麻烦。这部分可以看作“浩克变身”的副作用——力量无敌但消耗巨大还容易误伤队友。2. 环境准备与基础代码框架在开始写并发代码之前先统一一下实验环境。本文的示例基于以下环境操作系统Windows 10 / 11、LinuxCentOS 7、Ubuntu 20.04均可Python 版本3.8 及以上解释器CPython官方 Python开发工具VS Code 或 PyCharm 均可命令行运行没有区别如果你的电脑是多核 CPU现在基本都是建议在运行示例时打开任务管理器或top命令观察 CPU 使用率这样对“多线程无法并行”会有更直观的感受。为了后面演示方便先建一个项目目录python-concurrency-demo/ ├── thread_demo.py ├── process_demo.py ├── io_compare.py ├── cpu_compare.py └── README.md文件很简单大家按需创建即可。这里不需要安装任何第三方库全部基于 Python 标准库。3. Python 多线程实战threading 模块3.1 创建线程的最简单方式Python 的threading模块封装了线程创建与管理的常用接口。先看一个最简单的例子# 文件路径thread_demo.py import threading import time def worker(name): print(f线程 {name} 开始执行) time.sleep(1) print(f线程 {name} 执行结束) if __name__ __main__: start time.time() threads [] for i in range(3): t threading.Thread(targetworker, args(i,)) threads.append(t) t.start() for t in threads: t.join() print(f全部线程执行完毕耗时 {time.time() - start:.2f} 秒)运行方式python thread_demo.py预期输出线程顺序可能不同线程 0 开始执行 线程 1 开始执行 线程 2 开始执行 线程 0 执行结束 线程 1 执行结束 线程 2 执行结束 全部线程执行完毕耗时 1.01 秒三个线程各自sleep(1)但总耗时只有约 1 秒而不是 3 秒说明sleep期间线程确实并发切换了。这里的join()是让主线程等待子线程结束后再继续执行避免主线程提前退出。3.2 继承 Thread 类的方式除了传函数也可以继承Thread类重写run方法。这种写法适合逻辑较复杂的场景# 文件路径thread_demo.py追加 import threading import time class MyThread(threading.Thread): def __init__(self, name): super().__init__() self.name name def run(self): print(f自定义线程 {self.name} 启动) time.sleep(0.5) print(f自定义线程 {self.name} 结束) if __name__ __main__: threads [MyThread(fT-{i}) for i in range(2)] for t in threads: t.start() for t in threads: t.join()两种方式没有本质区别。简单任务用函数式写法更简洁需要在线程中维护状态时继承写法更清晰。3.3 线程锁与数据安全多线程共享全局变量是优势也是风险。看下面这段没有加锁的代码# 文件路径thread_demo.py追加 import threading counter 0 def add(loop_count): global counter for _ in range(loop_count): counter 1 if __name__ __main__: loop_count 1000000 threads [threading.Thread(targetadd, args(loop_count,)) for _ in range(4)] for t in threads: t.start() for t in threads: t.join() print(f期望结果: {loop_count * 4}) print(f实际结果: {counter})运行多次你会发现实际结果有时小于期望值。原因是counter 1这句代码在 Python 字节码层面并不是原子操作它分成“读取变量、计算加法、写回变量”三步。多个线程可能同时读到同一个旧值然后各自加 1 再写回导致更新丢失。解决办法是加锁# 文件路径thread_demo.py追加 import threading counter 0 lock threading.Lock() def safe_add(loop_count): global counter for _ in range(loop_count): with lock: counter 1 if __name__ __main__: loop_count 1000000 # 验证原版 counter 0 threads [threading.Thread(targetadd, args(loop_count,)) for _ in range(4)] for t in threads: t.start() for t in threads: t.join() print(f无锁结果: {counter}) # 验证加锁版 counter 0 threads [threading.Thread(targetsafe_add, args(loop_count,)) for _ in range(4)] for t in threads: t.start() for t in threads: t.join() print(f加锁结果: {counter})with lock等价于先调用lock.acquire()执行完后自动调用lock.release()。加锁之后结果必然等于期望值但代价是加锁、释放锁本身的性能开销以及可能的线程阻塞。4. Python 多进程实战multiprocessing 模块4.1 创建进程的基本方式多进程的使用方式和多线程几乎一一对应。先看一个简单的Process示例# 文件路径process_demo.py import multiprocessing import time def worker(name): print(f进程 {name} 开始执行) time.sleep(1) print(f进程 {name} 执行结束) if __name__ __main__: start time.time() processes [] for i in range(3): p multiprocessing.Process(targetworker, args(i,)) processes.append(p) p.start() for p in processes: p.join() print(f全部进程执行完毕耗时 {time.time() - start:.2f} 秒)运行方式python process_demo.py在 Windows 上多进程代码必须写在if __name__ __main__:下面否则会无限递归创建子进程。这是因为 Windows 使用spawn方式启动子进程会重新导入主模块如果模块顶层直接有创建进程的代码就会陷入死循环。Linux 和 macOS 使用fork方式相对宽松但为了跨平台兼容建议一律保留这个判断。4.2 使用进程池 Pool当任务数量较多时频繁创建和销毁进程的开销很大。更好的方式是用进程池预先创建固定数量的进程然后往池里提交任务# 文件路径process_demo.py追加 import multiprocessing import time def cpu_task(n): 模拟 CPU 密集型计算任务 total 0 for i in range(n): total i ** 2 return total if __name__ __main__: start time.time() with multiprocessing.Pool(processes4) as pool: results pool.map(cpu_task, [2000000, 2000000, 2000000, 2000000]) print(f结果: {results}) print(f进程池耗时: {time.time() - start:.2f} 秒)pool.map会把列表中的每个参数分发给池中的进程然后收集所有返回值。with语句会等待所有任务完成并自动关闭进程池。4.3 进程间通信Queue线程之间能直接访问共享变量但进程之间不行。好在multiprocessing提供了Queue可以让不同进程安全地传递数据# 文件路径process_demo.py追加 import multiprocessing def producer(q): for i in range(5): q.put(i) q.put(None) # 结束信号 def consumer(q): while True: item q.get() if item is None: break print(f消费者收到: {item}) if __name__ __main__: q multiprocessing.Queue() p1 multiprocessing.Process(targetproducer, args(q,)) p2 multiprocessing.Process(targetconsumer, args(q,)) p1.start() p2.start() p1.join() p2.join()这里特意用None作为结束信号避免消费者线程永远阻塞在q.get()。实际项目中也可以使用q.close()配合q.join_thread()做更精细的控制但对新手来说结束信号的方式更直观。5. 完整实战浩克 VS 超人用数据说话前面看了基础用法这一节我们来一场真正的“对决”。我会分别测试IO 密集型任务和CPU 密集型任务下多线程和多进程的表现差异。5.1 测试一IO 密集型任务对比模拟一个需要频繁等待 IO 的任务比如发送 HTTP 请求、读写文件。这里用time.sleep模拟等待# 文件路径io_compare.py import threading import multiprocessing import time def io_task(seconds0.5): time.sleep(seconds) return seconds def run_threads(task_count8): start time.time() threads [threading.Thread(targetio_task, args(0.5,)) for _ in range(task_count)] for t in threads: t.start() for t in threads: t.join() return time.time() - start def run_processes(task_count8): start time.time() processes [multiprocessing.Process(targetio_task, args(0.5,)) for _ in range(task_count)] for p in processes: p.start() for p in processes: p.join() return time.time() - start if __name__ __main__: thread_time run_threads() process_time run_processes() print(f多线程总耗时: {thread_time:.2f} 秒) print(f多进程总耗时: {process_time:.2f} 秒) print(f差异倍率: {process_time / thread_time:.2f} 倍)运行结果可能如下多线程总耗时: 0.52 秒 多进程总耗时: 0.54 秒 差异倍率: 1.04 倍结论很明确IO 密集型场景下多线程和多进程耗时几乎没有差别。既然多线程创建成本更低、内存占用更少那么 IO 密集型任务优先选择多线程完全合理。5.2 测试二CPU 密集型任务对比现在换成纯计算任务# 文件路径cpu_compare.py import threading import multiprocessing import time def cpu_task(limit30000000): total 0 for i in range(limit): total i * i return total def run_threads(task_count4): start time.time() threads [threading.Thread(targetcpu_task) for _ in range(task_count)] for t in threads: t.start() for t in threads: t.join() return time.time() - start def run_processes(task_count4): start time.time() processes [multiprocessing.Process(targetcpu_task) for _ in range(task_count)] for p in processes: p.start() for p in processes: p.join() return time.time() - start if __name__ __main__: thread_time run_threads() process_time run_processes() print(f多线程总耗时: {thread_time:.2f} 秒) print(f多进程总耗时: {process_time:.2f} 秒) print(f差异倍率: {process_time / thread_time:.2f} 倍)建议把cpu_task的参数根据本机性能调整范围控制在 1000 万到 5000 万之间。如果你有 4 核 CPU结果可能类似多线程总耗时: 5.86 秒 多进程总耗时: 1.71 秒 差异倍率: 3.43 倍这个结果直接验证了文章开头说的问题CPU 密集型任务下多线程因为 GIL 无法并行多个线程轮流抢占解释器甚至可能比单线程还慢。而多进程能利用多核 CPU几乎是线性加速差异非常明显。5.3 结果说明与选型判断两个实验合在一起结论非常清晰任务类型多线程多进程推荐选择IO 密集型网络请求、文件读写、数据库差距不大几乎等价耗时相近但资源开销大多线程CPU 密集型计算、数据处理、训练受到 GIL 限制明显偏慢多核并行速度优势大多进程需要共享大块内存数据天然支持但要加锁需要队列、管道或共享内存多线程需要彻底隔离的独立任务被异常影响的风险高进程间互相隔离多进程这个表格建议直接背下来后续选型基本够用。6. 常见问题与排查思路多线程和多进程的坑非常多这里挑几个最常遇到的集中说明。问题现象常见原因解决思路Windows 下运行多进程代码报错RuntimeError或死循环缺少if __name__ __main__:保护把入口代码全部放进if __name__ __main__:中多线程计算结果不对数字偏小多个线程同时修改共享变量导致更新丢失使用threading.Lock或改用queue.QueueCPU 密集型任务用多线程反而变慢GIL 导致线程无法并行执行字节码改用multiprocessing或concurrent.futures.ProcessPoolExecutor进程池中任务结果丢失进程间内存不共享使用Pool.map或Queue返回结果不要依赖全局变量线程卡死程序不退出可能发生了死锁检查多个锁的加锁顺序尽量使用with lock管理锁的释放创建大量进程后系统资源耗尽进程数量过多内存占用过高使用Pool限制进程数一般不超过 CPU 核心数的 1~2 倍6.1 深入排查案例多线程修改全局变量这个问题前面已经演示过。再补充一个排查思路如果你发现结果时对时错首先确认是不是并发修改共享状态。可以用dis模块查看字节码python -m dis thread_demo.py你会看到counter 1被拆成多行字节码这能直观解释为什么结果会错。不过不强制大家看懂字节码记住“共享可变状态要加锁”就够了。6.2 深入排查案例Pool 进程数设置不当进程池的默认进程数按 CPU 核心数决定但并不是“越多越好”。每个进程都有独立的内存空间Python 解释器本身的内存占用通常在几十 MB如果创建几百个进程内存可能直接被吃满。一般建议进程数设置为 CPU 核心数或略大于核心数。可以这样获取import multiprocessing print(multiprocessing.cpu_count())如果是 IO 密集型却用多进程那么进程数可以适当多一些但依然不建议超过核心数的 2 倍。不要盲目调大。6.3 容易忽略的坑多线程的异常隔离问题多进程的另一个隐藏优势是异常隔离。一个线程抛出未捕获异常会导致整个进程崩溃所有线程都会挂掉。但一个进程崩溃其他进程通常不受影响这个特性在运行多个独立任务时非常有用。因此如果你的任务非常重要且不允许因为单个任务失败而影响整体运行可以优先考虑多进程或者在多线程外层加try/except并在线程函数内部捕获所有异常。7. Python 并发的最佳实践与工程建议7.1 不要迷信“多线程 性能提升”GIL 是一个客观存在的事实。新手很容易被“并发 快”这个印象误导。先从任务类型判断IO 密集还是 CPU 密集再用time模块做基准测试最后再做技术选型。7.2 优先使用 concurrent.futures在实际项目中直接操作threading.Thread和multiprocessing.Process比较繁琐。标准库提供了concurrent.futures统一了线程池和进程池的接口切换起来非常方便# 文件路径executor_demo.py from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor import time def task(n): time.sleep(0.2) return n * n if __name__ __main__: # 线程池 with ThreadPoolExecutor(max_workers4) as executor: results list(executor.map(task, range(8))) print(f线程池结果: {results}) # 进程池 with ProcessPoolExecutor(max_workers4) as executor: results list(executor.map(task, range(8))) print(f进程池结果: {results})这种写法把“线程还是进程”的选择收敛成一个ThreadPoolExecutor或ProcessPoolExecutor的替换后续如果发现选型错误改动成本很低。7.3 合理设置并发数量并发数量不是越大越好。对于线程如果数量过多系统会频繁进行上下文切换反而降低效率对于进程内存占用会快速上升。建议基于压测数据来决定而不是拍脑袋。一个可参考的公式import multiprocessing # 线程池IO 密集型可以大于核心数比如 4~8 倍 io_thread_count multiprocessing.cpu_count() * 4 # 进程池CPU 密集型一般等于核心数 cpu_process_count multiprocessing.cpu_count()7.4 谨慎处理共享状态多线程里要尽量避免多个线程同时修改一个全局变量。除了加锁更好的思路是使用queue.Queue让线程之间通过消息传递数据这样每个线程只管“取任务、处理、放结果”不需要共享大块可变状态。多进程之间不要试图直接通过global变量传数据因为进程内修改不会影响到其他进程。统一使用Queue、Pipe或Manager。7.5 安全边界与生产环境注意事项在真实项目中要特别注意三点受控环境验证并发代码必须先在小规模测试环境跑通再上生产。你无法在本地完全模拟生产环境的 CPU 负载、网络延迟和并发量。可观测性在并发任务中务必添加日志记录任务开始时间、结束时间、处理结果、异常信息。否则一旦出现问题你很难定位是哪个线程或进程出了错。优雅退出善用join(timeout)和线程/进程的停止信号。不要在主程序结束时直接强制终止否则可能丢失未完成任务数据。7.6 超越 GIL 的另一条路协程如果你对“多线程受 GIL 限制”感到遗憾可以学习 Python 的协程。协程通过async/await在单线程内实现并发调度非常适合 IO 密集型任务在大量网络请求场景下性能甚至比多线程更好。对于 CPU 密集型仍然需要多进程或者把计算任务交给 C 扩展、NumPy 等底层并行能力。推荐的学习路线是先掌握多线程和多进程的适用边界再学习协程最后了解 asyncio 与线程池、进程池的组合用法。8. 总结回到开头的比喻浩克和超人谁更强答案永远是“看战场规则”。Python 的并发编程也一样——多进程在 CPU 密集型任务上是碾压级的优势多线程在 IO 密集型任务上则以极低的资源开销胜出。本文从进程与线程的基础概念出发解释了 GIL 对多线程的关键影响逐步演示了threading和multiprocessing的基本用法并用两份完整的对比实验数据证明了选型逻辑。最后整理了多线程和多进程的常见坑位、排查表格以及工程实践建议。如果你当前的任务是网络爬虫、大量文件读写、数据库批量查询优先考虑多线程或协程如果是图像处理、数据清洗、复杂计算多进程会更合适如果既需要并发又担心代码复杂度那就从concurrent.futures开始它足够简单也足够实用。下一步你可以尝试自己写一个小工具模拟 20 个网络请求分别用多线程和多进程实现对比资源占用再把其中一半任务改成纯计算重跑一次 CPU 密集型对比实验。自己动手跑一遍比看十篇文章都管用。如果这篇文章对你有帮助可以收藏备用也欢迎在评论区聊聊你项目里用到的并发方案以及踩过的 GIL 的坑。