Python multiprocessing 多进程共享 list/dict 的避坑指南

几乎每周都会有人在技术交流群里问同一个问题："我用 multiprocessing 起了 8 个进程，把结果追加到一个全局 list 里，跑完后 list 却是空的，到底怎么回事？"再追问下去，往往还能听到更进一步的版本："我换成了 Manager().dict()，子进程里写入的数据，主进程读出来会少一大截。"这个 python 多进程写同一个 list/dict 的使用场景，几乎覆盖了从爬虫采集、批量计算到分布式任务调度的所有并行化改造需求，而它的坑不在语法层，而在操作系统进程模型和跨进程通信机制上。这篇内容就围绕这个典型需求展开，把常见的共享写入方案、各自的性能特征、以及真实项目里反复踩过的坑一次讲透，适合刚接触 multiprocessing 想并行化批处理任务的开发者，也适合已经被 Manager 性能问题折磨过的老手。

1. 先复现问题：多进程里改 list/dict，为什么主进程看不到？

1.1 一个让初学者崩溃的最小示例

很多人的第一版代码长这样：

import multiprocessing result = [] def add_one(i): result.append(i) if __name__ == "__main__": ps = [multiprocessing.Process(target=add_one, args=(i,)) for i in range(8)] for p in ps: p.start() for p in ps: p.join() print(result) # 输出 []

代码没有任何报错，子进程也确实执行了 append，但主进程打印 result 的时候，得到的依然是一个空的列表。这一步直接击穿了很多人对 Python 的直觉：明明多线程里这样写就能共享数据，为什么换成多进程就不行了？

还有很多人会把第二个版本也试一遍，以为把 list 作为参数传给子进程，就能像传引用一样拿到同一个对象：

def add_one(lst, i): lst.append(i) if __name__ == "__main__": result = [] p = multiprocessing.Process(target=add_one, args=(result, 1)) p.start() p.join() print(result) # 还是 []

这次的结果依然让人失望。关键在于，进程启动时，传入的参数会被 pickle 序列化后交给子进程，子进程拿到的是一个完整的新拷贝。对这份拷贝做任何修改，原件都不会变化。如果你是做爬虫或批处理，同时又想在进程之间共享状态，第一步必须先接受这个底层事实：多进程之间，默认不存在内存共享这回事。

1.2 内存隔离的本质：fork、spawn 与写时复制

要理解上面两个示例，需要先弄清楚操作系统创建子进程时到底发生了什么。Linux 上默认的进程创建方式是 fork()，调用 fork 之后，操作系统会复制父进程的页表，父子进程最开始指向同一批物理内存页。为了让效率足够高，系统不会真的立即复制全部内存，而是把这些物理页标记为只读，当某个进程尝试写入其中的某一页时，触发写时复制，把该页单独复制一份出来给写入方。

这就意味着，即使子进程刚刚从父进程 fork 出来，看起来和父进程共享所有内存，但一旦子进程修改 result 这个 list，它写入的那个物理页就被复制成了子进程私有，父进程的页面完全不受影响。Windows 和 macOS 上默认使用 spawn() 模式，情况更直白：操作系统会启动一个全新的 Python 解释器，重新导入主模块，通过 pickle 接收参数，父子进程的数据从最开始就是两份。你可能在本地 Linux 上测试没问题印象，但如果换到 Windows 上跑同样的代码，脚本直接报 RuntimeError 的情况也不少见，因为 spawn 模式还要求主模块有正确的保护入口。

1.3 文件写入与内存写入的本质区别

有人可能会问：那我让子进程写一个文件，主进程后来读文件，为什么可以看到内容？这是因为文件是操作系统层面的共享资源，不归某个进程的虚拟内存管。每个子进程拿到的是同一个文件描述符指向的内核级文件对象，写入的数据最终落到文件系统或者页缓存中，主进程通过独立文件描述符也能读回来。和文件相反，list 和 dict 只是进程本地内存中的对象，没有任何系统级中间层让多个进程共同引用它们，所以写入不互通是必然结果。

理解了这一点，再看问题"多进程写同一个 list/dict"，本质上其实是在问：如何在进程之间搭建一条数据通路。路径无非有三种：通过代理对象包装共享数据、通过消息传递把数据集中到主进程、通过操作系统提供的共享内存或外部存储兜底。下面逐一拆解。

2. 方案一：用 Manager 代理容器，共享写 list/dict 的最直接路径

2.1 Manager().list() 与 Manager().dict() 的基本用法

multiprocessing 模块自带 Manager，它提供了一种看起来最贴近需求的方式：直接创建一个支持跨进程访问的 list 或 dict。用法非常简单：

import multiprocessing def worker(mdict, mlist): mdict["key"] = "from child" mlist.append("hello") if __name__ == "__main__": with multiprocessing.Manager() as manager: shared_dict = manager.dict() shared_list = manager.list() processes = [ multiprocessing.Process(target=worker, args=(shared_dict, shared_list)) for _ in range(4) ] for p in processes: p.start() for p in processes: p.join() print(shared_dict) # {'key': 'from child'} print(list(shared_list)) # ['hello', 'hello', 'hello', 'hello']

这确实能解决"子进程写入、主进程可见"的问题。manager.dict() 返回的是一个代理对象，代理背后的真实字典由 Manager 服务进程持有。子进程往代理对象写入数据时，数据会先被序列化，通过管道或 Socket 发送到 Manager 服务进程，由服务进程执行真正的修改。代理对象内部还绑定了锁，多个进程同时写入同一个 key 时，基本不会出现内存层面的脏数据问题。

2.2 代理对象背后的 IPC 链路，慢是必然的

Manager 的易用性很诱人，代价是性能。每次 mdict["key"] = value 的赋值操作，至少要走一条完整链路：调用方进程把 key 和 value 序列化、通过 Socket 发给 Manager 服务进程、服务进程反序列化后去改内部字典、再把确认结果回传。如果是读操作，还需要把目标对象从服务进程物理拷贝回来，这意味着每次 get 都可能拿到一个大型对象的新副本。

我在一台普通 Linux 服务器上实测过，对一个 manager.dict() 做一次 set，单次开销大约在几十微秒到一毫秒之间，具体数值受网络缓冲和序列化体积影响很大。如果只有一个进程写，倒也还好；一旦 8 个进程同时高频写同一个字典，锁竞争会让等待时间成倍放大。业务里如果总共只需要写几千次，这个开销完全可接受；但如果要写几十万次甚至上百万次，比如统计热词频率、累加计数、维护大表状态，Manager 就有很大概率成为整个程序的瓶颈，甚至比单进程还慢。不要把"能写"等同于"适合写"，这个区分在并发编程里特别重要。

2.3 Manager 的三大坑：嵌套容器、生命周期与跨平台

第一个坑是嵌套容器失效。如果初始化一个 manager.list()，往里 append 一个普通 Python list，然后在子进程里修改这个内层 list：

def worker(mlist): inner = mlist[0] inner.append("x") # 不会同步到共享容器

你会看到外部共享 list 明明包含了 inner，但 inner 的变化完全不可见。原因很简单：只有外层 list 是代理对象，它对内层普通 list 的修改路径完全没有拦截能力。正确做法是内层也用 manager.list() 创建，让整个嵌套结构都是代理对象。

第二个坑是生命周期。with multiprocessing.Manager() as manager 退出之后，Manager 服务进程会被关闭，所有从它那里拿到的代理对象都会失效。如果把代理对象存成全局变量、模块级变量，代码走到离线阶段再使用，就会报各种连接相关错误。Windows 平台上如果 Manager 没有被显式关闭，还会留下孤儿服务进程，反复运行几次之后任务管理器里会多出一堆残留进程。

第三个坑是入口保护。在 Windows 的 spawn 模式下，主模块没有 ifname== "main" 保护，创建 Manager 或 Process 时会递归导入主模块，轻则重复执行，重则直接抛出 RuntimeError。这行守卫代码看起来繁琐，但它是多进程程序能跨平台跑起来的前提条件。

3. 方案二：收集式写入模型，大数据量场景下的正确姿势

3.1 核心思路：各算各的，最后再合并

回到问题本身，多数场景想要的并不是"多个进程实时共享一个可变容器"，而是"多个进程算出的结果最后能汇总到一个 list 或 dict 里"。既然跨进程的每次写入都这么贵，那就尽量减少跨进程交互，把数据的产生和数据的汇总拆开：每个子进程在本地内存中维护自己的结果集合，任务完成后一次性把结果传回主进程，由主进程完成最终合并。这种模型在 MapReduce 里叫 map 和 reduce，在 Python 并发编程里就是返回结果加统一汇总。

这种模型还有一个隐性优势：子进程之间完全不存在状态纠缠，某个子进程计算中途崩溃，顶多丢失它自己那份任务，不会污染其他进程的结果。相比之下，Manager 模型里一个子进程写入了一半挂了，共享容器里可能留下半截数据，后续排查会很痛苦。

3.2 ProcessPoolExecutor + as_completed 实战代码

用 concurrent.futures 实现收集式写入非常直观：

from concurrent.futures import ProcessPoolExecutor, as_completed def process_item(x): # 模拟一个耗时计算 return x, x * x data = list(range(1, 1001)) if __name__ == "__main__": final_dict = {} with ProcessPoolExecutor(max_workers=8) as pool: futures = [pool.submit(process_item, item) for item in data] for future in as_completed(futures): k, v = future.result() final_dict[k] = v print(len(final_dict)) # 1000

这里子进程计算完成后，只把 (key, value) 作为返回值传回主进程，主进程在 future.result() 时接收数据并写入自己的 dict。整个生命周期里，每个任务只在完成时发生一次跨进程数据传输，而 Manager 模型是每次 set、每次 get 都发生一次。数据量越大，这个差距越明显。我自己试过 20 万条数据的场景，Manager 模型能跑到让人怀疑人生，同一个任务换成收集式，几秒钟搞定。

如果担心主进程逐个合并太慢，可以考虑让每个子进程直接返回一个局部容器，比如一个 Counter 或小 dict，主进程再把这些局部容器整体相加。合并本身就是内存操作，速度远快于跨进程通信。

3.3 什么时候改用 multiprocessing.Queue

ProcessPoolExecutor 的模型有一个前提：所有任务在提交时已经确定，而且任务之间没有依赖关系。如果你面对的是流水线式场景，比如一个生产者持续扫描一批 URL，解析出新的任务后又继续放给消费者，那么队列模型更合适。multiprocessing.Queue 可以一端放任务、一端收结果：

import multiprocessing def worker(task_q, result_q): for data in iter(task_q.get, None): # None 作为结束标志 result_q.put((data, data * 2)) if __name__ == "__main__": task_q = multiprocessing.Queue() result_q = multiprocessing.Queue() processes = [ multiprocessing.Process(target=worker, args=(task_q, result_q)) for _ in range(4) ] for p in processes: p.start() for i in range(100): task_q.put(i) for _ in range(4): task_q.put(None) for p in processes: p.join() while not result_q.empty(): k, v = result_q.get() print(k, v)

Queue 内部通过管道和锁配合实现线程安全，多个消费者进程可以安全地竞争任务。这里的注意点是：结束标志必须有，否则消费者永远在等待；结果队列最好在确认所有消费者退出后再统一收割，避免读到半截数据。

3.4 不要同时混用 Pool 与 Queue

我在项目里踩过一个很隐蔽的坑：在 ProcessPoolExecutor 的 worker 函数里用了 multiprocessing.Queue，结果程序总是莫名其妙卡住，偶尔还会抛出 BrokenPipe。原因是 Executor 本身用自己的方式管理工作进程和任务分发，再嵌套一层 Queue，相当于在一个进程模型里又叠了另一套进程模型，两边的线程和管道相互纠缠，行为非常不可预测。我的建议是二选一：已经选了进程池模型，就用返回结果合并；已经选了队列模型，就直接用 multiprocessing.Process 管理，别把 Queue 传给 Pool 的 worker。

另外，apply_async 的 callback 参数值得单独提一下。callback 回调是在主进程里执行的，所以它内部可以直接操作普通的 list 和 dict：

import multiprocessing result = [] def collect(kv): result.append(kv) def work(v): return (v, v * 2) if __name__ == "__main__": pool = multiprocessing.Pool(4) for i in range(100): pool.apply_async(work, (i,), callback=collect) pool.close() pool.join() print(len(result))

这种写法让代码看起来像"子进程直接往同一个 list 写"，实际是主进程在回调里完成了写入，安全性有保障。不过有一点要记住，callback 的调用顺序和任务提交顺序不一定一致，对顺序敏感的场景需要额外做排序处理。

4. 方案三：共享内存与外部存储的兜底路径

4.1 Value / Array 只能覆盖简单类型边界

如果共享的数据本身不是 list/dict 这种复杂结构，只是计数器、标志位、总量这种标量，可以用 multiprocessing.Value 和 Array，它们直接基于操作系统共享内存，速度比 Manager 快一个数量级以上：

import multiprocessing counter = multiprocessing.Value("i", 0) def inc(): with counter.get_lock(): counter.value += 1

Array 支持一维 C 类型数组，可以存放固定长度的数值或字节序列。但问题在于，Value/Array 并不直接支持 list/dict 的动态结构。有人曾尝试把 dict 序列化成 JSON 字节串，再塞进 Array("c")，每次读写都要序列化、加锁、拷贝回普通对象，这套组合的代码量和排查成本远超直接用 Manager，而且字典结构一变，缓冲区长度又得重新设计。如果需求仅仅是共享一个复杂字典的最终快照，这个方案并不值得。

4.2 用 SQLite 让多个进程共享一份结构化数据

当数据规模大到几十万甚至上千万条，而且多个进程需要共同更新一张结构明确的数据表时，SQLite 是一个很务实的选择。每个子进程独立打开同一个数据库文件，通过 INSERT/UPDATE 写入，由 SQLite 的锁机制保证并发一致性，主进程最后统一查询或聚合。只要控制好事务提交节奏，它的稳定性比 Manager 高很多：

import sqlite3 import multiprocessing DB_PATH = "shared_state.db" def worker(n): conn = sqlite3.connect(DB_PATH, timeout=10) conn.execute("PRAGMA journal_mode=WAL") conn.execute( "INSERT OR REPLACE INTO kv(key, value) VALUES (?, ?)", (n, n * 2) ) conn.commit() conn.close() if __name__ == "__main__": conn = sqlite3.connect(DB_PATH) conn.execute( "CREATE TABLE IF NOT EXISTS kv(key INTEGER PRIMARY KEY, value TEXT)" ) conn.commit() conn.close() processes = [ multiprocessing.Process(target=worker, args=(i,)) for i in range(1000) ] for p in processes: p.start() for p in processes: p.join() conn = sqlite3.connect(DB_PATH) print(conn.execute("SELECT COUNT(*) FROM kv").fetchone())

这里有两个实践要点。第一，每个进程只建立一次连接，用完关闭，不要在循环里反复 connect，那会带来大量不必要的文件锁竞争。第二，开启 WAL 模式可以显著降低读与写之间的阻塞，多个写者并发时由数据库层排队，避免应用层自己造锁。SQLite 本质上是文件型数据库，单机上百万级数据的并发写入和聚合完全够用，而且比 Manager 快得多。

4.3 需要外部组件时的决策点

如果数据量大到单机内存放不下、还要多机共享时，可以考虑引入外部组件，比如把 dict 存进内存数据库。但每次引入新组件之前，都应该先判断：这个共享需求是不是设计得太重了？我见过不少项目，最终把所有状态塞进外部内存数据库，结果性能瓶颈全在序列化和网络往返上，代码复杂度也直线上升。面对多进程写同一个 list/dict，优先考虑让进程间通信次数尽量减少，而不是找个地方存放共享对象。

5. 方案选型参考：四个维度快速做决策

5.1 从数据规模、写频率、实时性、可维护性判断

真正做技术选型时，可以问自己四个问题。

数据规模有多大？几百条还是几百万条？规模小，用什么都能跑起来；规模大，Manager 的 IPC 开销就会变成灾难。

写操作频率有多高？每秒几十次和每秒几万次，是两个完全不同的世界。高频同步状态，必须先考虑收敛写入次数。

是否真的需要实时看到其他进程写入的内容？很多应用只是收集最终结果，中间过程完全不需要共享可见性。既然不要求实时，收集式模型就是天然的最优解。

代码后续由谁维护？如果团队里有并发经验不深的人参与，Manager 那套代理模型写起来虽然简单，出了问题却极难排查，大量隐式的序列化和锁竞争会把问题藏得很深。

5.2 一张选型表

选型表的中心思想是：跨进程通信越少越好。Manager 看起来很省事，但它的每一次操作都是完整一次跨进程调用；收集式模型把大量通信压缩到一次返回值传输，这在任何规模下都是更健康的架构。

6. 实战复盘：10 万条分析任务，从 10 分钟提速到 5 秒

6.1 需求与第一版实现

当时需要处理一批 10 万条数据，每条数据经过清洗后要按城市维度统计数量，并累加每个城市对应的服务时长，最终得到一个 dict，键是城市名，值是 {count: n, duration: m}。因为单机 CPU 闲置资源多，很自然地想到用多进程并行。

第一版我使用了 Manager：

def process_item(item, city_counter, city_duration): city = extract_city(item) city_counter[city] = city_counter.get(city, 0) + 1 city_duration[city] = city_duration.get(city, 0) + get_duration(item)

代码非常干净，子进程直接更新两个共享管理器字典。跑起来之后，状态完全正确，但速度让人崩溃。8 个进程下，每百条数据的处理耗时会不停波动，整体跑完用了超过 10 分钟，而且任务管理器和单核 CPU 都显示进程根本没有被喂满。瓶颈不在计算，而在于每次自增都要跑到 Manager 服务进程去执行读改写，多进程写同一个 key 还会触发锁竞争，时间被成倍拖长。

6.2 重构为收集式分片合并

第二版完全放弃了共享字典。把 10 万条数据切成 8 份，每个子进程处理完自己的分片后，返回一个局部 Counter 和一个局部 duration dict：

from collections import Counter, defaultdict def process_batch(items): city_counter = Counter() city_duration = defaultdict(float) for item in items: city = extract_city(item) city_counter[city] += 1 city_duration[city] += get_duration(item) return city_counter, dict(city_duration) if __name__ == "__main__": # 分批调度，每个 task 返回局部结果 ... # 主进程统一合并 total_counter = Counter() total_duration = defaultdict(float) for counter, duration in batch_results: total_counter.update(counter) for k, v in duration.items(): total_duration[k] += v

合并操作本身是纯内存计算，几万项数据的字典合并连一秒钟都用不到。最终版本整体耗时约 5 秒，其中大头是业务计算，跨进程通信几乎可以忽略。和第一版对比：

这个复盘很有参考价值：决定多进程性能的，往往不是进程创建开销，而是跨进程数据交互频率。能把 N 次小通信合并成 1 次大通信，收益是数量级的。

7. 常见问题速查与排错心得

7.1 高频异常与对应处理

7.2 一套百试百灵的调试顺序

遇到多进程写完数据不生效的问题，我建议按这个顺序排查。首先，在主进程用一个非常小的数据量跑单进程版本，确认业务逻辑正确；第二步，改成多进程但不引入共享容器，让每个子进程通过 return 返回结果，用 future.result() 验证每个任务确实完成；第三步，如果确定必须要共享容器，再引入 Manager，并把随机的写入次数降到可观测量级。这套流程能把问题快速拆分成"业务逻辑错误"和"并发模型错误"，省去大量盲猜的时间。

打个比方，跨进程共享容器就像几个人同时往同一个账本上写字，每个人写字落笔的瞬间，墨水和纸张都要经过一段长长的传送带；而收集式模型是每个人在自己的小账本上先记完，最后统一交到一个人手里汇总。传送带快不起来，但记账这事原本可以不用实时共享。理解了这个差异，很多并发设计里的纠结就会迎刃而解。