Python进程间通信三种方法Queue Manager SharedM
笔者上个月处理200万条用户行为日志，单线程清洗要13秒。心想8核服务器用多进程应该飞快，于是写了如下代码：import multiprocessing as mp
CONFIG = {"date_format": "%Y-%m-%d", "max_len": 100}
results = []
def process_line(line):
cleaned = do_clean(line, CONFIG)
results.append(cleaned)
return cleaned
if __name__ == "__main__":
with mp.Pool(4) as pool:
pool.map(process_line, data)
print(len(results))# 输出0！
运行后主进程的results依然是空列表。原因很简单：每个子进程拥有独立的内存空间，主进程定义的results在子进程中是拷贝，修改不影响主进程。Python多进程间通信必须显式传递数据，下面介绍三种正确姿势。
---
## 一、Queue：传送带式通信
multiprocessing.Queue内部处理了同步问题，多个进程可以安全地put和get。适用于生产者-消费者模式、任务分发和结果收集。
### 经典用法
from multiprocessing import Process, Queue
def worker(q):
while True:
item = q.get()
if item is None:# 结束信号
break
result = item * 2
q.put(result)
if __name__ == "__main__":
q = Queue()
p = Process(target=worker, args=(q,))
p.start()
for i in range(10):
q.put(i)
q.put(None)# 通知子进程结束
p.join()
while not q.empty():
print(q.get())
关键点：put()和get()默认阻塞，队列满时put等待，队列空时get等待。常用None作为结束信号。适合任务数量不确定、动态分发的场景。性能中等，需要序列化。
---
## 二、Manager：共享复杂数据结构
Manager启动一个独立的管理器进程，通过代理对象实现共享。操作共享列表、字典就像操作普通对象，但每次读写都有进程间通信开销。
### 示例
from multiprocessing import Process, Manager
def worker(shared_list, shared_dict, value):
shared_list.append(value)
shared_dict = value * 2
if __name__ == "__main__":
with Manager() as manager:
shared_list = manager.list(["初始数据"])
shared_dict = manager.dict()
processes = []
for i in range(5):
p = Process(target=worker, args=(shared_list, shared_dict, i))
processes.append(p)
p.start()
for p in processes:
p.join()
print(list(shared_list))# ['初始数据', 0, 1, 2, 3, 4]
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数组共享。
### 基础用法
from multiprocessing import shared_memory
import numpy as np
# 主进程创建共享内存
shm = shared_memory.SharedMemory(create=True, size=1024)
arr = np.ndarray((256,), dtype=np.int32, buffer=shm.buf)
arr[:] = np.arange(256)
# 子进程通过shm.name连接同一块内存
# shm_b = shared_memory.SharedMemory(name=shm.name)
# arr_b = np.ndarray((256,), dtype=np.int32, buffer=shm_b.buf)
# 修改arr_b即修改arr
shm.close()
shm.unlink()# 必须释放
### 使用SharedMemoryManager自动管理生命周期
from multiprocessing.managers import SharedMemoryManager
with SharedMemoryManager() as smm:
sl = smm.ShareableList(range(2000))
# 多个进程共享sl
# 退出with自动释放所有共享内存
性能极高（零拷贝），但需要开发者管理内存分配和释放。适合MB级以上的数据频繁读写。
---
## 决策原则
| 通信方式 | 适用场景 | 性能 | 易用性 |
|---------|---------|------|-------|
| Queue | 任务分发、结果收集 | 中等（需序列化） | ★★★★★ |
| Manager | 共享列表/字典等复杂结构 | 较慢（协调开销） | ★★★★ |
| SharedMemory | 大数组、高性能需求 | 极快（零拷贝） | ★★★（需手动管理） |
一句话选择：只传递任务和结果选Queue；需要共享容器、数据量不大选Manager；共享大数组、追求极致性能选SharedMemory。
回到开头的数据处理任务，最终改用Queue方案：主进程将200万条日志分批放入Queue，4个子进程轮流取数据处理，结果放入另一个Queue，主进程收集。耗时从13秒降到3.6秒，再也不会出现results为0的尴尬。
多进程内存隔离是Python并发编程的常见陷阱。掌握Queue、Manager、SharedMemory三种通信方式，就能让数据在进程间安全、高效地流通。
Re: Python进程间通信三种方法Queue Manager SharedM
很好的经验分享！那个results为空的例子太真实了，新手很容易栽在这上面。表格总结得非常清晰，我平时多用Manager共享字典来存配置，没想到SharedMemory性能这么高，以后处理大日志文件可以试试这个零拷贝方案。感谢楼主整理！Re: Python进程间通信三种方法Queue Manager SharedM
楼主这帖子太实用了！之前用多进程也踩过 `results` 空列表的坑，看完才知道是子进程独立内存空间的问题。三种方法对比得很清楚，Queue 适合动态任务分发，Manager 操作简单但开销大，SharedMemory 零拷贝高性能，决策表也一目了然。想问下楼主实际处理那200万条数据时，最终选了哪种方案？分别试过对比性能吗？Re: Python进程间通信三种方法Queue Manager SharedM
感谢楼主的详细总结！之前我也踩过 `results` 在子进程里修改主进程看不到的坑，看完这篇终于理清楚了。 想请教一下：在实际使用 `SharedMemory` 时，如果多个子进程同时读写同一块区域，是否需要额外加锁？文档里提到 `ShareableList` 本身不是线程安全的，但多进程场景下怎么保证数据一致性呢？另外，楼主提到 `Manager` 性能较慢，那在数据量中等（比如几十万条字符串）时，用 `Queue` 收集结果和用 `Manager.list` 收集，实际差距大概有多大？
页:
[1]