multiprocessing.Queue 关闭以后再 get() 为什么会卡住？

import multiprocessing as mp

def p1(q):
    print("p1 start")
    print(q.get())
    print("p1 end")

if __name__ == "__main__":
    q = mp.Queue()
    q.close()
    mp.Process(target=p1, args=(q,)).start()
    print("closed")

输出，然后就卡在 q.get() 了：

closed
p1 start

但是如果在主进程 q.get() 就会正常的报错：

import multiprocessing as mp

def p1(q):
    print("p1 start")
    print(q.get())
    print("p1 end")

if __name__ == "__main__":
    q = mp.Queue()
    q.close()
    q.get() # <-- 在主进程 get()
    mp.Process(target=p1, args=(q,)).start()
    print("closed")

Traceback (most recent call last):
  File "/home/username/test.py", line 11, in <module>
    q.get()
  File "/usr/lib64/python3.12/multiprocessing/queues.py", line 100, in get
    raise ValueError(f"Queue {self!r} is closed")
ValueError: Queue <multiprocessing.queues.Queue object at 0x7f55c519ba40> is closed

7 条回复 • 2024-05-25 21:18:49 +08:00

lalalalacc

234 天前

在 Python 的 multiprocessing 模块中，Queue 用于进程间通信。当您尝试在关闭 (close()) Queue 后使用 get() 方法时，会遇到不同的行为，这取决于 get() 方法是在哪个进程中调用。

首先，需要明确的是，正常情况下，尝试在已关闭的 Queue 上调用 get() 应该会抛出一个 ValueError 异常，正如您在主进程中尝试时看到的那样。这是因为关闭 Queue 后，它不再允许任何进程进行读写操作，而尝试这样做会触发异常。

然而，您提到的在子进程中调用 get() 时程序卡住的情况，实际上是由于 multiprocessing.Queue 的行为特性导致的。在多数操作系统中，multiprocessing.Queue 实现依赖于底层的管道（ pipes ）和锁（ locks ）。当 Queue 被关闭时，其底层的管道也会被关闭，但是如果此时有进程尝试从这个已关闭的管道中读取数据，该进程会阻塞（ block ），因为它在等待管道中的数据，而这些数据永远不会到来。这就是为什么在子进程中调用 get() 会导致程序卡住的原因。

这种差异的根本原因在于，close() 方法实际上关闭了队列的底层资源，但并没有在所有上下文中显式地标记队列为“不可用”。因此，当在主进程中调用 get() 时，Queue 对象能够检查到它已经被关闭，并抛出一个异常。而在子进程中，尝试从这个关闭的 Queue 中 get() 数据时，进程会在底层的系统调用中阻塞，因为它在等待一个从未发生的事件。

为了避免这种情况，确保不要在关闭 Queue 后从它获取数据。如果需要在多个进程间安全地结束通信，可以考虑使用一个特殊的消息，如 None 或自定义的结束信号，来通知其他进程不再发送或接收数据，然后再关闭 Queue 。这样，所有进程都可以在尝试读取或写入 Queue 前先检查这个结束信号，从而避免在关闭 Queue 后还尝试进行操作。