一、先聊聊多进程通信那点事

在我们的日常开发里，用 Python 写多进程程序是很常见的事。可能是要做爬虫，可能是要批量处理图片，也可能是要搭建一个分布式的服务。你满心欢喜地把任务拆成好几个进程，却发现程序不仅没变快，反而像蜗牛一样慢。这时候，绝大多数的问题都出在“进程间通信”上。

1.1 为什么你的多进程程序越跑越慢

你想象一下，一个进程就像一个独立的房间，房间之间没有门，数据要共享就得通过一个小窗口递来递去。这个窗口就是“进程间通信”。在 Python 里，因为全局解释器锁的存在，多线程在 CPU 密集型任务上很难提速，所以很多人改用多进程。可多进程也有自己的代价——每个进程的内存是隔离的。你想把一个计算结果递给另一个进程，必须把数据“打包”起来，从窗口塞过去，对面再“拆包”。打包和拆包的过程，学名叫“序列化与反序列化”。

如果你的任务里频繁传输大量数据，比如一个大字典、一段长文本，甚至是一张图片的二进制内容，序列化和内存拷贝的时间就会远远超过计算时间。这就是性能瓶颈的主要来源。

1.2 进程间通信的基本玩法

Python 的多进程库 multiprocessing 提供了一些现成的通信工具，常用的有 Queue（队列）、Pipe（管道）和 Value/Array（共享内存）。此外还有 shared_memory、Manager 等。它们各有各的脾性：有的使用简单，但速度慢；有的速度快，但要注意同步；有的适合点对点，有的适合广播。

这里先提醒大家：multiprocessing.Queue 内部其实是“数据放入文件描述符 + 锁 + 缓冲区”的组合，它为了保证多线程安全，增加了不少额外操作。如果你只需要在父子进程之间传数据，Pipe 会更直接；如果你希望多个工人共享同一块数据，共享内存是更好的选择。理解了这些，优化方向自然就清晰了。

接下来我们就从一个常见的坑开始，一步步看看怎么优化。

二、最常见的坑：Queue 传输大数据

2.1 一个看似正常的示例

假设你是做数据处理的，有四个 worker 进程，主进程负责下发任务，worker 计算后返回结果。最顺手的方式就是用 Queue。下面这段代码看起来很正常：任务列表是 0 到 9999，每个 worker 拿到后乘个 2。

# -- coding: utf-8 -*-
# Python 技术栈：multiprocessing 的 Queue
from multiprocessing import Process, Queue
import time

def worker(q, result_q):
    while True:
        # 从队列里拿任务
        task = q.get()
        # 如果是终止信号，就退出循环
        if task is None:
            break
        # 假装做了一个耗时计算
        time.sleep(0.01)
        # 把结果放回结果队列
        result_q.put(task * 2)

if __name__ == "__main__":
    q = Queue()
    result_q = Queue()
    # 创建 4 个 worker 进程
    workers = [Process(target=worker, args=(q, result_q)) for _ in range(4)]
    for p in workers:
        p.start()
    # 生成 1 万个任务
    for i in range(10000):
        q.put(i)
    # 给每个 worker 发一个 None，表示没有新任务了
    for _ in range(4):
        q.put(None)
    # 等待并取回全部结果
    for _ in range(10000):
        result_q.get()
    # 等待所有 worker 退出
    for p in workers:
        p.join()
    print("All done")

这个例子很简单，任务是很小的数字，所以看起来还好。如果你把 task 换成一个几百 KB 的字典，瞬间你就知道什么叫“慢动作”。

2.2 问题出在哪

你运行后可能发现速度慢得离谱，尤其当任务变成大对象时。这里面的开销主要有三层：

第一层，序列化。用 Queue 传数据时，默认使用 pickle 把对象变成字节流。对象越复杂，序列化越耗时。第二层，内核缓冲拷贝。数据从发送进程的用户空间复制到内核空间，再从内核空间复制到接收进程的用户空间，中间还有加锁和信号同步。第三层，对象重建。接收方拿到字节流后要反序列化还原成 Python 对象。

如果你传的是一个 1MB 的字符串，序列化和拷贝加起来的时间大概能到毫秒级；如果每秒传几百次，CPU 小半都被耗在这儿了。

另外，Queue 还有一个容易被忽略的点：它内部还会用一个后台线程把放入队列的对象“搬运”到底层管道中。也就是说，你每 put 一个对象，先经过一次 pickle，放进一个缓冲区，后台线程再读取缓冲区并写入管道，接收端那边的线程再读取管道的字节流，经过 pickle 还原，最后放进接收队列。这一整套流水线，每一步都有锁和条件变量的参与。所以即便你传一个很小的数字，通信本身的成本也不低。

2.3 真实场景里的表现

我自己就遇到过类似的坑：当时做一个多进程日志分析服务，每个 worker 要接收一整块日志文件内容，少的几十 KB，多的几 MB。一开始用 Queue 传给子进程，结果发现日志一多，服务直接卡死。后来把 Queue 换成了 Pipe，速度提到了大约两倍。再后来把大块内容放到共享内存里，速度又翻了一倍。这就是优化带来的实际收益。

三、优化方案一：用更轻量的管道

3.1 Pipe 怎么用

Pipe 是点对点的管道通信，它比 Queue 轻量，没有复杂的缓存和锁机制。它适合在两个进程之间传递大量数据。使用方法很简单：用 Pipe() 创建一对连接，一个给父进程，一个给子进程，两边用 send 和 recv 传数据。

# -- coding: utf-8 -*-
# Python 技术栈：multiprocessing 的 Pipe
from multiprocessing import Process, Pipe

def child_process(conn):
    # 子进程这里先接收数据（阻塞等待）
    data = conn.recv()
    # 模拟处理一下：把字节内容转成大写
    processed = data.upper()
    # 把处理结果发回给父进程
    conn.send(processed)
    # 用完了关闭连接
    conn.close()

if __name__ == "__main__":
    # 创建双向管道
    parent_conn, child_conn = Pipe(duplex=True)
    # 启动子进程
    p = Process(target=child_process, args=(child_conn,))
    p.start()
    # 构造 1MB 的二进制数据（这里用字节串模拟）
    big_data = b"x" * 1024 * 1024
    # 从父进程这边发送给子进程
    parent_conn.send(big_data)
    # 等待子进程处理后的结果
    result = parent_conn.recv()
    print("收到结果长度：", len(result))
    # 等待子进程结束
    p.join()

这段代码里，父进程发 1MB 的字节串，子进程用 recv 一次性接收，这个过程比 Queue 减少了队列对象和锁的干涉，实际效果会快不少。不过要注意，这里传的仍然是一个对象，底层还是要经历序列化。

3.2 有什么要注意的

虽然 Pipe 快，但它本质上还是通过 socketpair 或匿名管道实现的，数据同样要序列化成字节流。如果数据太大，recv 会持续阻塞直到收到完整数据。此外，Pipe 只支持两个端点，如果你有四个 worker，就得建四条管道，不方便。还有，如果两个进程同时往管道里写，数据不会错乱，但如果同时读，就可能会抢，所以更适合一个进程写、一个进程读的场景。

四、优化方案二：共享内存，真正的“读写同一块地”

4.1 共享内存原理

共享内存可以让多个进程直接访问同一块物理内存，不需要数据拷贝，也不需要序列化。你可以想象成一栋楼里有一间活动室，几家公司的人都能进去用，不用每次把文件搬到楼外。

Python 标准库在 3.8 之后提供了 shared_memory 模块。它依赖于操作系统提供的共享内存机制，在 Linux 和 Windows 上都有实现。使用它的时候，有一个进程负责创建共享内存块，拿到一个名字，其他进程用这个名字打开同一块内存。数据就在那块内存里，谁想读谁读。

如果你用过 mmap，可能会觉得有点像。mmap 是把文件映射到内存，SharedMemory 则是直接分配一块系统共享内存。两者在底层有相似之处，但 SharedMemory 更贴合进程间共享数据的需求。

4.2 实战：多个进程直接读写同一块内存

这里我们用标准库里的 array 配合 shared_memory 来演示。为了安全，加一个锁，防止两个进程同时写。

# -- coding: utf-8 -*-
# Python 技术栈：multiprocessing 的 shared_memory
from multiprocessing import shared_memory, Process, Lock
import array

def read_shared(shm_name, size, lock):
    # 通过名字关联到同一块共享内存
    shm = shared_memory.SharedMemory(name=shm_name)
    try:
        with lock:
            # 从共享内存的字节区里按整型格式读出全部数据
            arr = array.array('i', shm.buf[:size * 4])
            # 打印前 5 个数字，验证数据是否正常
            print("读到前 5 个数字：", list(arr[:5]))
    finally:
        # 用完必须关闭，但不要 unlink（由创建方负责）
        shm.close()

if __name__ == "__main__":
    # 准备一列整数
    data = list(range(1000))
    # 创建共享内存，大小为 1000 个整数（每个 4 字节）
    shm = shared_memory.SharedMemory(create=True, size=len(data) * 4)
    # 把整数数组打包成字节，写入共享内存
    arr = array.array('i', data)
    shm.buf[:len(arr) * 4] = arr.tobytes()

    lock = Lock()
    # 子进程拿着共享内存的名字去访问
    p = Process(target=read_shared, args=(shm.name, len(data), lock))
    p.start()
    p.join()
    # 关闭并彻底释放共享内存
    shm.close()
    shm.unlink()
    print("共享内存使用完毕")

4.3 共享内存的正确姿势

共享内存虽然快，但有几个坑必须注意。第一，没有自动同步，你需要用锁来保护对共享区域的读写。第二，用完要释放，创建方要调用 unlink()，否则内存块会一直留在系统里，造成泄漏。第三，不能直接存 Python 可变对象，只能存原始字节流。如果你想存字典，可以用 json 序列化成字符串再写入，或者用 pickle，但那样又会引入序列化开销，所以共享内存比较适合存大块数据。

实际使用中，如果你有一块较大的数据要更新，建议锁定共享内存的某个区间，而不是锁整块区域。比如用多个 Lock 分别保护不同区域，能提高并发度。不过这个粒度需要根据实际场景调整，锁太多反而增加开销。

五、优化方案三：换一种思路——减少传输次数

很多时候瓶颈不在传得慢，而在传太多。把 100 个小包合并成 1 个大包，性能会好得多。

5.1 把多次小传输合并成一次大传输

比如你有 1 万个数字要发给 worker，每个数字单独发，会产生 1 万次通信。如果把它们拼成一个列表，一次发过去，通信次数就从 1 万次降到 1 次。虽然数据总量没变，但省掉了每次通信的往返开销和锁竞争。批量处理是性价比极高的优化方式。

# -- coding: utf-8 -*-
# Python 技术栈：multiprocessing 的 Queue（批量发送）
from multiprocessing import Process, Queue

def batch_worker(q):
    while True:
        tasks = q.get()
        if tasks is None:
            break
        # 一次处理一整批任务，这里仅做示例，不真正计算
        print("处理了一整批，数量：", len(tasks))

if __name__ == "__main__":
    q = Queue()
    p = Process(target=batch_worker, args=(q,))
    p.start()
    # 把 1000 个任务打包成一个列表，一次性发送
    batch = list(range(1000))
    q.put(batch)
    # 发送终止信号
    q.put(None)
    p.join()

5.2 使用工作进程池

multiprocessing.Pool 是为了简化任务分发设计的，它内部自己管理 workers，并且支持批量传输。imap 方法允许你传入一个 chunksize，它会一次性从迭代器里取出一批项，打包后发给某个 worker。这样你在写代码时不用手动批量，底层已经帮你优化好了。

# -- coding: utf-8 -*-
# Python 技术栈：multiprocessing 的 Pool
from multiprocessing import Pool
import math

def calc(x):
    # 求平方根，代表一个轻量计算任务
    return math.sqrt(x)

if __name__ == "__main__":
    # 生成 10 万个数字
    nums = range(100000)
    # 创建 4 个 worker 的进程池
    with Pool(4) as pool:
        # chunksize=1000，底层每次会把 1000 个任务打包成一批再发送
        results = pool.imap(calc, nums, chunksize=1000)
        # 累加所有结果，这一步会真正触发任务分发
        total = sum(results)
        print("总和：", total)

如果你的 worker 之间需要保存状态，Pool 也可以配合 initializer 来初始化每个 worker 的全局变量，这样每个 worker 不用重复加载资源，也能减少跨进程传输。不过要注意，initializer 传的参数也会经历一次序列化，所以适合传小配置，不适合传大数据。

六、怎么选：应用场景、优缺点及注意事项

6.1 不同通信方式对比

  • Queue：用法最简单，支持多生产多消费，但序列化和锁开销大，适合小任务、低频次。
  • Pipe：效率和速度居中，适合两个进程间的大规模数据传输，但只能点对点，需要注意死锁。
  • SharedMemory：速度最快，但需要手动同步和管理生命周期，适合高频率、大块数据共享。
  • Pool + imap：适合任务量巨大、每个任务很轻的场景，能有效降低通信次数，代码也容易维护。

另外，multiprocessing.Manager 也可以共享更高级的对象，但它其实是通过一个独立的服务进程来代理操作，每次访问都要经历“请求-响应”的往返通信，速度甚至比 Queue 还慢。所以除非你要共享一个 dict 或者 list 给十来个进程频繁读写，否则不建议用它。

6.2 场景建议

如果你在做爬虫系统，每个 URL 很短，用 Queue 或 Pool 都行。如果你是做机器学习数据预处理，每个样本是一大串文本，建议用共享内存或 mmap。如果你在高频数据服务里，要求极低延迟，共享内存几乎是唯一选择。

6.3 注意事项

  • 序列化格式要简单。能用 bytes 传就别用自定义类，能传 json 就别传 pickle 对象。
  • 数据量很大时，先压缩再传送。比如用 zlib 压缩字符串，CPU 压缩开销小于传输时间，往往更划算。
  • 共享内存记得 unlink，程序崩溃时也要考虑清理。
  • 一定要把代码入口放在 if name == "main": 里面，特别是 Windows 平台上，否则会疯狂 fork 子进程。
  • 注意进程的退出和清理，避免僵尸进程。

6.4 压箱底的小技巧

如果两个进程需要频繁交互，可以把“一问一答”的模式改成“批量处理”模式；如果某些 worker 只需要读数据，不需要写，可以用只读共享内存，避免加锁。还有，不要过度使用共享内存，因为共享内存相当于放弃了进程隔离保护，一旦写错，整个服务都会崩溃。

还有一个小经验：如果你的数据是字符串，可以考虑先转换成字节；如果是数字，尽量用 struct 或者 array 包装成二进制。因为 pickle 在处理二进制数据时虽然有点开销，但比处理 Python 对象要快得多。另外，进程数量不是越多越好，当进程数量超过 CPU 核心数时，上下文切换也会成为瓶颈。

七、总结

多进程的数据传输优化不是玄学，核心就两条：降低单次传输的成本，减少传输的次数。降低单次成本的方法是用共享内存或管道，减少次数的方法是批量合并。实际开发中，你要先观察自己的场景，数据量是几百字节还是几兆？频率是一秒一次还是一秒一万次？进程数是两个还是二十个？这些都会影响你的选择。记住，最快的传输就是不传输，尽量把数据留在同一个进程里；如果必须跨进程，就用最合适的方式去传。

可能有人会说，既然多进程通信这么麻烦，直接用 Redis 或消息队列不好吗？外部中间件确实能解耦，但它们需要网络往返，性能和系统复杂度都更高。在单机多进程场景下，Python 标准库的这些工具已经能做得很好。掌握了本节这些思路，你再去面对任何数据传递问题，都能自己动手找到最优解。希望这篇文章能让你少踩几个坑，写出更快更稳的多进程服务。