干过Python并行开发的人应该都经历过这么一幕写好的多进程脚本跑起来子进程吭哧吭哧算半天最后你发现主进程里的list还是空的或者dict还是初始值。最气人的是去技术社区问一句Python多进程怎么往同一个list里写东西底下回答五花八门有说加锁的有说用global的还有说直接append就行——照着试全翻车。这个标题里的写同一个list/dict本身就是个容易误导人的说法。今天我把这个问题的原理、可行方案、性能和坑一次性讲透重点介绍multiprocessing.Manager、Queue、共享内存三类主流做法以及每类方案在什么场景下才值得用。文章适合那些已经会用multiprocessing启动子进程但还没真正搞懂进程间数据如何同步的开发者看完可以直接照抄。1. 为什么多进程写同一个list/dict这句话本身就是个坑1.1 先看一段一定会翻车的代码假设你有一段处理数据的任务想把每个子进程的计算结果累积到一个全局list里from multiprocessing import Process import time result_list [] result_dict {} def worker(idx): # 模拟耗时计算 time.sleep(1) result_list.append(idx * 2) result_dict[idx] idx * idx if __name__ __main__: processes [] for i in range(4): p Process(targetworker, args(i,)) processes.append(p) p.start() for p in processes: p.join() print(list:, result_list) # 竟然打印 [] print(dict:, result_dict) # 竟然打印 {}跑完之后子进程里的append和赋值好像都没发生过。有同学会在worker里加一句print发现子进程明明执行了啊list也有内容啊怎么一回到主进程就空了。1.2 翻车的原因进程内存是独立地址空间关键就一句话multiprocessing创建子进程时会把父进程当前的内存状态复制一份给子进程但之后两者各过各的不再共享任何普通对象。这里的复制不是Java里那种引用传递而是在操作系统层面为子进程分配了独立的地址空间。你把result_list传给子进程子进程拿到的其实是同一份对象在它自己地址空间里的副本对这个副本做append真实写入的是子进程内存页主进程那边看到的还是原来那个空list。有人可能会提写时复制Copy-on-Write说fork时父子进程不是共享物理内存页吗没错fork确实让父子进程先共享同一批物理页但只要有一方去写这些页内核就会把对应的页复制一份再修改。子进程执行append时这个list所在的内存页被复制到子进程私有空间修改从此与父进程无关。所以不仅看不见就算在子进程里反复修改也不会对父进程有任何影响。这跟多线程有本质区别线程共享同一进程的堆空间一个线程new出来的对象其他线程通过同一个引用马上就能看到。进程则是复制整个世界然后在自己的世界里生活。1.3 为什么多线程的直觉在这里失灵我见过太多人把threading的经验直接搬到multiprocessing上。在threading里你写shared_list.append(x)所有线程操作的是同一个list所以问题只在并发冲突加个Lock搞定。但在多进程里问题压根不是并发冲突而是你操作的list根本不是主进程的那个list。再叠加Python的GIL很多人的误区就更深了。多线程因为GIL的存在CPU密集任务反而变慢于是大家转向多进程来用满多核。但进程间数据的传递方式完全变了多线程靠共享内存多进程必须靠进程间通信IPC机制比如队列、管道、共享内存、信号量。你没法再用大家把数据往同一个容器里丢这种思维除非那个容器本身挂在某个进程间共享的通道上。理解了这一点下面所有方案都会变得很好懂——无非是用什么机制让子进程的写入能被主进程看到。2. 最贴合标题的方案multiprocessing.Manager的代理容器2.1 Manager的本质中间人进程multiprocessing.Manager大概是所有想直接写同一个list/dict的人第一反应会找到的答案。它的原理并不神秘启动一个独立的manager进程由它来维护真正的list和dict对象你拿到的manager.list()返回的其实是一个代理对象Proxy。子进程对这个代理执行append时操作被序列化后发送给manager进程由manager进程在自己内存里修改真正的list主进程下次访问时再把最新状态拿回来。说白了manager就是给所有进程提供了一个公共前台大家去前台办业务前台背后才是真正的账本。整个过程涉及序列化和网络通信本地socket所以注定比直接操作内存慢但换来的是看起来像在操作普通list/dict的接口。2.2 正确写法和两个容易忽略的注意事项用Manager改造刚才的代码只需要把list/dict换成代理对象import multiprocessing import time def worker(shared_list, shared_dict, idx): time.sleep(1) shared_list.append(idx * 2) shared_dict[idx] idx * idx if __name__ __main__: with multiprocessing.Manager() as manager: shared_list manager.list() shared_dict manager.dict() processes [] for i in range(4): p multiprocessing.Process( targetworker, args(shared_list, shared_dict, i) ) processes.append(p) p.start() for p in processes: p.join() print(list:, shared_list) print(dict:, shared_dict)注意两点。一是尽量用with multiprocessing.Manager() as manager的写法退出时保证manager进程被正确关闭否则某些平台上可能留下残留进程。二是args里直接传代理对象就行不需要层层转发子进程拿到代理后会自己去跟manager进程通信。2.3 Manager不是万能药复合操作仍要加锁很多人以为用了Manager就彻底安全了这是个危险误解。Manager提供的list和dict单个方法调用确实是原子操作比如append、d[key] value因为每次调用都是一次独立的IPC请求。但多个方法组合在一起比如先判断key是否存在再赋值或者检查list长度后追加就会读出旧数据再写回中间可能被其他进程打断。我处理过一个真实案例8个子进程同时往manager.dict()里做if key not in d: d[key] []结果多个进程都判断key不存在各自创建新列表最后互相覆盖。这种检查再更新必须在外层加锁lock manager.Lock() def worker(shared_dict, lock, idx): with lock: if idx not in shared_dict: shared_dict[idx] [] shared_dict[idx].append(idx)Manager还自带Lock、Semaphore、Condition别自己再去import多线程的Lock那就是另一条路子了。另外Manager在Windows和Linux下都能用但Windows使用spawn方式启动子进程所有传给Process的参数都必须能pickleManager代理对象本身没问题可如果你在worker里用了lambda表达式在Windows上会直接报错。3. 换一个思路用Queue把结果上交给主进程3.1 生产消费模型为什么比共享容器更顺如果你只是需要把每个子进程的结果收集起来那么本质上你要的不是共享一个list而是让每个子进程把结果交到某个统一的地方。这种情况下multiprocessing.Queue往往比Manager更合适。Queue内部基于管道和缓冲区自带线程安全/进程安全的put和get数据从写端进入从读端取出天然不会出现两个进程同时改同一个内存位置的问题。子进程不需要知道别人写了什么只需要把结果put进来主进程负责get最后组装成list或dict。这完全符合生产-消费模型代码也更直白import multiprocessing import time def worker(q, idx): time.sleep(1) q.put((idx, idx * 2, idx * idx)) if __name__ __main__: q multiprocessing.Queue() processes [] for i in range(4): p multiprocessing.Process(targetworker, args(q, i)) processes.append(p) p.start() result_dict {} for _ in range(4): idx, double_value, square_value q.get(timeout10) result_dict[idx] {double: double_value, square: square_value} for p in processes: p.join() print(dict:, result_dict)这里主进程直接把每个结果放进dict子进程从头到尾只接触Queue不需要什么共享list最终成果反而更干净。3.2 收尾阶段最容易踩的坑还有多少条没读完用Queue很多人的习惯是先join所有子进程再while not q.empty() get数据这个顺序在数据量大时很容易丢数据。原因在于Queue的底层缓冲区是异步的子进程put之后立刻返回数据可能还在内存缓冲里还没有完全刷到管道对端。如果你立刻join子进程再判断q.empty()可能缓冲区还没被后台线程完全写入主进程就认为队列空了。正确姿势有几种。最简单的是先get完所有预期数据再join子进程像我上面那段代码因为知道总共4个子进程每条任务一个结果所以循环4次get即可。如果任务数未知可以用q.close()配合q.join_thread()确保写入完成或者干脆在子进程put后put一个哨兵值比如None主进程读到哨兵就知道结束。我个人最推荐按任务数收数据最不容易出问题。3.3 其实大多数人可以直接用Pool.map聊到收集结果还得提一句很多人根本不需要手动启动Process。如果你的任务是一个输入数据对应一个结果这种典型并行映射multiprocessing.Pool的map方法就是专门干这个的它内部已经处理好了任务分配和结果收集import multiprocessing def add_one(x): return x 1 if __name__ __main__: with multiprocessing.Pool(processes4) as pool: results pool.map(add_one, range(8)) print(list:, results)这段代码简洁到几乎没有出错的余地。Pool.map会阻塞等待所有任务完成返回一个和输入顺序一致的list。如果你想在任务完成后逐个处理结果可以用imap/imap_unordered。如果任务太多内存放不下还可以用imap惰性取值。这也是我给新人最常提的建议先想想能不能用Pool能的话就别去折腾Manager和Queue。Pool底层虽然也用队列但框架帮你把收集的活干了。4. 高性能场景放弃list/dict形态用共享内存4.1 共享内存的基本玩法Manager慢Queue也有序列化开销。如果你的子进程要频繁写入大量数据每秒几十万次那种再走进程间通信路线就扛不住了。这时候该上共享内存。Python自带的multiprocessing.Array和multiprocessing.Value封装了共享内存底层是一块各进程都能直接访问的物理内存区域读写不需要经过序列化和socket所以比Manager快一到两个数量级。代价是这些接口只支持C语言原生类型不支持list/dict这种Python对象结构。换句话说你得把自己的数据压扁成数组或结构体。4.2 用共享数组模拟一个固定容量dict假设你要让多个子进程往同一个字典里写数据key是整数value是浮点数已知最多100个键。可以这样设计用一个整型数组第0个位置维护当前已写入的键数量后面的位置按key, value交替排列再加一个锁防止两个进程同时写入同一个槽位。import multiprocessing def worker(shared_array, lock, idx): # 数据计算 key idx * 10 value idx * 0.5 with lock: # 找到当前数组写入位置 pos 1 shared_array[0] * 2 shared_array[pos] key shared_array[pos 1] value shared_array[0] 1 if __name__ __main__: # 容量100个键数组长度1(计数器)100*2 arr multiprocessing.Array(d, 201) lock multiprocessing.Lock() processes [] for i in range(4): p multiprocessing.Process(targetworker, args(arr, lock, i)) processes.append(p) p.start() for p in processes: p.join() for i in range(arr[0]): key int(arr[1 i * 2]) value arr[1 i * 2 1] print(key, -, value)这个方案的关键是把字典是否已有这个键的判断变成写入一个新条目因为每条任务写一个不同的键所以不存在重复覆盖问题。如果多个进程可能写同一个键那就得先遍历数组查一下key是否已存在或者用另一个共享数组记录空闲槽位的状态。4.3 写共享内存时锁和原子性怎么把控multiprocessing.Array(d, 201)默认自带进程锁单个元素的读写是原子的但查找到哪个位置再写入这个过程并不是原子的所以必须自己加Lock。一个常见坑是忘记加锁多个子进程同时执行pos 1 shared_array[0] * 2读到同一个位置然后互相覆盖。这种bug比Manager的检查再更新更难查因为它不是必现只在进程交错执行的瞬间发生。如果你想完全不写锁可以用multiprocessing.RawArray配合multiprocessing.Value的原子自增操作。不过说实话这种极限优化对绝大多数项目没有必要加Lock的实测开销远小于一次进程通信正确性远比那几百纳秒的锁开销重要。先把锁加上用perf_counter测一下吞吐如果还满足不了需求再考虑无锁方案。还有一点Python 3.8之后有multiprocessing.shared_memory.SharedMemory它给你一块裸的共享字节区可以配合numpy的frombuffer直接做高性能数值计算。如果你已经在用numpy处理数据这个方案非常顺手但实现复杂度又高一层我建议先掌握Array/Value这套再进阶。5. 所有方案对比与我的选型建议5.1 一张表把方案说明白这里把几类核心方案按性能、复杂度、适用条件放在一起方便直接对照方案是否保持list/dict形态性能代码复杂度适用场景Manager.list / dict是慢每次操作都是序列化IPC低小数据量、低频写入、需要跨平台Queue 主进程聚合否结果在主进程组装中等每次put/get有序列化中任务结果收集、生产消费模型Pool.map / imap否框架直接返回list中高极低输入输出一一对应的并行计算Array / RawArray Lock否需要自定义结构和索引高接近内存速度高高频数值写入、性能瓶颈场景SharedMemory numpy否裸字节内存最高很高大数据块、数值计算密集场景单看这张表你会发现多进程写同一个list/dict这个诉求本身也可以做技术降级多数时候你需要的不是共享容器而是高效地把子进程运算结果汇总成容器。想清楚这一点方案选择就豁然开朗了。5.2 我在真实项目中踩过的坑和排查过程有一次我在做数据聚合项目一开始图省事用了Manager.dict10个子进程每秒各写几百个键值对结果主进程打印结果时发现数据对不上最诡异的是单次运行结果还不一样有时候丢几个键有时候重复覆盖。排查过程我至今记得。第一步是怀疑代码逻辑给每个子进程打印它写入的键确认没有逻辑错误。第二步怀疑Manager代理的原子性查官方文档发现单方法原子但复合操作不原子于是我在所有检查再更新的地方加了manager.Lock问题依旧。第三步我开始怀疑Manager本身在高并发下的数据一致性做了个压测脚本100万次写入统计最终数量。结果发现即便加了锁Manager在小规模测试没问题压力一上来还是偶尔少数据。最后定位到原因manager进程是单线程逐条处理代理请求的当多个子进程同时提交复合操作时锁本身在manager进程内部并不能完全兜住所有跨方法的时序问题。正确的做法是把复合操作封装成一个manager暴露的自定义方法让它整体在manager进程内原子执行或者干脆改用Queue。那次之后我的原则就变成了需要复杂共享逻辑就别用Manager需要性能就别用ManagerManager只适合低频次、key值独立、单方法操作的轻量协作。另一个印象很深的坑来自Windows环境在Linux下.run()里传个函数给Process就完事到了Windows却报无法pickle本地函数。我后来养成习惯所有多进程相关的函数都写在模块顶层不写在if __name__的代码块内部参数只传可pickle的数据类型和Manager代理对象再也没出过这类问题。5.3 最后一次经验总结虽然标题是写同一个list/dict但真正写代码时我的决策路径其实很固定如果数据量小、操作简单、还要list/dict接口用Manager如果是任务结果汇总优先Pool.map如果任务之间接收的是流式数据用Queue如果高频写操作已经变成性能瓶颈那就放弃list/dict的执念上共享内存。每次给同事review代码看到他们纠结怎么让多进程共享dict的时候我都会反问一句你想要的究竟是共享这个dict还是拿到最终汇总结果大多数时候答案是后者。想明白这个问题很多技术选型就简单了。希望这篇文章能帮你少走一点弯路。
