为何Python多进程队列会因返回值差异产生死锁？

问题原因分析 #

你遇到的死锁确实是因为f函数生成的数据量过大导致的：
每个元素是一个包含50000维numpy数组和浮点数的元组，10000个这样的元素总数据量远超multiprocessing.Queue的内部缓冲区上限。子进程执行q.put()时，缓冲区被占满后会进入阻塞状态，而主进程在q.get()时只能拿到部分数据，剩下的数据因为子进程阻塞无法继续写入，最终双方互相等待形成死锁。之前尝试交换get()和join()顺序或移除join()没用，是因为子进程根本没完成put()操作，一直卡在那里。

进程间传递大数据的标准方案 #

1. 使用Manager.Queue替代原生Queue

multiprocessing.Manager()创建的Queue由独立的管理进程托管，缓冲区限制更小，能处理更大规模的数据传输：

import time
from multiprocessing import Process, Manager
import numpy as np

def f(q):
    q.put([(np.random.rand(50000), 0.993) for _ in range(10000)])

def g(q):
    time.sleep(3)
    q.put('X' * 100000)

if __name__ == '__main__':
    # 使用Manager的Queue
    with Manager() as manager:
        queue = manager.Queue()
        p = Process(target=g, args=(queue,))
        p.start()
        obj = queue.get()
        p.join()
        print("Done g")
        print(obj[:10])  # 只打印前10个字符避免刷屏

        queue = manager.Queue()
        p = Process(target=f, args=(queue,))
        p.start()
        obj = queue.get()
        p.join()
        print("Done f")
        print(f"数据长度: {len(obj)}, 第一个数组形状: {obj[0][0].shape}")

2. 用共享内存传递numpy数组（推荐）

对于numpy这类数值型数据，使用共享内存可以避免序列化/反序列化的开销，直接在进程间共享内存区域，效率更高：

import time
from multiprocessing import Process
import numpy as np
from multiprocessing import shared_memory

def f(shm_name, shape, dtype):
    # 连接到已创建的共享内存
    existing_shm = shared_memory.SharedMemory(name=shm_name)
    # 创建共享内存对应的numpy数组
    arr = np.ndarray(shape, dtype=dtype, buffer=existing_shm.buf)
    # 填充数据
    arr[:] = np.random.rand(*shape)
    # 关闭共享内存连接
    existing_shm.close()

if __name__ == '__main__':
    # 定义数据形状和类型
    shape = (10000, 50000)
    dtype = np.float64
    # 创建共享内存
    shm = shared_memory.SharedMemory(create=True, size=np.prod(shape)*dtype.itemsize)
    # 创建主进程侧的numpy数组
    main_arr = np.ndarray(shape, dtype=dtype, buffer=shm.buf)
    
    p = Process(target=f, args=(shm.name, shape, dtype))
    p.start()
    p.join()
    
    # 使用数据
    print(f"共享内存数组形状: {main_arr.shape}")
    print(f"第一个元素值: {main_arr[0][0]}")
    
    # 释放共享内存
    shm.close()
    shm.unlink()

注：如果需要传递元组（数组+浮点数），可以额外用一个小Queue传递浮点数，或者再创建一块共享内存存储浮点数数组。

3. 磁盘文件中转（超大规模数据）

如果数据大到内存无法容纳，可以将数据写入临时文件，主进程读取文件内容：

import time
from multiprocessing import Process
import numpy as np
import tempfile

def f(temp_file_path):
    data = [(np.random.rand(50000), 0.993) for _ in range(10000)]
    # 保存数据到临时文件（用numpy的savez更高效）
    np.savez(temp_file_path, arrays=np.array([x[0] for x in data]), scores=np.array([x[1] for x in data]))

if __name__ == '__main__':
    with tempfile.NamedTemporaryFile(delete=False) as tmp:
        temp_path = tmp.name
    
    p = Process(target=f, args=(temp_path,))
    p.start()
    p.join()
    
    # 读取数据
    loaded = np.load(temp_path)
    arrays = loaded['arrays']
    scores = loaded['scores']
    print(f"加载的数组形状: {arrays.shape}, 分数长度: {len(scores)}")
    
    # 删除临时文件
    import os
    os.unlink(temp_path)

内容的提问来源于stack exchange，提问作者julian