前言多进程有一条铁律进程之间不共享内存。每个进程有自己独立的地址空间你在子进程里改了一个变量父进程看不到任何变化。这不是缺陷而是隔离带来的稳定性——一个进程崩了不会污染另一个。代价是想让数据在两个进程之间流动必须显式地「传」过去。而「传」这件事有成本。多进程传递数据要走序列化pickle发送方把对象变成字节流接收方再还原成对象。这带来两个直接的约束——传过去的东西必须可 picklelambda、打开的文件句柄、生成器都不行并且传大对象时要掂量序列化和复制的开销。这也解释了为什么共享内存在某些场景下更快它绕开了「把整个对象序列化一遍」这一步。本文用一个「多进程分工 汇总结果」的小项目串起multiprocessing的四类通信手段Queue队列、Pipe管道、Manager代理对象、Value/Array共享内存并说清各自的适用场景与代价。示例基于 Python 3.8 及以上所有需要启动子进程的代码都写在if __name__ __main__:保护块里原因见后文坑点。还有一处旧写法提醒Python 2.7 已于2020 年 1 月 1 日停止维护它的dict.iteritems()/iterkeys()在 Python 3 里已移除。下面用Manager().dict()做汇总时遍历结果写法是for k, v in shared.items():返回视图而不是shared.iteritems()。一、项目把一批日志分片统计出来需求很朴素有一批文本块要统计每个块的行数、最长行长度和字符数最后汇总。单进程当然能做但这里特意拆成「主进程分发任务、多个工作进程并行处理、结果汇总回主进程」用来展示通信的两条路——任务往下走结果往上走。# 适用于 Python 3.8# project.py —— 完整的任务分发 结果汇总示例import multiprocessing as mpdef make_chunks(names):主进程里执行把『文件名』展开成待处理的数据块。真实项目里这里可能是读文件为了让示例自洽用合成内容代替。return [(name, f{name} 的第 1 行\n{name} 的第 2 行\n) for name in names]def summarize(chunk):【工作进程里执行】统计一个块的 (行数, 最长行长, 字符数)。这个函数必须定义在模块顶层否则无法被 pickle 传到子进程。name, text chunklines text.splitlines()longest max((len(line) for line in lines), default0)return name, len(lines), longest, len(text)def worker(task_q, result_q):【工作进程里执行】不断取任务直到遇到哨兵 None 才退出。while True:chunk task_q.get() # 阻塞等待任务if chunk is None: # 哨兵通知本进程可以收工了breakresult_q.put(summarize(chunk)) # 结果序列化后送回主进程主进程负责三件事把任务放进队列、启动并等待工作进程、从结果队列收结果。# 适用于 Python 3.8def main():names [fapp{i}.log for i in range(8)]chunks make_chunks(names)task_q mp.Queue() # 任务队列主进程 → 工作进程result_q mp.Queue() # 结果队列工作进程 → 主进程# 【主进程】先把任务全部放进去再启动进程避免子进程空等for chunk in chunks:task_q.put(chunk)n_workers min(4, len(chunks))workers [mp.Process(targetworker, args(task_q, result_q))for _ in range(n_workers)]for w in workers:w.start() # 【主进程】启动之后 worker 里的循环开始跑# 【主进程】每个工作进程放一个哨兵它取到 None 就结束for _ in range(n_workers):task_q.put(None)# 【主进程】结果条数等于任务条数逐个收results [result_q.get() for _ in chunks]for w in workers:w.join() # 【主进程】等工作进程全部退出total_lines sum(r[1] for r in results)longest_all max(r[2] for r in results)print(f共 {len(results)} 个块合计 {total_lines} 行最长行 {longest_all} 字符)if __name__ __main__:main()执行python project.py主进程会把 8 个任务放进task_q启动至多 4 个工作进程每个工作进程反复get()任务、算出三元组、put()结果直到读到哨兵None退出。主进程收满 8 条结果后join()退出。这里有几个刻意的设计属于「实践」里最值得抄的部分任务先入队再start()。反过来的话工作进程可能先跑起来、发现队列空而空转。哨兵的数量等于工作进程数。每个worker读到自己的那个None才break少放一个就会有进程永远阻塞在get()。也可以用JoinableQueue的task_done()/join()替代哨兵但哨兵更直观。结果的条数已知。因为「一个任务一条结果」result_q.get()可以安心地循环固定次数。如果结果条数不确定就要靠哨兵或超时来收尾。二、Queue最常用的去向与回报multiprocessing.Queue是用「一个管道 若干锁和信号量」实现的进程共享队列。官方说明的有两点很关键任何放进队列的对象都会被序列化队列是线程和进程安全的。第一次put()时它会启动一个「供给线程feeder thread」把缓冲里的对象搬到管道里。方法作用备注put(obj[, block[, timeout]])放一个对象默认阻塞直到有空位get([block[, timeout]])取一个对象超时抛queue.Emptyput_nowait/get_nowait不阻塞版本立即失败或立即返回empty()/full()/qsize()查看状态官方明确说不可靠close()/join_thread()释放资源join_thread需在close之后官方特地写明了qsize()「由于多线程/多进程语义这个数字不可靠」empty()/full()也一样不可靠。所以不要用if q.empty():来判断「是不是没活干了」——这是并发下最经典的错判来源。三、Pipe两个进程之间的专线Pipe()返回一对连接对象默认是双向duplex的。每个连接对象都有send()与recv()。它适合「就两个进程一对一」的场景比队列轻。# 适用于 Python 3.8import multiprocessing as mpdef child(conn):【子进程里执行】收到请求回一个响应。msg conn.recv() # 阻塞等父进程发来conn.send({echo: msg, len: len(msg)})conn.close() # 用完自己这一端就关掉if __name__ __main__:parent_conn, child_conn mp.Pipe() # 得到一个双向管道p mp.Process(targetchild, args(child_conn,))p.start()parent_conn.send(hello)print(parent_conn.recv()) # {echo: hello, len: 5}p.join()文档给了两条硬性提醒send()会序列化对象、recv()会重建对象所以传的东西同样要可 pickle另外如果两个进程或线程同时读写管道的同一端数据可能损坏——用不同端并发则没有问题。所以Pipe的正确用法是「一端一个进程」不要把一个连接对象再分给多个进程共享。四、Value / Array共享内存Value和Array把数据放在共享内存里多个进程直接读写同一块区域不需要来回复制。构造时要给 ctypes 的类型码d是双精度浮点i是有符号整数。# 适用于 Python 3.8import multiprocessing as mpdef bump(counter, lock):【子进程里执行】给共享计数器加一加锁保证读改写不交错。for _ in range(100):with lock:counter.value 1if __name__ __main__:counter mp.Value(i, 0) # 共享的整数lock mp.Lock() # 共享的锁procs [mp.Process(targetbump, args(counter, lock)) for _ in range(4)]for p in procs:p.start()for p in procs:p.join()print(counter.value) # 400Array用法类似mp.Array(i, range(10))建一个有 10 个整数的共享数组。文档说明这些共享对象是进程与线程安全的但它们只能用 C 类型——想共享一个字典或一组字符串Value/Array做不到得换Manager。另外需要更灵活的共享内存比如一块裸字节缓冲区时标准库从 Python 3.8 起提供了multiprocessing.shared_memory模块。要强调的一点共享内存在这里指的是「专门创建的共享对象」不是「普通变量自动共享」。counter之所以能被四个进程一起改是因为它由mp.Value创建、内部指向一块共享内存并且加了锁如果你把counter 0这样的普通整数传进子进程每个子进程拿到的都是自己那份副本改完照样互不可见。五、Manager让任意对象都能被共享Manager()返回的管理器对象会启动一个服务器进程它持有真正的 Python 对象其他进程通过代理proxy去操作这些对象。文档说明管理器支持list、dict、set、Namespace、Lock、RLock、Semaphore、BoundedSemaphore、Condition、Event、Barrier、Queue、Value、Array等类型。# 适用于 Python 3.8import multiprocessing as mpdef collect(shared, name):【子进程里执行】把本进程的结果写进共享字典的对应条目。shared[name] len(name) # 走代理实际对象在服务器进程里if __name__ __main__:with mp.Manager() as manager:shared manager.dict() # 共享字典其实是代理procs [mp.Process(targetcollect, args(shared, n))for n in (alpha, beta, gamma)]for p in procs:p.start()for p in procs:p.join()print(dict(shared)) # {alpha: 5, beta: 4, gamma: 5}文档对它的定位说得很清楚管理器比共享内存更灵活可以支持任意对象类型但更慢而且单个管理器可以让不同机器上的进程通过网络共享。所以选型上可以记成——Value/Array求快、类型受限Manager求灵活、有额外开销。方式能传什么是否共享内存相对开销典型场景Queue任意可 pickle 对象否走管道 序列化中任务分发、结果汇总Pipe任意可 pickle 对象否走管道 序列化低两个进程一对一Manager代理多种容器类型与同步原语否走服务器进程高需要共享 dict / listValue/Array有限 C 类型是低计数器、数值数组shared_memory裸字节缓冲区是最低大块数值数据的零拷贝共享常见坑点坑点 1用q.empty()判断「活干完了没」。❌while not q.empty(): item q.get()—— 文档明确说empty()不可靠任务可能还在管道里没到位。 ✅ 用「哨兵」放一个特定的结束标记或已知数量的get()需要超时用get(timeout...)并接住queue.Empty。坑点 2哨兵数量和工作进程数不一致。❌ 起了 4 个工作进程却只放 1 个None—— 剩下 3 个永远阻塞在get()主进程join()卡死。 ✅ 每个工作进程放一个哨兵数量与进程数相等。坑点 3把不可 pickle 的对象放进队列。❌ 往Queue里put一个lambda、打开的文件对象或生成器 —— 序列化失败报错。 ✅ 传可 pickle 的数据数字、字符串、列表、字典、元组文件改成传路径读文件在工作进程里做。坑点 4以为子进程改了普通变量父进程能看到。❌ 子进程给一个普通全局变量赋值父进程读到的还是旧值。 ✅ 进程间不共享内存要共享就显式用Value/Array/Manager/shared_memory。坑点 5给Value的不加锁。❌counter.value 1在多进程下裸跑 —— 读、加、写三步可能交错结果偏小。 ✅ 配一个mp.Lock()把这段包起来组合操作不会自动原子。坑点 6把Manager当高性能共享内存用。❌ 在紧循环里频繁读写manager.dict()—— 每次访问都要跨进程走代理开销明显。 ✅ 密集数值共享用Value/Array/shared_memoryManager留给「需要共享复杂对象」的场合。坑点 7多个进程共用Pipe的同一端。❌ 把同一个连接对象分给几个进程同时recv()—— 文档明确说数据可能损坏。 ✅Pipe只给两个进程一对一使用多方通信改用Queue。坑点 8忘了入口保护。❌ 在模块顶层直接Process(...).start()—— Windows 与 macOS 默认用spawn启动子进程重新导入主模块时会再次执行这行进程会不断自我复制。 ✅ 所有启动进程的代码放进if __name__ __main__:。总结你想做的事选它代价多个工作进程领任务、交结果Queue序列化 管道开销两个进程一对一往返Pipe仅限两端同端并发会损坏共享一个 dict / listManager服务器进程 代理往返最慢共享一个计数或数值数组Value/Array配Lock类型受限需自己保证原子性共享一大块数值shared_memory要自己管生命周期把「任务往下、结果往上」这条数据流画出来选型就清晰了需要排队就用Queue只涉及两个进程就用Pipe要共享结构复杂的对象就用Manager纯粹共享数值就用Value/Array配锁。无论选哪种都别忘了那两条底层约束——传的东西必须可 pickle进程之间不共享内存。项目规模变大以后出错的地方几乎总是这两条被忽略了而不是 API 用错了。
