Python多进程场景下如何正确使用Queue.join()方法

基础逻辑先明确 #

• join()的作用是阻塞当前线程，直到队列中所有已取出的任务都被标记为完成
• 每次调用q.get()取出一个任务后，必须主动调用q.task_done()标记该任务处理完毕，join()才会统计到完成的任务数，当完成数等于队列总放入任务数时，join()才会解除阻塞。你所有代码里都没写q.task_done()，这是join()不能正常工作的核心原因。
• join()必须放在所有队列取任务逻辑的外部，也就是你把队列里所有任务都取完、并且所有任务都已经调用过task_done()之后，再调用join()做最终等待，绝对不能放在取任务的循环内部。

常见疑问解答 #

• 不推荐用q.not_empty作为循环条件，这是队列的内部私有属性，状态判断在并发场景下不可靠。如果你的队列是预先放满所有任务再开始取，可以直接按任务总数循环，或者捕获队列空的异常来退出循环。
• join()和q.not_empty的作用完全不同：q.not_empty是判断队列当前有没有未取出的任务，join()是等待所有已取出的任务处理完成，二者不存在替代关系。

两段错误代码的问题分析 #

第一段代码问题 #

import queue
def main():
q = queue.Queue()
[q.put((compute_primes, (1, 30000))) for _ in range(8)]
with multiprocessing.Pool(processes=4) as pool:
    while q.not_empty:
        result = q.get()
        function = pool.apply_async(result[0], args=(result[1][0], result[1][1]))
        function.get()
    q.join()

• 用了私有属性q.not_empty做循环条件，并发场景下可能队列还有任务就提前退出，或者空了还进入循环导致q.get()阻塞
• 没有调用q.task_done()，所以末尾的q.join()要么永远阻塞，要么逻辑不对直接返回，完全没有起到作用

第二段代码问题 #

def main():
q = queue.Queue()
[q.put((compute_primes, (1, 30000))) for _ in range(8)]
with multiprocessing.Pool(processes=4) as pool:
    while True:
        result = q.get()
        function = pool.apply_async(result[0], args=(result[1][0], result[1][1]))
        function.get()
        q.join()

• while True没有退出条件，队列任务取空后q.get()会永久阻塞，代码永远无法结束
• join()放在循环内部，每次处理完一个任务就调用一次join()，第一次调用的时候队列还有7个已取出/未取出的任务，没有全部标记task_done()，join()直接就卡住了。

修正后的参考写法（仅修复join逻辑，串行问题你后续自行调整） #

import queue
import multiprocessing
import time

def compute_primes(start, end):
    start_time = time.time()
    primes = []
    for number in range(start, end + 1):
        flag = True
        for i in range(2, number):
            if (number % i) == 0:
                flag = False
                break
        if flag:
            primes.append(number)
    end_time = time.time()
    print(f"Time taken: {end_time - start_time}\n"
          f"Amount primes: {len(primes)}")
    return primes

def main():
    q = queue.Queue()
    task_count = 8
    # 预先放入所有任务
    for _ in range(task_count):
        q.put((compute_primes, (1, 30000)))
    
    with multiprocessing.Pool(processes=4) as pool:
        # 按任务总数循环取，避免用不可靠的not_empty判断
        for _ in range(task_count):
            result = q.get()
            function = pool.apply_async(result[0], args=(result[1][0], result[1][1]))
            function.get()
            # 必须标记任务完成
            q.task_done()
        # 所有任务取完后，调用join等待全部任务处理完成
        q.join()

if __name__ == "__main__":
    main()

额外提示：你现在用的queue.Queue是线程安全队列，只能在同一个进程的多个线程间用，如果你后续要在子进程里操作队列，需要换成multiprocessing.Queue才支持进程间通信。

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