使用multiprocessing Queue时程序为何挂起？

正在探索multiprocessing模块用法，参考某指南的代码运行时出现Runtime错误，添加if __name__ == "__main__":块后，程序仅打印两个进程中的print语句，随后挂起，无队列元素输出。

原指南代码：

import multiprocessing as mp

def addPositive(queue):
    print("I am in addPositive.")
    for i in range(1, 100):
        queue.put(i)

def addNegative(queue):
    print("I am in addNegative.")
    for i in range(-100, 0):
        queue.put(i)

myQueue = mp.Queue()
process1 = mp.Process(target=addPositive, args=(myQueue,))
process2 = mp.Process(target=addNegative, args=(myQueue,))
process1.start()
process2.start()
process1.join()
process2.join()
while myQueue:
    print(myQueue.get(), end=",")
myQueue.close()

修改后的代码：

import multiprocessing as mp


def addPositive(queue):
    print("I am in addPositive.")
    for i in range(1, 100):
        queue.put(i)


def addNegative(queue):
    print("I am in addNegative.")
    for i in range(-100, 0):
        queue.put(i)


def run():
    myQueue = mp.Queue()
    process1 = mp.Process(target=addPositive, args=(myQueue,))
    process2 = mp.Process(target=addNegative, args=(myQueue,))
    process1.start()
    process2.start()
    process1.join()
    process2.join()
    while myQueue:
        print(myQueue.get(), end=",")
    myQueue.close()

 
if __name__ == "__main__":
    run()

期望输出类似：

1,2,3,4,5,6,7,8,9,10,-100,-99,-98,...

环境信息：Python 3.9.6，macOS Monterey 12.4

核心问题 #

while myQueue:的判断逻辑错误——multiprocessing.Queue对象在布尔判断中始终返回True，不会因为队列空了变为False。当队列中的所有元素被取完后，myQueue.get()会进入阻塞状态，等待新元素加入，导致程序无限挂起。

解决方案 #

方案1：按已知元素总数循环取数

既然知道两个进程总共会向队列中放入99+99=198个元素，直接循环对应次数即可：

import multiprocessing as mp


def addPositive(queue):
    print("I am in addPositive.")
    for i in range(1, 100):
        queue.put(i)


def addNegative(queue):
    print("I am in addNegative.")
    for i in range(-100, 0):
        queue.put(i)


def run():
    myQueue = mp.Queue()
    process1 = mp.Process(target=addPositive, args=(myQueue,))
    process2 = mp.Process(target=addNegative, args=(myQueue,))
    process1.start()
    process2.start()
    process1.join()
    process2.join()
    
    # 循环固定次数取完所有元素
    for _ in range(198):
        print(myQueue.get(), end=",")
    
    myQueue.close()


if __name__ == "__main__":
    run()

方案2：用结束标记判断队列处理完成

在每个进程结束前向队列放入一个特殊标记（比如None），主进程循环取元素，直到收集到所有进程的结束标记：

import multiprocessing as mp


def addPositive(queue):
    print("I am in addPositive.")
    for i in range(1, 100):
        queue.put(i)
    queue.put(None)  # 放入结束标记


def addNegative(queue):
    print("I am in addNegative.")
    for i in range(-100, 0):
        queue.put(i)
    queue.put(None)  # 放入结束标记


def run():
    myQueue = mp.Queue()
    process1 = mp.Process(target=addPositive, args=(myQueue,))
    process2 = mp.Process(target=addNegative, args=(myQueue,))
    process1.start()
    process2.start()
    
    finished_processes = 0
    while finished_processes < 2:
        item = myQueue.get()
        if item is None:
            finished_processes += 1
        else:
            print(item, end=",")
    
    process1.join()
    process2.join()
    myQueue.close()


if __name__ == "__main__":
    run()

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