Python多进程通信是并发编程中提升程序性能的重要手段，但当需要传递的数据量达到GB级别时，传统的管道、普通队列等通信方式会因为频繁的数据拷贝导致内存暴涨、传输延迟过高，甚至引发程序崩溃。针对大容量数据的多进程通信场景，需要采用更高效的策略来减少数据拷贝、降低内存开销。
大容量数据通信的核心痛点
常规的多进程通信方式比如multiprocessing.Queue，在传递数据时会对数据进行序列化后拷贝到进程间的共享缓冲区，接收进程再反序列化读取。如果数据量达到几百MB甚至几GB，序列化、反序列化以及两次数据拷贝的开销会非常可观，同时会占用大量内存，这也是大容量数据通信需要解决的核心问题。
常用处理策略
1. 共享内存（Shared Memory）
共享内存是多个进程直接访问同一块物理内存区域，不需要进行数据拷贝，是处理大容量数据最高效的方式之一。Python3.8+的multiprocessing.shared_memory模块提供了便捷的共享内存操作接口，适合存储数组、矩阵等结构化大容量数据。
适用场景：多个进程需要读写同一份大容量结构化数据，且数据格式固定。
import numpy as np
from multiprocessing import Process
from multiprocessing.shared_memory import SharedMemory
def read_shm(shm_name, shape):
# 连接已有的共享内存
shm = SharedMemory(name=shm_name)
# 将共享内存转换为numpy数组
arr = np.ndarray(shape, dtype=np.float64, buffer=shm.buf)
print(f"子进程读取到的数据第一个元素: {arr[0]}")
shm.close()
if __name__ == "__main__":
# 创建原始大容量数据
data = np.random.rand(1000000).astype(np.float64)
shape = data.shape
# 创建共享内存，大小为数据字节数
shm = SharedMemory(create=True, size=data.nbytes)
# 将数据拷贝到共享内存
shm_arr = np.ndarray(shape, dtype=np.float64, buffer=shm.buf)
shm_arr[:] = data[:]
print(f"主进程写入数据第一个元素: {shm_arr[0]}")
# 启动子进程，传递共享内存名称和形状
p = Process(target=read_shm, args=(shm.name, shape))
p.start()
p.join()
# 清理共享内存
shm.close()
shm.unlink()
2. 内存映射文件（Memory Mapped File）
内存映射文件是将磁盘文件直接映射到进程的虚拟内存空间，进程可以像操作内存一样操作文件内容，多个进程映射同一个文件就能实现数据共享，同时数据会按需加载到内存，不会一次性占用全部内存，适合超大大容量数据的场景。
适用场景：数据量超过物理内存大小，或者需要持久化存储的大容量数据。
import mmap
import os
from multiprocessing import Process
def read_mmap(file_path, size):
# 打开文件
with open(file_path, "r+b") as f:
# 创建内存映射
mm = mmap.mmap(f.fileno(), length=size, access=mmap.ACCESS_READ)
# 读取前10个字节
content = mm.read(10)
print(f"子进程读取到的内容: {content.decode('utf-8')}")
mm.close()
if __name__ == "__main__":
file_path = "large_data.txt"
# 创建大容量测试文件，写入100MB数据
with open(file_path, "w") as f:
f.write("test_data_" * 12500000)
file_size = os.path.getsize(file_path)
# 启动子进程读取映射文件
p = Process(target=read_mmap, args=(file_path, file_size))
p.start()
p.join()
# 清理测试文件
os.remove(file_path)
3. 优化后的队列通信
如果必须使用队列传递大容量数据，可以通过减少序列化开销、分块传输的方式来优化。比如使用pickle的更高版本协议，或者将数据拆分为多个小块分批传输，避免单次传输数据过大导致内存溢出。
适用场景：数据需要按顺序传递，且无法使用共享内存或内存映射的场景。
from multiprocessing import Process, Queue
import pickle
def send_large_data(queue, data_chunks):
for chunk in data_chunks:
# 使用pickle协议4，支持更大的数据对象
serialized = pickle.dumps(chunk, protocol=4)
queue.put(serialized)
queue.put(None) # 结束标记
def receive_large_data(queue):
result = []
while True:
chunk = queue.get()
if chunk is None:
break
result.append(pickle.loads(chunk))
print(f"子进程接收完所有数据块，总块数: {len(result)}")
if __name__ == "__main__":
# 生成大容量数据并分块
large_data = ["chunk_" + str(i) for i in range(1000)]
chunk_size = 100
data_chunks = [large_data[i:i+chunk_size] for i in range(0, len(large_data), chunk_size)]
queue = Queue()
# 启动发送和接收进程
sender = Process(target=send_large_data, args=(queue, data_chunks))
receiver = Process(target=receive_large_data, args=(queue,))
sender.start()
receiver.start()
sender.join()
receiver.join()
策略选择参考
可以根据以下维度选择合适的通信策略：
| 策略类型 | 适用数据量 | 内存开销 | 传输效率 | 适用场景 |
|---|---|---|---|---|
| 共享内存 | MB到GB级 | 低 | 最高 | 结构化数据，多进程频繁读写 |
| 内存映射文件 | GB到TB级 | 按需加载 | 高 | 超大数据，需要持久化 |
| 优化队列 | MB级 | 中 | 中 | 顺序传递，无法使用共享内存 |
注意事项
- 使用共享内存时需要注意进程间的同步，避免多个进程同时写入导致数据错乱，可以配合
multiprocessing.Lock实现互斥访问。 - 内存映射文件需要保证所有访问进程对文件有对应的读写权限，且文件被删除后映射会失效。
- 分块传输队列数据时，需要设计好结束标记，避免接收进程无限等待。