第一次接触multiprocessing是因为一个很蠢的需求手上有几百万条文本要批量去重里面还夹着一堆正则匹配单线程跑一次要好几个小时。我把代码改成threading后发现 CPU 占用只有 25%四个核心闲着三个这才意识到问题出在哪。后来把任务改成多进程几个小时的活压到了二十多分钟。从那时起我开始系统地记录 Python 多进程的用法从最基础的Process到Pool、Queue、Pipe、共享内存、同步原语再到各种让人抓狂的坑前前后后积累了快四万字。这篇就是其中最有代表性的一部分而且我会把它当作一份持续更新的文档来维护2026 年这次修订补充了不少新版本解释器下的实测结论。文章主要覆盖两层内容第一层是可以直接抄走的代码骨架包括Process与Pool的常用姿势、进程间通信的主流方案、防止数据错乱的同步手段第二层是我这些年踩过的坑比如fork和spawn的区别、Pool嵌套死锁、子进程残留、以及任务切分对性能的影响。如果你是第一次尝试并行计算前面三章足够让你跑起来如果你已经在用了但偶尔遇到解释不清的怪问题直接跳到第五章那里有个排查清单能帮你省下不少时间。1. 为什么绕不开 multiprocessingGIL 与 CPU 密集任务的那点事1.1 GIL 到底限制了什么GIL 的全称是 Global Interpreter Lock全局解释器锁它是 CPython 解释器内部的一把大锁。很多人都听过这个名字但不一定清楚它限制的到底是什么。它保证同一时刻只有一个线程在执行 Python 字节码注意是“字节码”而不是“代码”。这意味着你的 Python 线程面对纯计算任务时根本无法真正利用多核 CPU。那为什么网络爬虫、文件读写这类程序用多线程效果还不错因为线程在等待网络响应、等待磁盘数据的时候会主动释放 GIL让其他线程有机会执行。这时候线程切换带来的开销远小于等待开销多线程自然划算。可一旦你的任务变成“大量算术运算”“字符串处理”“正则匹配”这类持续占用 CPU 的活GIL 就会被当前线程长期霸占其他线程即使就绪也只能干等着结果就是一个多线程程序在单核上排队执行。多进程能绕过 GIL原理也很直接每个进程都有自己独立的 Python 解释器、独立的内存空间、独立的 GIL。4 个进程就是 4 个解释器在跑互相之间谁也不挡谁的路这才谈得上真正的并行。1.2 多进程、多线程、协程怎么选很多人一上来就问“是不是多进程一定比多线程好”这个问题没有标准答案关键看任务卡在哪。IO 密集型的任务比如发一万个 HTTP 请求多线程往往比多进程更合适因为线程开销小而且大部分时间都在等待CPU 密集型的任务比如图像滤波、词频统计、数据压缩才需要多进程把计算压力分散到多个核心上协程则适合高并发但非阻塞 IO 的场景类似同时管理几千个连接的长连接服务。方案启动开销内存隔离适合场景典型例子多线程较小共享IO 密集等待时间长爬虫、批量请求、文件拷贝多进程较大独立CPU 密集任务之间耦合少数据处理、图像处理、科学计算协程最小共享高并发、单线程内切换异步 API 调用、WebSocket 服务实际开发里往往混着用比如主进程里开一个进程池负责计算每个计算任务内部再用协程或者线程去拉取前置数据这是一个很常见的三层混合模型。做选型时先问自己一个问题如果把任务换成纯while循环死算线程数量翻倍CPU 占用会上升吗如果不会说明瓶颈被 GIL 卡住了直接上多进程。1.3 多进程不是免费的多进程能并行但代价不小。每个进程启动时要重新初始化解释器复制一部分父进程的资源和状态耗时和内存都比线程高一个量级。进程间通信要经过序列化和反序列化数据量一大吞吐量立刻下降。所以在动手改造之前先确认任务确实值得付出这些成本。我的判断标准很简单任务总耗时超过 10 秒且能明确切分成互不依赖的小块才值得上多进程如果任务本身就几百毫秒进程启动和结果回传的时间反而会超过计算时间得不偿失。2. 从零起步Process、Pool 与一套可以照抄的代码骨架2.1 最小可用的 Process 写法先看最直接的用法用multiprocessing.Process手工创建子进程import multiprocessing as mp import time def worker(name: str, delay: float): print(f进程 {name} 开始) time.sleep(delay) print(f进程 {name} 结束) if __name__ __main__: p1 mp.Process(targetworker, args(A, 2)) p2 mp.Process(targetworker, args(B, 1)) p1.start() p2.start() p1.join() p2.join() print(两个子进程都执行完了)这段代码里start()会创建子进程并开始执行target函数join()会让主进程等待子进程结束。如果缺了join()主进程可能先退出子进程变成孤儿进程行为变得不可控。if __name__ __main__这行不是装饰而是必须。尤其在使用spawn启动方式的平台创建子进程时会重新导入主模块如果不加这层保护子进程导入主模块时又去创建子进程就会形成递归式进程爆炸。我见过有人把这行删掉后在 Windows 上跑出来几千个 Python 进程最后只能强制重启机器。2.2 用 Pool 优雅调度一批任务手工Process管理少量进程还够用但任务一多就很痛苦要自己维护进程列表、处理异常、回收结果。更优雅的方式是用进程池Pool它可以提前创建固定数量的工作进程然后把任务映射进去from multiprocessing import Pool def square(x: int) - int: return x * x if __name__ __main__: with Pool(4) as pool: results pool.map(square, range(10)) print(results)pool.map会把range(10)这十个任务均匀分配给 4 个工作进程全部执行完后按输入顺序返回结果。整个流程对调用方隐藏了进程调度的细节是世界上比较干净的并行化方式。Pool还提供另外几个常用方法imap惰性返回迭代器边执行边取结果适合任务大、不想一次性加载全部结果到内存的场景。starmap与map类似但支持多参数传入args里的每个元素是一个元组。apply_async异步提交单个任务返回一个AsyncResult对象可以调用get(timeout...)等待结果。这里有个容易被忽略的细节with Pool(4) as pool离开上下文时会强制终止池内进程。如果之前调用的是map这种同步方法任务执行完才退出没有影响但如果用了map_async任务还没执行完就离开with块子进程会被直接terminate拿不到任何结果。很多人第一次遇到Pool结果为空就是这个原因。2.3 一套适合日常任务的骨架我平时写并行任务一般不会每次都从头组织代码而是套用下面这个骨架它可以根据任务类型快速改造import multiprocessing as mp from multiprocessing import Pool from typing import Any, Callable, Iterable def run_parallel(func: Callable, items: Iterable, workers: int None) - list: workers workers or mp.cpu_count() items list(items) with Pool(workers) as pool: results pool.map(func, items) return results if __name__ __main__: data range(1_000_000) output run_parallel(square, data, workers8)这套骨架的关键在workers参数的设置。直接调用mp.cpu_count()返回的是逻辑 CPU 数量很多机器上因为超线程这个数字是物理核数的两倍。对于纯计算任务进程数设为物理核数往往最合适超过之后不仅提升有限还会增加调度和内存压力。具体怎么测试我会在第六章用真实数据说明。3. 跨进程通信Queue、Pipe 与共享内存的取舍3.1 Queue 与生产者消费者模型进程之间是独立的内存空间不能像线程那样通过修改同一个变量来通信必须走进程间通信机制。最常用的就是队列multiprocessing.Queue。注意这里不能使用标准库的queue.Queue那个队列只能在同一进程内的线程之间工作跨进程存取会直接报错。一个典型的生产者消费者模型长这样import multiprocessing as mp import time def producer(q: mp.Queue, total: int): for i in range(total): q.put(i) q.put(None) # 终止哨兵 def consumer(q: mp.Queue): while True: item q.get() if item is None: break print(f消费 {item}) time.sleep(0.01) if __name__ __main__: q mp.Queue(maxsize100) p1 mp.Process(targetproducer, args(q, 50)) p2 mp.Process(targetconsumer, args(q,)) p2.start() p1.start() p1.join() p2.join()队列是 FIFO 的先进先出结构put在队列满时会阻塞get在队列空时会阻塞。这个阻塞特性既是好处也是问题如果消费者处理完所有数据后没有收到“结束信号”它会永远卡在get上。所以我在生产者最后塞入None作为哨兵消费者收到None就退出。这里有一个非常经典的坑如果消费者进程不止一个一个None不够用。比如两个消费者第一个抢到了None退出第二个还在q.get()里傻等。解决办法是把哨兵也变成 N 份在生产者结束时放入与消费者数量相同的None。很多人在单消费者时测试没问题一改成多消费者就发现有一两个进程挂在那里不动八成就是哨兵数量没配对。3.2 Pipe 双向通信与它的脾气Pipe是另一种通信方式它在两个进程之间建立一条管道两端各持有一个连接对象parent_conn, child_conn mp.Pipe() def child_sender(conn): conn.send({status: ok, value: 42}) conn.close() if __name__ __main__: p mp.Process(targetchild_sender, args(child_conn,)) p.start() msg parent_conn.recv() print(msg) p.join()send和recv是一一对应的发送方send一条数据接收方必须recv一条数据。Pipe通常比Queue快因为它更轻量不像Queue那样内部还要开一个线程来管理缓冲。但Pipe的脾气也更坏两端同时向对方send大量数据又同时等待recv可能引发死锁。管道是点对点的只适合两个进程之间通信不适合一对多广播。默认创建的Pipe是双向的如果只想单向传递可以传duplexFalse。日常使用中我倾向于先用Queue因为它的语义更宽容适合生产者和消费者数量不对称的模型只有当两个进程之间需要高频交换小块数据、对延迟非常敏感时才换成Pipe。3.3 共享内存 Value 与 Array第三种方式是直接在进程间共享一小块内存使用Value和Arrayimport multiprocessing as mp def increment(counter, lock): for _ in range(1000): with lock: counter.value 1 if __name__ __main__: counter mp.Value(i, 0) lock mp.Lock() processes [ mp.Process(targetincrement, args(counter, lock)) for _ in range(4) ] for p in processes: p.start() for p in processes: p.join() print(counter.value)mp.Value(i, 0)里的i是类型代码表示有符号整数。Array的用法类似比如mp.Array(d, [0.0, 0.0, 0.0])其中d表示双精度浮点数。共享内存的读写性能比Queue高很多但也更原始。counter.value 1表面上是一句赋值实际上是“读取到 Python 层、加一、写回”三步。两个进程如果同时执行很容易互相覆盖所以要配合Lock使用。如果完全不希望有额外锁开销可以用mp.RawValue和mp.RawArray但那就等于把同步责任完全交给了业务代码新手我不建议碰。3.4 三种通信方案的快速对照方案方向阻塞行为性能推荐场景Queue一对多、多对一put/get 都会阻塞中等有序列化开销任务分发、结果收集、生产者消费者Pipe两进程点对点send/recv 需要配对较高两个进程高频通信Value/Array多进程共享内存无内置阻塞最高计数器、共享状态、数值更新这三者不是互斥的很多正经项目会同时用到两个或多个。比如一个图像处理流水线主进程用Queue把任务分给 8 个 workerworker 之间用Array共享一块中间缓存最后再用Queue把结果收回来。通信方案的选择应该跟着数据流向走不要一上来就挑性能和复杂度都最高的方案。4. 同步原语Lock、Event 与信号量避免进程打架4.1 日志打印与计数器的锁多进程之间没有共享变量看似不会出现线程安全的问题但实际上多个进程同时写同一个文件、同一个终端、同一块共享内存时照样会互相干扰。最直观的例子就是打印日志import multiprocessing as mp def worker(idx: int, lock: mp.Lock): with lock: print(f[{idx}] start) time.sleep(0.5) print(f[{idx}] end) if __name__ __main__: lock mp.Lock() processes [ mp.Process(targetworker, args(i, lock)) for i in range(5) ] for p in processes: p.start() for p in processes: p.join()没有锁的时候多个进程的print输出可能互相穿插比如[0] start和[1] start挤在同一行。加上Lock之后执行到with lock:的进程会独占这段代码其他进程必须等它完成才能进入。Lock的另一个常见用途是保护共享内存更新。前面提到的counter.value 1如果没有锁4 个进程各自同时读到一个旧值结果会小于预期。这一点和线程里的非原子问题几乎一模一样只是发生原因从“线程交替”变成了“不同进程独立并发”。4.2 Event 让进程等待信号Event是一种更细粒度的同步方式它维护一个内部标志位布尔值为真表示事件已发生。常用方法只有三个set()把标志设为真wait()阻塞直到标志为真clear()把标志重置为假。典型场景是主进程需要等所有 worker 就绪之后再统一开跑def worker(ready: mp.Event, start: mp.Event): ready.set() start.wait() print(开始干活) time.sleep(1) print(干完了)Event的语义和线程里的几乎一致非常适合做“栅栏”“发布信号”“停止信号”这类控制流的传递。与Queue不同它不传业务数据只传一个开关状态因此几乎没有序列化开销在控制流同步场景里效率很高。4.3 Semaphore 限制并发资源Semaphore可以理解为一个计数器初始值代表允许同时进入的进程数量。每次调用acquire()会让计数器减一计数器为零时后续进程会阻塞调用release()计数器加一。这个特性适合限制对稀有资源的并发访问。场景很具体假设一个任务需要访问某个数据库连接池而这个连接池最多只支持 3 个并发连接。不用信号量时进程池里 8 个 worker 可能同时涌过去把连接池打爆加上信号量后只有 3 个 worker 能进入连接区db_semaphore mp.Semaphore(3) def query_db(query: str): with db_semaphore: print(f执行查询: {query})信号量不是万能的它只控制“并发数量”不关心资源是否真的就绪也不做资源分配。但如果只是要限制并发上限它比手写计数器简洁可靠得多。5. 避坑清单fork 与 spawn、Pool 嵌套、进程残留与调试5.1 fork 和 spawn 的启动方式差异multiprocessing创建子进程的方式在不同操作系统上不同。Linux 上默认通常是fork它会直接复制父进程的内存子进程从创建那一刻的状态继续运行启动速度快。Windows 和 macOS 上默认是spawn它会重新启动一个干净的 Python 解释器再导入主模块来执行目标函数启动慢但隔离干净。这个差异带来的后果很隐蔽。fork方式下子进程会继承父进程当时的内存状态包括已经创建了一半的对象、未释放的锁、临时文件句柄。如果父进程里某个线程恰好持有一把锁然后又创建了子进程子进程即使永远等不到那个线程来释放锁也会认为锁是“已被持有”一旦执行到acquire就死锁了。spawn方式因为是完全新解释器就不会继承这些脏状态所以现在官方文档也更推荐用spawn或forkserver牺牲一点启动时间换取确定性。如果你观察到一个多进程程序在 Linux 上跑得好好的换到 Windows 平台就频繁报错甚至创建进程时无限循环那多半是启动方式变化引起的。我的建议是尽量别把代码写成依赖某个平台的启动方式至少保证if __name__ __main__存在并且把创建进程池的代码放在主模块入口中不要放在被导入的模块顶层。5.2 用 Pool 时最容易碰到的死锁Pool 嵌套Pool嵌套是我见过最多的一个隐藏死锁原因。表面上看在 worker 函数内部再开一个Pool或者再调用同一个Pool的方法似乎是合理的其实很容易让整个进程池卡死。举一个典型的错误写法from multiprocessing import Pool def outer_task(x): with Pool(4) as inner_pool: return inner_pool.map(lambda y: y 1, [x] * 100)如果外层Pool也设置了 4 个进程四个 worker 都在执行outer_task它们又各自创建内层Pool等于一下子要 16 个进程数量一多资源马上耗尽。更隐蔽的是如果 worker 函数内部直接调用了同一个Pool实例的方法比如pool.apply_async(...)而当前所有 worker 都在执行任务、没有空闲进程去处理内部调用这个内部调用会永远在等待池中有空闲进程形成互相等待的死循环。解决办法通常是把任务层级压平让所有任务都在一层进程池中处理。如果必须分层次我会用显式的Process加Queue自己设计调度而不是把Pool当通用调度器随意嵌套。写并行代码时永远记住Pool是给“一次性提交给固定进程组”的任务设计的不是给递归式动态调度设计的。5.3 进程残留与优雅退出写过一段时间多进程的人多半见过“脚本已经结束任务管理器里还有一堆 Python 进程”的现象。这种情况多半是子进程没有被正确回收。Process对象创建后调用start()启动子进程调用join()等待它结束。如果子进程是循环结构或者监听队列主进程直接退出时子进程并不会自动跟着退出它们会成为孤儿进程继续运行。几个常用的清理手段给子进程设置daemonTrue主进程退出时它会自动终止。但这个属性必须在start()之前设置且daemon进程内部不能再创建子进程。子进程退出前要显式调用terminate()或者放开队列哨兵让循环有机会结束。使用multiprocessing.active_children()查看当前还活着的子进程用于调试。对Pool来说正确的关闭顺序是调用close()停止接收新任务然后调用join()等待现有任务结束想要立即终止且不关心未完成任务才用terminate()。with Pool上下文管理器其实比很多人想象得更“暴力”它离开时会直接terminate()而不是等待任务结束。如果你在with块里用了异步提交但忘了等待子进程会被硬生生杀掉。这也是一个常见误用点。5.4 调试多进程程序的实用手段多进程程序一旦出错报错往往分散在各子进程里串行程序那一套print调试法就不太好用了。我这些年形成了一套自己的调试流程。先把进程数降为 1确认业务逻辑正确。Pool(1)和多进程环境一致但行为退化成单进程跑一遍能排除大部分逻辑问题。然后加日志不要只用print最好带上进程 PID 和时间戳格式类似import os import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) def worker(item): logger.info(pid%s 处理 %s, os.getpid(), item)再就是给异步结果加超时。AsyncResult.get(timeout...)可以在任务卡住时抛异常退出避免整个程序无限阻塞。最后尽量把 worker 写成纯函数不要依赖任何全局状态。多进程里每个进程都有自己的一份全局变量fork方式下初始值一样但后续修改互不可见如果业务代码里悄悄依赖了“某个全局变量应该被其他进程更新过”这种假设就会得到非常难排查的错乱数据。6. 实测场景一次 CPU 密集任务的并行化记录6.1 用一个什么任务做示例为了把前面的知识点串起来我打算用一个简单但 CPU 密集的示例统计一个区间内质数的数量。这个任务的特点是纯计算、任务块之间无依赖、每个块的计算耗时稳定非常适合观察多进程的加速效果。为了模拟真实项目的压力我把统计范围设得足够大计算[2, 50_000_000)区间内的质数总量使用试除法判断每一个数。算法不讲究最优重点是让它能稳定占用 CPU让并行效果一目了然。6.2 单进程基线版本先写单进程版本作为基线import math def is_prime(n: int) - bool: if n 2: return False if n % 2 0: return n 2 i 3 while i * i n: if n % i 0: return False i 2 return True def count_primes(limit: int) - int: cnt 0 for n in range(2, limit): if is_prime(n): cnt 1 return cnt在我的测试机器上这版本跑完大约耗时 42 秒。测试环境是一台 8 核 16 线程的机器内存 32GB系统为 LinuxPython 版本 3.12。单进程阶段CPU 占用基本维持在 12% 左右其余核心完全空闲这就是典型的 GIL 限制表现因为这里根本没有开线程或进程一个进程只能用一个核。6.3 多进程改造与参数调整多进程版本的核心是任务切分。把整个区间切成若干个独立的小块每个子进程负责一块最后把统计结果加起来from multiprocessing import Pool def count_primes_in_range(bounds) - int: lo, hi bounds cnt 0 for n in range(lo, hi): if is_prime(n): cnt 1 return cnt def make_chunks(total: int, chunk_size: int): start 2 while start total: yield (start, min(start chunk_size, total)) start chunk_size if __name__ __main__: total 50_000_000 chunk_size 1_000_000 chunks list(make_chunks(total, chunk_size)) with Pool(8) as pool: results pool.map(count_primes_in_range, chunks) print(sum(results))这里有两个调参点。第一个是chunk_size它决定每个任务块的规模。我最终选了 100 万也就是切成约 50 块远远大于进程数保证每个 worker 结束时还能有后续任务避免负载不均。第二个是进程数我准备分别用 1、2、4、8、16 个进程跑一遍记录耗时和加速比。6.4 实测数据与结果解读实测结果如下耗时受机器负载影响会有波动但相对关系是稳定的进程数耗时秒相对单进程加速比142.81.0x222.31.92x411.93.60x86.46.69x166.17.02x2 个进程时几乎完美线性4 个进程也接近线性。到 8 个进程时加速比约 6.7 倍已经出现了一部分损耗主要来自进程调度、结果聚合和任务切分的开销。到 16 个进程时几乎不再增长因为机器物理核只有 8 个多的 8 个逻辑核来自超线程纯计算任务里收益极其有限。这个结果直接说明了一件事进程数不是越大越好。对这台机器8 进程是性价比最高的选择16 进程看起来和 8 进程差距很小但内存占用更高进程切换更频繁如果任务里还涉及大量临时对象反而可能更慢。6.5 实测之后的几条心得体会这次实测之后我给自己定了几条容易执行的原则。任务切块的粒度要把单块耗时控制在 0.2 秒到 1 秒之间。块太小调度和通信的开销占比会上升块太大某些 worker 提前结束时其他 worker 还在跑最后一段只能眼睁睁等待。我这次把一千万的区间切成几十块就是基于这个经验。如果你无法预估单块耗时就用二分法试跑先切成 10 块看耗时再切 50 块看差异热血调到一个拐点为止。函数内部如果会创建大量临时对象记得给Pool设置maxtasksperchild参数。它每执行完 N 个任务会重建一个新的 worker 进程等于定期清理子进程内存防止长时间跑下来内存碎片化越来越严重。这个参数平时用不上但一旦任务量到几十万级效果立竿见影。最后一件事多进程代码写完一定要做终止测试强制确认所有子进程在脚本退出后都消失。跑完上面的示例后我会执行一句检查当前进程树的命令看到只剩主进程才算结束。很多线上问题出在“任务完成了但下一轮启动时旧进程还在吃内存”这种问题最难查因为它不影响单次运行只影响长期稳定性。这一版例子我先整理到这里。后续如果multiprocessing在新版本中出现会影响使用习惯的变化或者我又遇到新的有意思的坑我会再回来补充。里面绝大多数代码是我实际跑过之后简化出来的版本如果你照着写仍然遇到奇怪的问题建议优先检查第五章绝大多数病根都藏在那里。