Python multiprocessing Queue与Process组合行为优化及异常修复

原代码存在三个直接导致卡死、偶发异常的核心问题：

• multiprocessing.Queue的empty()、qsize()方法在多进程并发场景下不具备可靠性：多个进程并发访问队列时，方法返回值和队列实际状态可能存在时间差，比如多个进程同时判定队列非空，但实际队列仅剩1个任务，某一个进程取走任务后，其余进程调用无参数的get()方法会永久阻塞，这就是调试时观察到卡在in_data最后一个元素的直接原因。
• Queue底层基于操作系统管道实现，自带固定大小的缓冲区，当写入队列的数据量超过缓冲区上限时，写入进程会阻塞直到有其他进程从队列取走数据。原代码先调用join()等待所有子进程结束、之后才从out_data队列取结果，当nres较大时（比如300），子进程写入的numpy数组很快占满缓冲区，子进程卡在写入步骤无法退出，主进程卡在join()步骤等待子进程退出，形成双向死锁，这也是nres=300时任务永远无法成功的原因。
• 任务函数calculate_row嵌套定义在主函数内部，在Windows、macOS默认的spawn多进程启动模式下，嵌套函数无法被正常序列化，会触发额外的运行时异常。

修复逻辑遵循Python多进程编程的最佳实践：

• 用哨兵值替代empty()方法做任务结束判断，所有任务入队后，为每个子进程入队一个None作为结束标记，子进程取到该标记就主动退出，完全避免阻塞。
• 调整执行顺序：先从out_data队列接收完全部计算结果，再调用join()回收子进程，从根源上避免缓冲区满导致的死锁。
• 将任务函数提升到模块顶层，避免跨进程序列化失败问题。
• 优化任务生成逻辑，去掉低效的集合差集运算，减少不必要的性能损耗。

from multiprocessing import Process, Queue
import numpy as np


def calculate_row(nres, in_data, out_data):
    """计算结果矩阵的单行数据"""
    while True:
        task = in_data.get()
        # 哨兵值标记当前子进程所有任务处理完成
        if task is None:
            break
        res, others = task
        row = np.zeros((nres,))
        for other in others:
            # 可替换为实际业务计算逻辑
            row[other] = other
        out_data.put((res, row))


def start_multicore_task(nres, cores_per_task):
    in_data = Queue()
    out_data = Queue()

    total_tasks = 0
    # 预先生成所有任务
    for res1 in range(nres):
        others = list(range(res1 + 1, nres))
        if others:
            in_data.put((res1, others))
            total_tasks += 1

    # 为每个子进程添加结束哨兵
    for _ in range(cores_per_task):
        in_data.put(None)

    ps = []
    for _ in range(cores_per_task):
        p = Process(target=calculate_row, args=(nres, in_data, out_data))
        p.start()
        ps.append(p)

    # 先接收所有计算结果，避免队列缓冲区满导致子进程阻塞
    corr = np.zeros((nres, nres))
    for _ in range(total_tasks):
        res, row = out_data.get()
        corr[res, :] = row

    # 所有结果接收完成后再回收子进程，彻底避免死锁
    for p in ps:
        p.join()

    return corr

• 原代码每次生成任务时通过大集合差集计算others列表，时间复杂度高，优化后直接通过range(res1 + 1, nres)生成目标列表，大任务量下性能提升显著。
• 预先统计总任务数，接收结果时按固定数量读取，不再依赖不可靠的qsize()方法判断队列剩余数据量，逻辑完全确定，不会出现偶发异常。
• 所有代码遵循PEP8编程规范，函数添加规范的文档字符串，兼容性覆盖全平台多进程启动模式。

内容的提问来源于stack exchange，提问作者Francho Nerín Fonz