Python multiprocessing队列存大数据时子进程无法抛异常问题

问题本质 #

你遇到的核心问题是：put_nowait的非阻塞特性仅针对队列的存储环节，但数据必须先通过pickle序列化才能存入队列。超大数据的序列化过程耗时极长，程序会卡在序列化步骤，根本没机会执行队列满的检查逻辑，自然不会抛出异常。而小数据序列化速度快，能瞬间完成检查，所以能正常触发Queue.Full异常（你提到的StopIteration大概率是代码中迭代逻辑的异常，因put操作卡住导致迭代无法走到终止步骤而未触发）。

解决办法 #

1. 先检查队列状态再处理大数据 #

在生成或序列化大数据之前，先调用q.full()判断队列是否已满。如果队列满了直接处理异常，避免浪费时间在不必要的大数据序列化上。

小提醒：q.full()和put_nowait之间存在微小的竞态条件（比如刚判断完队列未满，其他进程就把队列占满了），但用try-except兜底即可，总比程序卡住强。

示例代码：

from multiprocessing import Process, Queue

def worker(q):
    if q.full():
        print("队列已满，无法存入数据")
        raise StopIteration("队列已满，终止迭代")
    
    big_data = [[9] * 10000000]
    try:
        q.put_nowait(big_data)
    except Queue.Full:
        print("竞态条件：队列在检查后变满，存入失败")
        raise StopIteration("存入失败，终止迭代")

if __name__ == "__main__":
    q = Queue(maxsize=1)
    q.put("占位数据")  # 让队列先处于满状态
    p = Process(target=worker, args=(q,))
    p.start()
    p.join()

2. 将大数据拆分为小块分批存入 #

把超大数据拆分成多个小批次，分批放入队列。每个小块的序列化时间短，能及时触发队列满的检查逻辑，正常抛出异常。

示例代码：

from multiprocessing import Process, Queue

def worker(q):
    big_data = [[9] * 10000000]
    # 将大数据拆分为每1000个元素一组的小块
    chunks = [big_data[0][i:i+1000] for i in range(0, len(big_data[0]), 1000)]
    
    for chunk in chunks:
        try:
            q.put_nowait([chunk])
        except Queue.Full:
            print("队列已满，终止数据存入")
            raise StopIteration("存入失败，终止迭代")

if __name__ == "__main__":
    q = Queue(maxsize=1)
    q.put("占位数据")
    p = Process(target=worker, args=(q,))
    p.start()
    p.join()

3. 换用Pipe（一对一通信场景） #

如果不需要多生产者/多消费者的队列特性，可以改用multiprocessing.Pipe。Pipe的序列化开销更低，非阻塞操作能更快触发异常，但仅支持一对一的进程通信。

关键提示 #

put_nowait本身只会抛出Queue.Full异常，而非StopIteration。你遇到的StopIteration未触发，是因为put操作卡住导致迭代逻辑无法走到终止步骤。上述方案通过提前检查或拆分数据，确保迭代逻辑能正常触发预期异常。

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