Python进程间通信三种方法Queue Manager SharedM

脚本专家

笔者上个月处理200万条用户行为日志，单线程清洗要13秒。心想8核服务器用多进程应该飞快，于是写了如下代码：

2. import multiprocessing as mp
4. CONFIG = {"date_format": "%Y-%m-%d", "max_len": 100}
5. results = []
7. def process_line(line):
8.     cleaned = do_clean(line, CONFIG)
9.     results.append(cleaned)
10.     return cleaned
12. if __name__ == "__main__":
13.     with mp.Pool(4) as pool:
14.         pool.map(process_line, data)
15.     print(len(results))  # 输出0！

复制代码

运行后主进程的results依然是空列表。原因很简单：每个子进程拥有独立的内存空间，主进程定义的results在子进程中是拷贝，修改不影响主进程。Python多进程间通信必须显式传递数据，下面介绍三种正确姿势。

---

## 一、Queue：传送带式通信

multiprocessing.Queue内部处理了同步问题，多个进程可以安全地put和get。适用于生产者-消费者模式、任务分发和结果收集。

### 经典用法

2. from multiprocessing import Process, Queue
4. def worker(q):
5.     while True:
6.         item = q.get()
7.         if item is None:  # 结束信号
8.             break
9.         result = item * 2
10.         q.put(result)
12. if __name__ == "__main__":
13.     q = Queue()
14.     p = Process(target=worker, args=(q,))
15.     p.start()
16.     for i in range(10):
17.         q.put(i)
18.     q.put(None)  # 通知子进程结束
19.     p.join()
20.     while not q.empty():
21.         print(q.get())

复制代码

关键点：put()和get()默认阻塞，队列满时put等待，队列空时get等待。常用None作为结束信号。适合任务数量不确定、动态分发的场景。性能中等，需要序列化。

---

## 二、Manager：共享复杂数据结构

Manager启动一个独立的管理器进程，通过代理对象实现共享。操作共享列表、字典就像操作普通对象，但每次读写都有进程间通信开销。

### 示例

2. from multiprocessing import Process, Manager
4. def worker(shared_list, shared_dict, value):
5.     shared_list.append(value)
6.     shared_dict[value] = value * 2
8. if __name__ == "__main__":
9.     with Manager() as manager:
10.         shared_list = manager.list(["初始数据"])
11.         shared_dict = manager.dict()
12.         processes = []
13.         for i in range(5):
14.             p = Process(target=worker, args=(shared_list, shared_dict, i))
15.             processes.append(p)
16.             p.start()
17.         for p in processes:
18.             p.join()
19.         print(list(shared_list))  # ['初始数据', 0, 1, 2, 3, 4]
20.         print(dict(shared_dict))  # {0: 0, 1: 2, 2: 4, 3: 6, 4: 8}

复制代码

支持list、dict、Namespace、Lock、Queue等。数据量不大、需要共享复杂容器时使用。注意：每次操作都有IPC开销，性能比直接内存访问慢。

---

## 三、SharedMemory：零拷贝内存共享

SharedMemory让多个进程直接读写同一块物理内存，无需序列化，速度最快。Python 3.8+内置。适合大数据量、高性能计算，尤其是NumPy数组共享。

### 基础用法

2. from multiprocessing import shared_memory
3. import numpy as np
5. # 主进程创建共享内存
6. shm = shared_memory.SharedMemory(create=True, size=1024)
7. arr = np.ndarray((256,), dtype=np.int32, buffer=shm.buf)
8. arr[:] = np.arange(256)
10. # 子进程通过shm.name连接同一块内存
11. # shm_b = shared_memory.SharedMemory(name=shm.name)
12. # arr_b = np.ndarray((256,), dtype=np.int32, buffer=shm_b.buf)
13. # 修改arr_b即修改arr
15. shm.close()
16. shm.unlink()  # 必须释放

复制代码

### 使用SharedMemoryManager自动管理生命周期

2. from multiprocessing.managers import SharedMemoryManager
4. with SharedMemoryManager() as smm:
5.     sl = smm.ShareableList(range(2000))
6.     # 多个进程共享sl
7.     # 退出with自动释放所有共享内存

复制代码

性能极高（零拷贝），但需要开发者管理内存分配和释放。适合MB级以上的数据频繁读写。

---

## 决策原则

| 通信方式 | 适用场景 | 性能 | 易用性 |
|---------|---------|------|-------|
| Queue | 任务分发、结果收集 | 中等（需序列化） | ★★★★★ |
| Manager | 共享列表/字典等复杂结构 | 较慢（协调开销） | ★★★★ |
| SharedMemory | 大数组、高性能需求 | 极快（零拷贝） | ★★★（需手动管理） |

一句话选择：只传递任务和结果选Queue；需要共享容器、数据量不大选Manager；共享大数组、追求极致性能选SharedMemory。

回到开头的数据处理任务，最终改用Queue方案：主进程将200万条日志分批放入Queue，4个子进程轮流取数据处理，结果放入另一个Queue，主进程收集。耗时从13秒降到3.6秒，再也不会出现results为0的尴尬。

多进程内存隔离是Python并发编程的常见陷阱。掌握Queue、Manager、SharedMemory三种通信方式，就能让数据在进程间安全、高效地流通。

热心网友4

很好的经验分享！那个results为空的例子太真实了，新手很容易栽在这上面。表格总结得非常清晰，我平时多用Manager共享字典来存配置，没想到SharedMemory性能这么高，以后处理大日志文件可以试试这个零拷贝方案。感谢楼主整理！

热心网友4

楼主这帖子太实用了！之前用多进程也踩过 `results` 空列表的坑，看完才知道是子进程独立内存空间的问题。三种方法对比得很清楚，Queue 适合动态任务分发，Manager 操作简单但开销大，SharedMemory 零拷贝高性能，决策表也一目了然。想问下楼主实际处理那200万条数据时，最终选了哪种方案？分别试过对比性能吗？

热心网友4

感谢楼主的详细总结！之前我也踩过 `results` 在子进程里修改主进程看不到的坑，看完这篇终于理清楚了。 想请教一下：在实际使用 `SharedMemory` 时，如果多个子进程同时读写同一块区域，是否需要额外加锁？文档里提到 `ShareableList` 本身不是线程安全的，但多进程场景下怎么保证数据一致性呢？另外，楼主提到 `Manager` 性能较慢，那在数据量中等（比如几十万条字符串）时，用 `Queue` 收集结果和用 `Manager.list` 收集，实际差距大概有多大？