别让你的变量“迷路”：Python进程间通信的3种正确姿势

免费编程软件「python+pycharm」

链接：https://pan.quark.cn/s/48a86be2fdc0

一个让我抓狂的数据处理任务

上个月我接了个活，处理200万条用户行为日志。每条日志都要做格式清洗、字段提取、类型转换。单线程跑完要13秒，但服务器是8核的，我心想：用多进程总该起飞了吧？

代码很快写好了：

import multiprocessing as mp

CONFIG = {"date_format": "%Y-%m-%d", "max_len": 100}
results = []  # 用来收集所有子进程的结果

def process_line(line):
   # 用CONFIG里的配置处理数据
   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！居然是0！


我盯着屏幕上的“0”，懵了整整五分钟。

主进程里的results明明是全局列表，子进程往里面加了数据，为什么主进程拿不到？

原因很简单：进程之间是隔阂的。每个子进程有自己独立的内存空间，它修改的是自己那一份results的拷贝，主进程的results从来没变过。

今天我就把Python多进程里“让变量不迷路”的3种正确姿势讲清楚。

为什么子进程总是“拿不到”数据？

在讲解决方案之前，先花一分钟搞清楚问题根源。

当你用multiprocessing.Process或Pool创建子进程时，Python会：

1. 启动一个新的Python解释器（相当于重新打开了一个Python程序）
2. 在新的解释器里重新执行你的代码（包括导入模块、定义函数、执行全局代码）
3. 子进程拥有独立的内存空间，和主进程完全隔开

操作系统不允许一个进程随便读写另一个进程的内存。所以你在主进程里定义的CONFIG、results、global_counter，子进程都看不见。即使子进程“看起来”访问了同名变量，它操作的也是一份独立的拷贝。这就是“变量迷路”的根本原因。

搞清楚了“为什么”，接下来看“怎么办”。Python提供了3种主流方案，让数据在进程之间安全地流通。

姿势一：Queue——像传送带一样传递数据

multiprocessing.Queue是最常用的进程间通信方式。它像一个传送带，一个进程把数据放上去，另一个进程从另一端取下来。

适用场景：生产者-消费者模式、任务分发、结果收集。

为什么用它：Queue内部处理了所有同步问题，多个进程可以安全地往里面放数据和取数据。它是线程安全和进程安全的。

怎么写：

from multiprocessing import Process, Queue

def worker(q):
   # 从队列取数据，直到收到结束信号
   while True:
       item = q.get()
       if item is None:  # 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())


关键点：Queue的put()和get()都是阻塞操作，如果队列满了，put()会等待；如果队列空了，get()会等待。你可以用q.put(None)作为“任务结束”的信号，让子进程优雅退出。

什么时候用：任务数量不确定、需要动态分发工作、多个生产者和多个消费者。

姿势二：Manager——让多个进程共享同一个“黑板”

如果你需要多个进程共享同一个列表、字典或其他复杂数据结构，Manager是更直接的选择。

适用场景：多个进程需要读写同一个数据容器（比如共享任务列表、共享配置字典）。

为什么用它：Manager会启动一个独立的“管理器进程”，所有对共享对象的操作都通过这个进程来协调。你操作共享列表就像操作普通列表一样，但背后会自动进行进程间通信和同步。

怎么写：

from multiprocessing import Process, Manager

def worker(shared_list, shared_dict, value):
   shared_list.append(value)
   shared_dict[value] = 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}


关键点：Manager支持的类型很丰富——list、dict、Namespace、Lock、Queue等。但需要注意，Manager的每次读写操作都有进程间通信的开销，性能比直接访问内存慢。如果你只需要共享简单的数值，用Value或Array效率更高。

什么时候用：需要共享复杂数据结构、代码可读性比极致性能更重要、数据量不大。

姿势三：SharedMemory——直达内存的“高速公路”

如果你的数据量很大，或者对性能有要求，SharedMemory是最优解。它让多个进程直接读写同一块物理内存，不需要序列化和反序列化，速度最快。

适用场景：大量数据共享、高性能计算、NumPy数组跨进程共享。

为什么用它：相比Queue和Manager，SharedMemory避开了序列化/反序列化和复制数据的开销，直接在内存层面共享数据，性能优势极其明显。

怎么写：

from multiprocessing import shared_memory
import numpy as np

# 主进程：创建共享内存
shm = shared_memory.SharedMemory(create=True, size=1024)
# 用numpy访问这块内存
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配合with语句，自动管理共享内存的生命周期，避免忘记unlink导致内存泄漏：

from multiprocessing.managers import SharedMemoryManager

with SharedMemoryManager() as smm:
   sl = smm.ShareableList(range(2000))
   # 多个进程可以共享sl
   # 退出with块时，所有共享内存自动释放


如果你用NumPy做科学计算，还可以用第三方库shared-ndarray2进一步简化操作。

什么时候用：数据量大（MB级以上）、对性能敏感、需要频繁读写共享数据。

一张图帮你选

一句话决策：

• 只是传递任务和结果 → 用Queue
• 需要共享列表/字典，数据量不大 → 用Manager
• 共享大数组、追求性能 → 用SharedMemory

回到开头的那个Bug

我那个数据处理任务最后用了Queue。主进程把200万行数据分批放进Queue，4个子进程轮流取数据、处理、把结果放回另一个Queue。主进程再从结果Queue里收集所有清洗后的数据。

改了之后，耗时从13秒降到了3.6秒，而且再也不会出现“results是0”的尴尬了。

记住：多进程之间是隔阂的，变量不会自动共享。想让数据不迷路，就得用对姿势——Queue传递消息，Manager共享容器，SharedMemory直达内存。选对方案，你的多进程程序才能真正跑起来。