如何结合Queue使用join()且避免进程挂起？以及如何确保所有进程执行完毕后再从队列中收集数据？

我完全懂你现在的困境——不调用join()的话，主进程会在子进程还在跑的时候就去队列捞数据，结果拿到的肯定是不完整的；可一旦调用join()，所有子进程好像都卡壳了，主进程就僵在那里，后面取数据的代码根本没机会执行。

问题的核心原因 #

这是multiprocessing.Queue的典型坑：它内部有个默认大小的缓冲区，当子进程调用put()往队列塞数据时，如果缓冲区满了，这个操作会直接阻塞子进程，得等主进程调用get()腾出空间才能继续。但你现在的逻辑是先等所有子进程结束（join()），再去取数据——这就形成了死锁：子进程卡着等主进程取数据，主进程卡着等子进程结束，两边都动弹不得。

两种可行的解决办法 #

办法1：在join()前启动线程消费队列

这样能避免子进程因为队列满而阻塞，同时保证所有数据都被收集完整：

import multiprocessing as mp
from threading import Thread

def foo(q1, q2, data):
    # 模拟实际业务逻辑生成结果
    var1 = f"result1_{data}"
    var2 = f"result2_{data}"
    q1.put(var1)
    q2.put(var2)

def consume_queue(q, result_list):
    while True:
        try:
            # 超时时间可根据实际场景调整
            item = q.get(timeout=1)
            result_list.append(item)
        except mp.queues.Empty:
            # 队列为空且超时，说明所有数据都取完了
            break

def main_function(list_of_stuff):
    q1 = mp.Queue()
    q2 = mp.Queue()
    jobs = []
    list1 = []
    list2 = []

    # 先启动消费队列的线程
    t1 = Thread(target=consume_queue, args=(q1, list1))
    t2 = Thread(target=consume_queue, args=(q2, list2))
    t1.start()
    t2.start()

    # 启动所有子进程
    for data in list_of_stuff:
        p = mp.Process(target=foo, args=(q1, q2, data))
        p.start()
        jobs.append(p)

    # 等待所有子进程执行完毕
    for job in jobs:
        job.join()

    # 等待消费线程结束
    t1.join()
    t2.join()

    return list1, list2

# 测试代码
if __name__ == "__main__":
    res1, res2 = main_function([1,2,3,4])
    print(res1)
    print(res2)

办法2：用Manager.Queue()替代普通Queue

Manager.Queue()由进程间共享的管理器创建，它的缓冲区管理逻辑和普通Queue不同，不会因为缓冲区满而阻塞子进程，能直接避免死锁：

import multiprocessing as mp

def foo(q1, q2, data):
    var1 = f"result1_{data}"
    var2 = f"result2_{data}"
    q1.put(var1)
    q2.put(var2)

def main_function(list_of_stuff):
    with mp.Manager() as manager:
        q1 = manager.Queue()
        q2 = manager.Queue()
        jobs = []

        # 启动子进程
        for data in list_of_stuff:
            p = mp.Process(target=foo, args=(q1, q2, data))
            p.start()
            jobs.append(p)

        # 等待所有子进程结束
        for job in jobs:
            job.join()

        # 收集结果
        list1 = []
        while not q1.empty():
            list1.append(q1.get())
        list2 = []
        while not q2.empty():
            list2.append(q2.get())

        return list1, list2

# 测试代码
if __name__ == "__main__":
    res1, res2 = main_function([1,2,3,4])
    print(res1)
    print(res2)

额外推荐：用Pool简化逻辑 #

如果你的场景只是简单的多进程计算+结果收集，完全可以不用手动管理队列和进程，multiprocessing.Pool会帮你搞定所有细节：

import multiprocessing as mp

def foo(data):
    var1 = f"result1_{data}"
    var2 = f"result2_{data}"
    return var1, var2

def main_function(list_of_stuff):
    with mp.Pool() as pool:
        # 自动分配进程并收集结果
        results = pool.map(foo, list_of_stuff)
    # 拆分结果到两个列表
    list1 = [r[0] for r in results]
    list2 = [r[1] for r in results]
    return list1, list2

# 测试代码
if __name__ == "__main__":
    res1, res2 = main_function([1,2,3,4])
    print(res1)
    print(res2)

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