想先问一个很现实的问题你写过这样的代码吗——在一个 4 核机器上用ThreadPoolExecutor(max_workers4)开了 4 个线程做纯计算任务满心期待 CPU 能直接在 4 个核上并行运行。结果跑起来之后打开任务管理器CPU 总占用率只有 110% 左右4 个核里只有 1 个核在“认真工作”另外几个核基本处于“围观”状态。我第一次遇到这个问题时第一反应是任务调度写错了。后来查了很多资料才发现问题不在我的业务代码而是出在 CPython 解释器内部的一个机制GIL。这篇文章就围绕这个场景展开重点讲清楚三件事GIL 到底是什么它到底限制了什么、没有限制什么为什么 CPU 密集场景下Python 多线程不仅不能加速反而可能更慢什么情况下应该换成多进程以及一个可以直接参考的多进程实战案例。适合以下读者阅读写过threading或ThreadPoolExecutor但对“线程没跑满 CPU”有疑惑的开发者打算用 Python 做并发计算、批量任务处理在“多线程 vs 多进程”之间摇摆的人面试前想系统梳理 GIL 相关知识的 Python 求职者。读完这篇文章你会得到一个判断方法拿到一个并发需求先看它是 CPU 密集还是 IO 密集再决定用多线程、多进程还是协程。下面我们正式开始。1. 背景与核心概念GIL 到底限制了什么1.1 从“4 个线程没跑满 4 个核”说起先解释一下这个现象的原理。当你用 Python 写多线程代码时操作系统层面确实创建了 4 个线程操作系统也确实把 4 个线程分配给了多个 CPU 核心。但问题是CPython 解释器在执行 Python 字节码时同一时刻只允许一个线程执行 Python 代码。这个限制不是 Python 语言本身的规定而是 CPython 解释器的一个实现细节。这个“限制机制”就是 GIL全称 Global Interpreter Lock也就是“全局解释器锁”。你可以把 GIL 理解成一把“解释器级别的互斥锁”在一个 Python 进程内任何线程想要执行 Python 字节码都必须先拿到这把锁。拿到锁的线程才能干活干一会儿之后释放锁其他线程再抢。于是虽然你有 4 个线程但真正在“执行 Python 代码”的线程通常只有 1 个。听到这里你可能会问那写多线程还有什么意义有意义。因为 GIL 锁住的只是“执行 Python 字节码”这一段。当线程执行到某些会主动释放 GIL 的操作时比如sleep、文件读写、网络请求、数据库等待其他线程就能拿到锁继续运行。这也是为什么 IO 密集场景下Python 多线程依然有明显效果。1.2 GIL 为什么会出现GIL 的出现和 CPython 的内存管理方式有很大关系。CPython 使用引用计数来管理对象生命周期。简单来说每个 Python 对象都有一个计数器记录当前有多少个引用指向它。当引用计数变为 0 时对象会被立即回收。问题在于如果多个线程同时读取、修改同一个对象的引用计数计数器就会出现数据竞争。比如两个线程同时让同一对象的引用计数加 1最终计数可能只加了 1而不是 2导致内存管理出错。解决这个问题大致有两个方向给每个对象都加一把细粒度的锁在整个解释器层面加一把全局锁同一时刻只允许一个线程修改内部状态。CPython 当年选择了后者也就是 GIL。这个选择让单线程性能更好实现更简单也避免了很多死锁问题但也留下了今天这个“多线程难用好 CPU”的经典话题。这里要补充一个容易误解的点GIL 不是 Python 语言的强制要求。像 Jython、IronPython 这些基于 JVM 或 .NET 的 Python 实现就没有 GIL。但大多数开发者日常使用的官方 CPython一直是带 GIL 的。1.3 GIL 保护了什么没保护什么GIL 保护的是 CPython 解释器内部的线程安全比如对象的引用计数、解释器全局状态。它不等于“你的业务代码自动线程安全”。两个线程同时执行counter 1依然可能丢失更新。因为counter 1在 Python 中不是一条原子指令它可能被拆成读取、计算、写回多个步骤而 GIL 可能在中间切换线程。下面这个例子可以直观说明import threading counter 0 def increment(): global counter for _ in range(1_000_000): counter 1 threads [threading.Thread(targetincrement) for _ in range(2)] for t in threads: t.start() for t in threads: t.join() print(f最终计数值: {counter})理论上两个线程各加 100 万次最终结果应该是 200 万。但由于counter 1不是原子操作运行多次后可能会出现小于 200 万的结果。所以正确理解是GIL 保护的是解释器内部状态GIL 不保护你的共享业务数据线程安全仍需通过threading.Lock、queue.Queue、concurrent.futures等方式自己保证。2. 环境准备与实验设计在写代码验证之前先确认一下实验环境。本文示例代码全部基于 Python 标准库不需要安装任何第三方依赖。运行方式也简单把代码保存成.py文件用 Python 直接执行即可。项目说明操作系统Linux / macOS / Windows 均可Python 版本Python 3.8 均可本文示例在 3.10 环境验证第三方库无全部使用标准库代码文件直接保存为.py文件运行硬件环境建议在 4 核及以上机器上观察效果更明显在 Windows 上运行ProcessPoolExecutor时必须把程序入口用if __name__ __main__:保护起来。这一点后面会专门说明先记住这个习惯。实验设计思路是对比三种执行方式的总耗时单线程一次性计算一个大范围内的连续整数累加多线程把同一个范围拆成 4 段用 4 个线程分别计算多进程把同一个范围拆成 4 段用 4 个进程分别计算。为了保证公平无论哪种方式总共完成的工作量都是一样的对 8000 万个数做累加。3. 核心验证CPU 密集场景下多线程为什么更慢3.1 单线程基线版本先写一个单线程版本作为性能基线。为了减少模块导入和进程启动对结果的干扰代码里只保留核心计算逻辑。# cpu_dense_compare.py import time TOTAL 80_000_000 def compute_range(start, end): total 0 for i in range(start, end 1): total i return total def single_thread(): return compute_range(1, TOTAL) if __name__ __main__: start time.perf_counter() result single_thread() elapsed time.perf_counter() - start print(f单线程 结果: {result}, 耗时: {elapsed:.2f}s)在我的测试机器上这段代码耗时大约 8 到 10 秒。这个数字并不要紧重要的是后面多线程版本和它对标的结果。3.2 多线程版本代码接下来把一个大任务拆成 4 份交给 4 个线程并行处理。# cpu_dense_compare.py import time from concurrent.futures import ThreadPoolExecutor TOTAL 80_000_000 WORKERS 4 def compute_range(start, end): total 0 for i in range(start, end 1): total i return total def multi_thread(): chunk TOTAL // WORKERS parts [(i * chunk 1, (i 1) * chunk) for i in range(WORKERS)] with ThreadPoolExecutor(max_workersWORKERS) as executor: futures [executor.submit(compute_range, s, e) for s, e in parts] return sum(f.result() for f in futures) if __name__ __main__: start time.perf_counter() result multi_thread() elapsed time.perf_counter() - start print(f多线程 4 worker 结果: {result}, 耗时: {elapsed:.2f}s)运行结果会让你意外多线程版本不仅没有比单线程快有时候还会更慢。在相同的工作量下它通常和单线程耗时接近甚至略慢一些。原因就是 GIL。4 个线程虽然创建出来了但在执行compute_range这种纯 Python 计算的循环时每个线程都在抢同一把 GIL。真正同时运行的 Python 字节码指令流只有一条线程之间还需要频繁切换切换本身还会产生额外开销。在 Python 中可以通过下面两个系统函数查看和修改 GIL 的切换时间间隔import sys # 查看当前切换间隔默认通常是 0.005 秒也就是 5ms print(sys.getswitchinterval()) # 手动调小切换间隔比如 1ms sys.setswitchinterval(0.001)sys.setswitchinterval设置的是“主线程持有 GIL 的时间片参考值”。设置为更小的值会让线程切换更频繁对交互型程序可能更友好但对 CPU 密集多线程程序来说切换开销反而会进一步拖慢总耗时。3.3 多进程版本代码多进程的方式才能绕开 GIL。每个进程都有自己独立的 Python 解释器和独立的 GIL互不干扰。这样4 个进程就能真正使用 4 个 CPU 核心。# cpu_dense_compare.py import time from concurrent.futures import ProcessPoolExecutor TOTAL 80_000_000 WORKERS 4 def compute_range(start, end): total 0 for i in range(start, end 1): total i return total def multi_process(): chunk TOTAL // WORKERS parts [(i * chunk 1, (i 1) * chunk) for i in range(WORKERS)] with ProcessPoolExecutor(max_workersWORKERS) as executor: futures [executor.submit(compute_range, s, e) for s, e in parts] return sum(f.result() for f in futures) if __name__ __main__: start time.perf_counter() result multi_process() elapsed time.perf_counter() - start print(f多进程 4 worker 结果: {result}, 耗时: {elapsed:.2f}s)这个版本的耗时通常会明显下降。在 4 核机器上理论上接近单线程耗时的四分之一。当然进程创建、参数序列化、结果回传都会有一些额外开销所以不会严格达到线性加速比但趋势是非常明确的。三种方式的结果对比如下表所示执行方式预期耗时趋势说明单线程基准耗时全部计算由一个线程完成多线程 4 worker接近基准可能略慢受 GIL 限制实际串行执行且有切换开销多进程 4 worker明显少于基准4 个进程可在多个核上并行一定要记住这个实验验证的是“趋势”不是“精确数值”。不同机器、不同 Python 小版本会产生不同的绝对耗时但 CPU 密集任务下多线程无法利用多核这一点在主流 CPython 中是非常稳定的结论。4. 什么时候该换多进程判据与上手方式4.1 判断任务类型CPU 密集还是 IO 密集先讲一个通用判断标准。如果一个任务主要消耗 CPU 计算能力比如数值计算、加密解密、哈希运算、图像像素处理、视频编解码它属于 CPU 密集任务。如果一个任务大部分时间在等待外部资源比如请求 HTTP 接口、读取数据库、读写文件、等待用户输入它属于 IO 密集任务。对于 CPU 密集任务Python 多线程因为 GIL 的存在很难利用到多核优势。此时优先考虑ProcessPoolExecutor。对于 IO 密集任务线程执行到阻塞式 IO 时会主动释放 GIL让其他线程继续运行。此时多线程仍然是一个性价比很高的方案代码简单资源开销小。4.2 从ThreadPoolExecutor切换到ProcessPoolExecutorconcurrent.futures模块的设计非常友好ThreadPoolExecutor和ProcessPoolExecutor有完全一致的接口。也就是说当你从多线程切换到多进程时核心改动往往只有一行from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor # 线程版本 # with ThreadPoolExecutor(max_workers4) as executor: # 进程版本 with ProcessPoolExecutor(max_workers4) as executor: futures [executor.submit(compute_range, s, e) for s, e in parts] result sum(f.result() for f in futures)当然切换到多进程后有几个新问题需要注意submit的函数参数必须可以被 pickle 序列化submit的函数必须尽量放到模块顶层避免使用 lambda、局部类、闭包Windows 平台下必须使用if __name__ __main__:保护程序入口进程创建开销比线程创建大任务太小时收益不明显。4.3 如何确定合适的进程数量一个常见的误区是“进程数越大越快”。实际上进程数和 CPU 核心数有关也和你运行环境中是否还有别的重要任务有关。最简单合理的做法是使用os.cpu_count()或multiprocessing.cpu_count()然后根据任务量调整。import os workers os.cpu_count() or 1 print(f当前 CPU 逻辑核心数: {workers})需要注意os.cpu_count()返回的是逻辑核心数。如果在超线程环境下数值可能是物理核心数的两倍。对于纯 CPU 计算任务进程数一般设置为物理核心数附近即可具体需要通过实测调整。如果机器上还有其他服务在运行还可以适当减少进程数量避免 CPU 争抢。5. IO 密集场景多线程依然是高性价比方案5.1 为什么 IO 密集时多线程有效再回到 GIL 的机制上来。当线程执行到time.sleep、socket.recv、requests.get、file.read等阻塞操作时它会主动释放 GIL让其他线程有机会运行。因此IO 密集任务的瓶颈不在 CPU而在“等待外部响应”。多线程可以在一个线程等待的时候切换去处理另一个任务整体吞吐量会比串行高很多。为了更直观地体验这种差异可以用一个模拟实验来看效果。5.2 模拟 IO 阻塞场景对比假设有 20 个任务每个任务模拟一次网络请求耗时 0.2 秒。# io_dense_compare.py import time from concurrent.futures import ThreadPoolExecutor def io_task(seconds): time.sleep(seconds) return seconds def run_single(n20, duration0.2): start time.perf_counter() for _ in range(n): io_task(duration) return time.perf_counter() - start def run_thread(n20, duration0.2): start time.perf_counter() with ThreadPoolExecutor(max_workersn) as executor: list(executor.map(io_task, [duration] * n)) return time.perf_counter() - start if __name__ __main__: print(f串行耗时: {run_single():.2f}s) print(f多线程耗时: {run_thread():.2f}s)预期结果串行方式20 个任务依次执行总耗时约 4 秒多线程方式20 个任务同时睡眠总耗时约 0.2 秒。这个对比非常典型。time.sleep虽然简单但它模拟的正是“线程等待外部资源”的过程。真实场景中的网络请求、数据库查询本质上也都是这种等待型任务所以 IO 密集任务中 Python 多线程效果明显。5.3 多线程、多进程、协程如何选Python 中的并发方案主要有三种方案适合场景优势劣势多线程IO 密集、任务数量多、要求响应快开发简单线程开销较小CPU 密集受 GIL 限制多进程CPU 密集、需要真正并行可充分利用多核进程开销大通信成本高协程 asyncio高并发 IO 密集、大量网络请求单线程内切换成本低并发量极高需要整个调用链都支持异步实际项目中也经常看到混合模式多进程配合多线程。比如一个服务用多进程承担主要计算任务每个进程内部再用多线程处理 IO 请求。不过这是后期优化方向初学者先把“CPU 密集选多进程、IO 密集选多线程/协程”这个基本判断拿稳。6. 实战案例多进程加速批量文件哈希前面讲了很多原理这一节用一个贴近实际工程的小项目把ProcessPoolExecutor的完整用法串起来。6.1 业务场景假设你有一个文件归档系统需要校验大量文件的完整性。操作思路是给每个文件计算 SHA256 哈希值然后把哈希值存入数据库。文件数量有几万个单个文件读入内存计算哈希需要一定时间串行处理太慢。这类任务兼具 IO 和 CPU 消耗因为要读文件还要做哈希运算。在文件较大、计算量较重的情况下多进程能有效利用多核是一个合理的选择。6.2 准备测试文件先创建一个data目录并生成一些测试文件。如果是 Linux 或 macOS可以用以下命令mkdir data echo hello world data/a.txt echo python gil data/b.txt dd if/dev/urandom ofdata/c.bin bs1M count30 dd if/dev/urandom ofdata/d.bin bs1M count30在 Windows 的 PowerShell 中可以用以下命令生成随机文件New-Item -ItemType Directory -Force data Set-Content -Path data/a.txt -Value hello world Set-Content -Path data/b.txt -Value python gil [System.IO.File]::WriteAllBytes(data/c.bin, (New-Object byte[] (30MB)))文件不用太多几个就够了重点是验证代码逻辑。6.3 完整多进程哈希代码将下面的代码保存为batch_hash.py放在data目录的同级目录下。# batch_hash.py import hashlib import time from concurrent.futures import ProcessPoolExecutor, as_completed from pathlib import Path def file_sha256(file_path): 计算单个文件的 SHA256 哈希 h hashlib.sha256() with open(file_path, rb) as f: while True: chunk f.read(1024 * 1024) if not chunk: break h.update(chunk) return file_path, h.hexdigest() def main(): files [str(p) for p in Path(data).glob(*) if p.is_file()] print(f待处理文件数: {len(files)}) results {} start time.perf_counter() with ProcessPoolExecutor(max_workers4) as executor: future_map { executor.submit(file_sha256, file_path): file_path for file_path in files } for future in as_completed(future_map): file_path future_map[future] try: path, digest future.result() except Exception as exc: print(f{file_path} 处理失败: {exc}) else: results[path] digest print(f完成: {path} - {digest}) print(f总耗时: {time.perf_counter() - start:.2f}s) print(f成功处理: {len(results)} 个文件) if __name__ __main__: main()下面逐段解释这段代码file_sha256是任务函数接收文件路径分块读取文件并计算哈希最后返回文件路径和哈希值。把文件路径一起返回是为了在结果收集阶段能识别哪个结果对应哪个文件。future_map是一个字典键是Future对象值是原始文件路径。这样在as_completed遍历时能通过future_map[future]拿到当前完成的是哪个文件。as_completed的作用是“谁先完成谁先返回”适合处理大量任务并实时输出进度。如果你更希望按照传入顺序处理结果可以使用executor.map。future.result()如果抛异常可以在当前循环捕获避免一个任务失败导致整个程序退出。运行方式python batch_hash.py预期输出大致如下待处理文件数: 4 完成: data/a.txt - 2ef7bde608ce5404e97dfe2b0f58e7a4e87aa5fcd69e0f1e64df8b0d7c0c1c20 完成: data/b.txt - 7fa2f22e96ce0a7aab62f5e2e4f5d25e2d02aef5c29d1d9c9c5f1e2a1b3c4d5e ...6.4 任务结果的等待与收集在真实项目中经常需要“提交所有任务后等所有任务完成再继续”。concurrent.futures这里有两套常用方式。方式一使用wait等待全部完成from concurrent.futures import ProcessPoolExecutor, wait with ProcessPoolExecutor(max_workers4) as executor: futures [executor.submit(file_sha256, fp) for fp in files] done, pending wait(futures) for future in done: print(future.result())wait返回两个集合done是已完成的任务集合pending是未完成的任务集合。默认wait会等待所有任务完成。方式二使用map保持输入顺序with ProcessPoolExecutor(max_workers4) as executor: for path, digest in executor.map(file_sha256, files): print(path, digest)map会按传入顺序返回结果写法更简洁。但它只能按顺序拿结果无法像as_completed那样第一时间处理某个已完成任务的结果。实际项目中如果任务是“提交一批、等待一批、再提交下一批”可以配合信号量控制并发数。不过对大多数中小型批处理任务直接submit全部任务到进程池即可ProcessPoolExecutor内部有任务队列不会把所有任务一次性跑完而是按 worker 数量调度执行。7. 常见问题与排查思路7.1 常见问题汇总问题现象常见原因解决思路开了 4 个线程CPU 只有 1 个核在忙CPU 密集任务受 GIL 限制确认任务类型换成ProcessPoolExecutor多进程运行后反而更慢任务本身太小进程创建和通信开销占比过高放大单任务粒度减少提交次数多进程报PicklingError参数或任务函数无法被 pickle 序列化把函数放到模块顶层避免 lambda 和闭包Windows 下运行进程池直接报错缺少if __name__ __main__:保护加上保护并让任务函数位于独立模块多线程共享计数结果不对GIL 不保护业务变量非原子操作使用threading.Lock或使用queue.Queue多进程返回大量结果导致主进程内存暴涨子进程统一回传所有结果在子进程内先聚合数据只回传必要信息进程数设置过大系统响应变慢worker 数超过物理核心数用os.cpu_count()做上限结合实际负载调整7.2 排查清单如果你遇到“Python 多线程提速失败”类问题可以按下面的顺序排查先确认 CPU 占用率。如果是单核满、多核闲大概率是 GIL 限制。判断任务类型。如果任务是纯计算多线程基本不会有加速效果。确认是否有隐藏的阻塞操作。print、锁等待、网络请求等都会影响表现。尝试把ThreadPoolExecutor换成ProcessPoolExecutor观察耗时变化。检查任务规模和单个任务耗时。任务太小时进程创建成本可能超过并行收益。逐步调整max_workers找到当前机器上的最优并发度。8. 最佳实践与工程建议8.1 先做任务分类再选并发方案很多并发性能问题的根源并不是代码写得不好而是并发模型选错了。建议在写代码之前先问自己一个问题这个任务是在“等 CPU”还是“等 IO”如果是等 CPU用多进程或把计算部分交给 C 扩展例如 NumPy、Pandas 底层计算不要试图用多线程硬扛。如果是等 IO多线程通常是成本最低、效果最明显的方式。8.2 合理使用concurrent.futures在 Python 中concurrent.futures已经封装了线程池和进程池的常用功能建议优先使用而不是自己管理threading.Thread或multiprocessing.Process。使用with语句管理线程池和进程池可以确保任务结束后资源正确释放with ProcessPoolExecutor(max_workers4) as executor: futures [executor.submit(task, arg) for arg in args] for future in as_completed(futures): # 处理结果 pass8.3 子进程任务函数要放在模块顶层ProcessPoolExecutor需要把任务函数和参数序列化后发送给子进程因此任务函数必须是可 pickled 的。一个很常见的坑是在函数内部定义局部函数然后把这个局部函数提交给进程池结果运行时报错。正确做法是把任务函数定义在模块顶层参数尽量使用基础类型比如字符串、数字、元组。如果需要传递复杂对象考虑先转成可序列化的数据结构。8.4 注意进程数限制和资源开销不要无脑设置max_workers100。每个进程都有自己的 Python 解释器、内存空间和文件描述符进程数过多会带来很大的资源开销。建议先使用os.cpu_count()获得当前机器的核心数import os default_workers min(os.cpu_count() or 1, 8)然后在测试环境中用不同的 worker 数跑一遍观察耗时和 CPU 占用选择收益最高的值。8.5 生产级任务要处理异常和日志在多进程编程中主进程和子进程的输出并不完全共享。直接在每个子进程中调用print输出内容可能出现顺序混乱在某些平台上甚至可能被缓冲。更稳的做法是子进程只负责计算结果并将结果返回或写入任务队列由主进程统一负责日志、数据库写入、异常上报。另外future.result()要捕获异常防止单个任务失败导致整个程序中断for future in as_completed(future_map): try: result future.result() except Exception as exc: log.error(f任务执行失败: {exc}) else: log.info(f任务成功: {result})8.6 关于 GIL 的常见误区这里总结几个高频误区GIL 是 Python 语言的缺陷不是。它是 CPython 解释器实现的一种机制Python 语言本身没有规定必须存在 GIL。有 GIL 就不需要加锁不对。GIL 不能保护你的业务共享数据。多线程完全不能用不对。IO 密集场景下多线程仍然非常好用。换成 PyPy 就没问题不完全对。PyPy 有自己的并发模型和 STM 方案但不是所有第三方库都支持得很好迁移成本需要考虑。9. 总结与下一步学习建议回到开头的问题4 个线程为什么没跑满 4 个核答案是在 CPython 中GIL 限制了同一进程内多个线程同时执行 Python 字节码因此 CPU 密集任务用多线程几乎无法利用多核。正确做法是判断任务类型CPU 密集改为多进程IO 密集保留多线程或改用协程。本文通过三个实验帮助你建立了这个认知CPU 密集任务单线程、多线程、多进程耗时对比IO 密集任务串行与多线程耗时对比批量文件哈希完整的多进程实战代码。如果你还想继续深入学习建议按下面的路线展开学习threading.Lock、queue.Queue理解线程安全的基本手段学习multiprocessing.Queue和multiprocessing.Pipe掌握多进程通信学习asyncio和协程掌握高并发 IO 场景的更好解法学习 C 扩展或 NumPy 的调用方式了解为什么 C 扩展能绕开 GIL 的限制关注 Python 版本更新例如 Python 3.13 开始提供实验性的自由线程构建未来 GIL 的形态可能会进一步变化。GIL 并不是一个需要害怕的概念。理解它之后你会对 Python 的并发模型有更清晰的全局认识。以后再遇到“多线程没跑满多核”的怪问题第一反应应该是先看任务类型别再盲目加线程了。
想先问一个很现实的问题你写过这样的代码吗——在一个 4 核机器上用ThreadPoolExecutor(max_workers4)开了 4 个线程做纯计算任务满心期待 CPU 能直接在 4 个核上并行运行。结果跑起来之后打开任务管理器CPU 总占用率只有 110% 左右4 个核里只有 1 个核在“认真工作”另外几个核基本处于“围观”状态。我第一次遇到这个问题时第一反应是任务调度写错了。后来查了很多资料才发现问题不在我的业务代码而是出在 CPython 解释器内部的一个机制GIL。这篇文章就围绕这个场景展开重点讲清楚三件事GIL 到底是什么它到底限制了什么、没有限制什么为什么 CPU 密集场景下Python 多线程不仅不能加速反而可能更慢什么情况下应该换成多进程以及一个可以直接参考的多进程实战案例。适合以下读者阅读写过threading或ThreadPoolExecutor但对“线程没跑满 CPU”有疑惑的开发者打算用 Python 做并发计算、批量任务处理在“多线程 vs 多进程”之间摇摆的人面试前想系统梳理 GIL 相关知识的 Python 求职者。读完这篇文章你会得到一个判断方法拿到一个并发需求先看它是 CPU 密集还是 IO 密集再决定用多线程、多进程还是协程。下面我们正式开始。1. 背景与核心概念GIL 到底限制了什么1.1 从“4 个线程没跑满 4 个核”说起先解释一下这个现象的原理。当你用 Python 写多线程代码时操作系统层面确实创建了 4 个线程操作系统也确实把 4 个线程分配给了多个 CPU 核心。但问题是CPython 解释器在执行 Python 字节码时同一时刻只允许一个线程执行 Python 代码。这个限制不是 Python 语言本身的规定而是 CPython 解释器的一个实现细节。这个“限制机制”就是 GIL全称 Global Interpreter Lock也就是“全局解释器锁”。你可以把 GIL 理解成一把“解释器级别的互斥锁”在一个 Python 进程内任何线程想要执行 Python 字节码都必须先拿到这把锁。拿到锁的线程才能干活干一会儿之后释放锁其他线程再抢。于是虽然你有 4 个线程但真正在“执行 Python 代码”的线程通常只有 1 个。听到这里你可能会问那写多线程还有什么意义有意义。因为 GIL 锁住的只是“执行 Python 字节码”这一段。当线程执行到某些会主动释放 GIL 的操作时比如sleep、文件读写、网络请求、数据库等待其他线程就能拿到锁继续运行。这也是为什么 IO 密集场景下Python 多线程依然有明显效果。1.2 GIL 为什么会出现GIL 的出现和 CPython 的内存管理方式有很大关系。CPython 使用引用计数来管理对象生命周期。简单来说每个 Python 对象都有一个计数器记录当前有多少个引用指向它。当引用计数变为 0 时对象会被立即回收。问题在于如果多个线程同时读取、修改同一个对象的引用计数计数器就会出现数据竞争。比如两个线程同时让同一对象的引用计数加 1最终计数可能只加了 1而不是 2导致内存管理出错。解决这个问题大致有两个方向给每个对象都加一把细粒度的锁在整个解释器层面加一把全局锁同一时刻只允许一个线程修改内部状态。CPython 当年选择了后者也就是 GIL。这个选择让单线程性能更好实现更简单也避免了很多死锁问题但也留下了今天这个“多线程难用好 CPU”的经典话题。这里要补充一个容易误解的点GIL 不是 Python 语言的强制要求。像 Jython、IronPython 这些基于 JVM 或 .NET 的 Python 实现就没有 GIL。但大多数开发者日常使用的官方 CPython一直是带 GIL 的。1.3 GIL 保护了什么没保护什么GIL 保护的是 CPython 解释器内部的线程安全比如对象的引用计数、解释器全局状态。它不等于“你的业务代码自动线程安全”。两个线程同时执行counter 1依然可能丢失更新。因为counter 1在 Python 中不是一条原子指令它可能被拆成读取、计算、写回多个步骤而 GIL 可能在中间切换线程。下面这个例子可以直观说明import threading counter 0 def increment(): global counter for _ in range(1_000_000): counter 1 threads [threading.Thread(targetincrement) for _ in range(2)] for t in threads: t.start() for t in threads: t.join() print(f最终计数值: {counter})理论上两个线程各加 100 万次最终结果应该是 200 万。但由于counter 1不是原子操作运行多次后可能会出现小于 200 万的结果。所以正确理解是GIL 保护的是解释器内部状态GIL 不保护你的共享业务数据线程安全仍需通过threading.Lock、queue.Queue、concurrent.futures等方式自己保证。2. 环境准备与实验设计在写代码验证之前先确认一下实验环境。本文示例代码全部基于 Python 标准库不需要安装任何第三方依赖。运行方式也简单把代码保存成.py文件用 Python 直接执行即可。项目说明操作系统Linux / macOS / Windows 均可Python 版本Python 3.8 均可本文示例在 3.10 环境验证第三方库无全部使用标准库代码文件直接保存为.py文件运行硬件环境建议在 4 核及以上机器上观察效果更明显在 Windows 上运行ProcessPoolExecutor时必须把程序入口用if __name__ __main__:保护起来。这一点后面会专门说明先记住这个习惯。实验设计思路是对比三种执行方式的总耗时单线程一次性计算一个大范围内的连续整数累加多线程把同一个范围拆成 4 段用 4 个线程分别计算多进程把同一个范围拆成 4 段用 4 个进程分别计算。为了保证公平无论哪种方式总共完成的工作量都是一样的对 8000 万个数做累加。3. 核心验证CPU 密集场景下多线程为什么更慢3.1 单线程基线版本先写一个单线程版本作为性能基线。为了减少模块导入和进程启动对结果的干扰代码里只保留核心计算逻辑。# cpu_dense_compare.py import time TOTAL 80_000_000 def compute_range(start, end): total 0 for i in range(start, end 1): total i return total def single_thread(): return compute_range(1, TOTAL) if __name__ __main__: start time.perf_counter() result single_thread() elapsed time.perf_counter() - start print(f单线程 结果: {result}, 耗时: {elapsed:.2f}s)在我的测试机器上这段代码耗时大约 8 到 10 秒。这个数字并不要紧重要的是后面多线程版本和它对标的结果。3.2 多线程版本代码接下来把一个大任务拆成 4 份交给 4 个线程并行处理。# cpu_dense_compare.py import time from concurrent.futures import ThreadPoolExecutor TOTAL 80_000_000 WORKERS 4 def compute_range(start, end): total 0 for i in range(start, end 1): total i return total def multi_thread(): chunk TOTAL // WORKERS parts [(i * chunk 1, (i 1) * chunk) for i in range(WORKERS)] with ThreadPoolExecutor(max_workersWORKERS) as executor: futures [executor.submit(compute_range, s, e) for s, e in parts] return sum(f.result() for f in futures) if __name__ __main__: start time.perf_counter() result multi_thread() elapsed time.perf_counter() - start print(f多线程 4 worker 结果: {result}, 耗时: {elapsed:.2f}s)运行结果会让你意外多线程版本不仅没有比单线程快有时候还会更慢。在相同的工作量下它通常和单线程耗时接近甚至略慢一些。原因就是 GIL。4 个线程虽然创建出来了但在执行compute_range这种纯 Python 计算的循环时每个线程都在抢同一把 GIL。真正同时运行的 Python 字节码指令流只有一条线程之间还需要频繁切换切换本身还会产生额外开销。在 Python 中可以通过下面两个系统函数查看和修改 GIL 的切换时间间隔import sys # 查看当前切换间隔默认通常是 0.005 秒也就是 5ms print(sys.getswitchinterval()) # 手动调小切换间隔比如 1ms sys.setswitchinterval(0.001)sys.setswitchinterval设置的是“主线程持有 GIL 的时间片参考值”。设置为更小的值会让线程切换更频繁对交互型程序可能更友好但对 CPU 密集多线程程序来说切换开销反而会进一步拖慢总耗时。3.3 多进程版本代码多进程的方式才能绕开 GIL。每个进程都有自己独立的 Python 解释器和独立的 GIL互不干扰。这样4 个进程就能真正使用 4 个 CPU 核心。# cpu_dense_compare.py import time from concurrent.futures import ProcessPoolExecutor TOTAL 80_000_000 WORKERS 4 def compute_range(start, end): total 0 for i in range(start, end 1): total i return total def multi_process(): chunk TOTAL // WORKERS parts [(i * chunk 1, (i 1) * chunk) for i in range(WORKERS)] with ProcessPoolExecutor(max_workersWORKERS) as executor: futures [executor.submit(compute_range, s, e) for s, e in parts] return sum(f.result() for f in futures) if __name__ __main__: start time.perf_counter() result multi_process() elapsed time.perf_counter() - start print(f多进程 4 worker 结果: {result}, 耗时: {elapsed:.2f}s)这个版本的耗时通常会明显下降。在 4 核机器上理论上接近单线程耗时的四分之一。当然进程创建、参数序列化、结果回传都会有一些额外开销所以不会严格达到线性加速比但趋势是非常明确的。三种方式的结果对比如下表所示执行方式预期耗时趋势说明单线程基准耗时全部计算由一个线程完成多线程 4 worker接近基准可能略慢受 GIL 限制实际串行执行且有切换开销多进程 4 worker明显少于基准4 个进程可在多个核上并行一定要记住这个实验验证的是“趋势”不是“精确数值”。不同机器、不同 Python 小版本会产生不同的绝对耗时但 CPU 密集任务下多线程无法利用多核这一点在主流 CPython 中是非常稳定的结论。4. 什么时候该换多进程判据与上手方式4.1 判断任务类型CPU 密集还是 IO 密集先讲一个通用判断标准。如果一个任务主要消耗 CPU 计算能力比如数值计算、加密解密、哈希运算、图像像素处理、视频编解码它属于 CPU 密集任务。如果一个任务大部分时间在等待外部资源比如请求 HTTP 接口、读取数据库、读写文件、等待用户输入它属于 IO 密集任务。对于 CPU 密集任务Python 多线程因为 GIL 的存在很难利用到多核优势。此时优先考虑ProcessPoolExecutor。对于 IO 密集任务线程执行到阻塞式 IO 时会主动释放 GIL让其他线程继续运行。此时多线程仍然是一个性价比很高的方案代码简单资源开销小。4.2 从ThreadPoolExecutor切换到ProcessPoolExecutorconcurrent.futures模块的设计非常友好ThreadPoolExecutor和ProcessPoolExecutor有完全一致的接口。也就是说当你从多线程切换到多进程时核心改动往往只有一行from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor # 线程版本 # with ThreadPoolExecutor(max_workers4) as executor: # 进程版本 with ProcessPoolExecutor(max_workers4) as executor: futures [executor.submit(compute_range, s, e) for s, e in parts] result sum(f.result() for f in futures)当然切换到多进程后有几个新问题需要注意submit的函数参数必须可以被 pickle 序列化submit的函数必须尽量放到模块顶层避免使用 lambda、局部类、闭包Windows 平台下必须使用if __name__ __main__:保护程序入口进程创建开销比线程创建大任务太小时收益不明显。4.3 如何确定合适的进程数量一个常见的误区是“进程数越大越快”。实际上进程数和 CPU 核心数有关也和你运行环境中是否还有别的重要任务有关。最简单合理的做法是使用os.cpu_count()或multiprocessing.cpu_count()然后根据任务量调整。import os workers os.cpu_count() or 1 print(f当前 CPU 逻辑核心数: {workers})需要注意os.cpu_count()返回的是逻辑核心数。如果在超线程环境下数值可能是物理核心数的两倍。对于纯 CPU 计算任务进程数一般设置为物理核心数附近即可具体需要通过实测调整。如果机器上还有其他服务在运行还可以适当减少进程数量避免 CPU 争抢。5. IO 密集场景多线程依然是高性价比方案5.1 为什么 IO 密集时多线程有效再回到 GIL 的机制上来。当线程执行到time.sleep、socket.recv、requests.get、file.read等阻塞操作时它会主动释放 GIL让其他线程有机会运行。因此IO 密集任务的瓶颈不在 CPU而在“等待外部响应”。多线程可以在一个线程等待的时候切换去处理另一个任务整体吞吐量会比串行高很多。为了更直观地体验这种差异可以用一个模拟实验来看效果。5.2 模拟 IO 阻塞场景对比假设有 20 个任务每个任务模拟一次网络请求耗时 0.2 秒。# io_dense_compare.py import time from concurrent.futures import ThreadPoolExecutor def io_task(seconds): time.sleep(seconds) return seconds def run_single(n20, duration0.2): start time.perf_counter() for _ in range(n): io_task(duration) return time.perf_counter() - start def run_thread(n20, duration0.2): start time.perf_counter() with ThreadPoolExecutor(max_workersn) as executor: list(executor.map(io_task, [duration] * n)) return time.perf_counter() - start if __name__ __main__: print(f串行耗时: {run_single():.2f}s) print(f多线程耗时: {run_thread():.2f}s)预期结果串行方式20 个任务依次执行总耗时约 4 秒多线程方式20 个任务同时睡眠总耗时约 0.2 秒。这个对比非常典型。time.sleep虽然简单但它模拟的正是“线程等待外部资源”的过程。真实场景中的网络请求、数据库查询本质上也都是这种等待型任务所以 IO 密集任务中 Python 多线程效果明显。5.3 多线程、多进程、协程如何选Python 中的并发方案主要有三种方案适合场景优势劣势多线程IO 密集、任务数量多、要求响应快开发简单线程开销较小CPU 密集受 GIL 限制多进程CPU 密集、需要真正并行可充分利用多核进程开销大通信成本高协程 asyncio高并发 IO 密集、大量网络请求单线程内切换成本低并发量极高需要整个调用链都支持异步实际项目中也经常看到混合模式多进程配合多线程。比如一个服务用多进程承担主要计算任务每个进程内部再用多线程处理 IO 请求。不过这是后期优化方向初学者先把“CPU 密集选多进程、IO 密集选多线程/协程”这个基本判断拿稳。6. 实战案例多进程加速批量文件哈希前面讲了很多原理这一节用一个贴近实际工程的小项目把ProcessPoolExecutor的完整用法串起来。6.1 业务场景假设你有一个文件归档系统需要校验大量文件的完整性。操作思路是给每个文件计算 SHA256 哈希值然后把哈希值存入数据库。文件数量有几万个单个文件读入内存计算哈希需要一定时间串行处理太慢。这类任务兼具 IO 和 CPU 消耗因为要读文件还要做哈希运算。在文件较大、计算量较重的情况下多进程能有效利用多核是一个合理的选择。6.2 准备测试文件先创建一个data目录并生成一些测试文件。如果是 Linux 或 macOS可以用以下命令mkdir data echo hello world data/a.txt echo python gil data/b.txt dd if/dev/urandom ofdata/c.bin bs1M count30 dd if/dev/urandom ofdata/d.bin bs1M count30在 Windows 的 PowerShell 中可以用以下命令生成随机文件New-Item -ItemType Directory -Force data Set-Content -Path data/a.txt -Value hello world Set-Content -Path data/b.txt -Value python gil [System.IO.File]::WriteAllBytes(data/c.bin, (New-Object byte[] (30MB)))文件不用太多几个就够了重点是验证代码逻辑。6.3 完整多进程哈希代码将下面的代码保存为batch_hash.py放在data目录的同级目录下。# batch_hash.py import hashlib import time from concurrent.futures import ProcessPoolExecutor, as_completed from pathlib import Path def file_sha256(file_path): 计算单个文件的 SHA256 哈希 h hashlib.sha256() with open(file_path, rb) as f: while True: chunk f.read(1024 * 1024) if not chunk: break h.update(chunk) return file_path, h.hexdigest() def main(): files [str(p) for p in Path(data).glob(*) if p.is_file()] print(f待处理文件数: {len(files)}) results {} start time.perf_counter() with ProcessPoolExecutor(max_workers4) as executor: future_map { executor.submit(file_sha256, file_path): file_path for file_path in files } for future in as_completed(future_map): file_path future_map[future] try: path, digest future.result() except Exception as exc: print(f{file_path} 处理失败: {exc}) else: results[path] digest print(f完成: {path} - {digest}) print(f总耗时: {time.perf_counter() - start:.2f}s) print(f成功处理: {len(results)} 个文件) if __name__ __main__: main()下面逐段解释这段代码file_sha256是任务函数接收文件路径分块读取文件并计算哈希最后返回文件路径和哈希值。把文件路径一起返回是为了在结果收集阶段能识别哪个结果对应哪个文件。future_map是一个字典键是Future对象值是原始文件路径。这样在as_completed遍历时能通过future_map[future]拿到当前完成的是哪个文件。as_completed的作用是“谁先完成谁先返回”适合处理大量任务并实时输出进度。如果你更希望按照传入顺序处理结果可以使用executor.map。future.result()如果抛异常可以在当前循环捕获避免一个任务失败导致整个程序退出。运行方式python batch_hash.py预期输出大致如下待处理文件数: 4 完成: data/a.txt - 2ef7bde608ce5404e97dfe2b0f58e7a4e87aa5fcd69e0f1e64df8b0d7c0c1c20 完成: data/b.txt - 7fa2f22e96ce0a7aab62f5e2e4f5d25e2d02aef5c29d1d9c9c5f1e2a1b3c4d5e ...6.4 任务结果的等待与收集在真实项目中经常需要“提交所有任务后等所有任务完成再继续”。concurrent.futures这里有两套常用方式。方式一使用wait等待全部完成from concurrent.futures import ProcessPoolExecutor, wait with ProcessPoolExecutor(max_workers4) as executor: futures [executor.submit(file_sha256, fp) for fp in files] done, pending wait(futures) for future in done: print(future.result())wait返回两个集合done是已完成的任务集合pending是未完成的任务集合。默认wait会等待所有任务完成。方式二使用map保持输入顺序with ProcessPoolExecutor(max_workers4) as executor: for path, digest in executor.map(file_sha256, files): print(path, digest)map会按传入顺序返回结果写法更简洁。但它只能按顺序拿结果无法像as_completed那样第一时间处理某个已完成任务的结果。实际项目中如果任务是“提交一批、等待一批、再提交下一批”可以配合信号量控制并发数。不过对大多数中小型批处理任务直接submit全部任务到进程池即可ProcessPoolExecutor内部有任务队列不会把所有任务一次性跑完而是按 worker 数量调度执行。7. 常见问题与排查思路7.1 常见问题汇总问题现象常见原因解决思路开了 4 个线程CPU 只有 1 个核在忙CPU 密集任务受 GIL 限制确认任务类型换成ProcessPoolExecutor多进程运行后反而更慢任务本身太小进程创建和通信开销占比过高放大单任务粒度减少提交次数多进程报PicklingError参数或任务函数无法被 pickle 序列化把函数放到模块顶层避免 lambda 和闭包Windows 下运行进程池直接报错缺少if __name__ __main__:保护加上保护并让任务函数位于独立模块多线程共享计数结果不对GIL 不保护业务变量非原子操作使用threading.Lock或使用queue.Queue多进程返回大量结果导致主进程内存暴涨子进程统一回传所有结果在子进程内先聚合数据只回传必要信息进程数设置过大系统响应变慢worker 数超过物理核心数用os.cpu_count()做上限结合实际负载调整7.2 排查清单如果你遇到“Python 多线程提速失败”类问题可以按下面的顺序排查先确认 CPU 占用率。如果是单核满、多核闲大概率是 GIL 限制。判断任务类型。如果任务是纯计算多线程基本不会有加速效果。确认是否有隐藏的阻塞操作。print、锁等待、网络请求等都会影响表现。尝试把ThreadPoolExecutor换成ProcessPoolExecutor观察耗时变化。检查任务规模和单个任务耗时。任务太小时进程创建成本可能超过并行收益。逐步调整max_workers找到当前机器上的最优并发度。8. 最佳实践与工程建议8.1 先做任务分类再选并发方案很多并发性能问题的根源并不是代码写得不好而是并发模型选错了。建议在写代码之前先问自己一个问题这个任务是在“等 CPU”还是“等 IO”如果是等 CPU用多进程或把计算部分交给 C 扩展例如 NumPy、Pandas 底层计算不要试图用多线程硬扛。如果是等 IO多线程通常是成本最低、效果最明显的方式。8.2 合理使用concurrent.futures在 Python 中concurrent.futures已经封装了线程池和进程池的常用功能建议优先使用而不是自己管理threading.Thread或multiprocessing.Process。使用with语句管理线程池和进程池可以确保任务结束后资源正确释放with ProcessPoolExecutor(max_workers4) as executor: futures [executor.submit(task, arg) for arg in args] for future in as_completed(futures): # 处理结果 pass8.3 子进程任务函数要放在模块顶层ProcessPoolExecutor需要把任务函数和参数序列化后发送给子进程因此任务函数必须是可 pickled 的。一个很常见的坑是在函数内部定义局部函数然后把这个局部函数提交给进程池结果运行时报错。正确做法是把任务函数定义在模块顶层参数尽量使用基础类型比如字符串、数字、元组。如果需要传递复杂对象考虑先转成可序列化的数据结构。8.4 注意进程数限制和资源开销不要无脑设置max_workers100。每个进程都有自己的 Python 解释器、内存空间和文件描述符进程数过多会带来很大的资源开销。建议先使用os.cpu_count()获得当前机器的核心数import os default_workers min(os.cpu_count() or 1, 8)然后在测试环境中用不同的 worker 数跑一遍观察耗时和 CPU 占用选择收益最高的值。8.5 生产级任务要处理异常和日志在多进程编程中主进程和子进程的输出并不完全共享。直接在每个子进程中调用print输出内容可能出现顺序混乱在某些平台上甚至可能被缓冲。更稳的做法是子进程只负责计算结果并将结果返回或写入任务队列由主进程统一负责日志、数据库写入、异常上报。另外future.result()要捕获异常防止单个任务失败导致整个程序中断for future in as_completed(future_map): try: result future.result() except Exception as exc: log.error(f任务执行失败: {exc}) else: log.info(f任务成功: {result})8.6 关于 GIL 的常见误区这里总结几个高频误区GIL 是 Python 语言的缺陷不是。它是 CPython 解释器实现的一种机制Python 语言本身没有规定必须存在 GIL。有 GIL 就不需要加锁不对。GIL 不能保护你的业务共享数据。多线程完全不能用不对。IO 密集场景下多线程仍然非常好用。换成 PyPy 就没问题不完全对。PyPy 有自己的并发模型和 STM 方案但不是所有第三方库都支持得很好迁移成本需要考虑。9. 总结与下一步学习建议回到开头的问题4 个线程为什么没跑满 4 个核答案是在 CPython 中GIL 限制了同一进程内多个线程同时执行 Python 字节码因此 CPU 密集任务用多线程几乎无法利用多核。正确做法是判断任务类型CPU 密集改为多进程IO 密集保留多线程或改用协程。本文通过三个实验帮助你建立了这个认知CPU 密集任务单线程、多线程、多进程耗时对比IO 密集任务串行与多线程耗时对比批量文件哈希完整的多进程实战代码。如果你还想继续深入学习建议按下面的路线展开学习threading.Lock、queue.Queue理解线程安全的基本手段学习multiprocessing.Queue和multiprocessing.Pipe掌握多进程通信学习asyncio和协程掌握高并发 IO 场景的更好解法学习 C 扩展或 NumPy 的调用方式了解为什么 C 扩展能绕开 GIL 的限制关注 Python 版本更新例如 Python 3.13 开始提供实验性的自由线程构建未来 GIL 的形态可能会进一步变化。GIL 并不是一个需要害怕的概念。理解它之后你会对 Python 的并发模型有更清晰的全局认识。以后再遇到“多线程没跑满多核”的怪问题第一反应应该是先看任务类型别再盲目加线程了。
