multiprocessing.queue.get() 阻塞并且不返回答案

【问题标题】：multiprocessing.queue.get() blocks and does not returnmultiprocessing.queue.get() 阻塞并且不返回
【发布时间】：2023-03-04 03:39:01
【问题描述】：

我正在嵌入式 Linux 设备上构建一个网络服务器，该设备的 RAM 非常少（只有 256 MB）。该网络服务器应该能够使用subprocess.check_output 发出shell 命令，但由于每个check_output 似乎需要与父进程消耗的内存一样多的可用RAM，所以我使用多处理在开始时创建第二个进程，而Python 仍然不会消耗太多内存。然后这个进程使用multiprocessing.Queue 接收来自主进程的命令，执行它们并使用另一个multiprocessing.Queue 返回输出。这曾经有效，但我似乎有某种竞争条件导致整个事情陷入困境。

这是我复制问题的最小测试用例：

# shwrapper.py
from multiprocessing import Process, Queue

def f(iq,oq):
    oq.put("Ready")
    while True:
        oq.put(iq.get()+" out")

def init():
    iq = Queue()
    oq = Queue()
    p = Process(target=f, args=(iq,oq,))
    p.start()
    print oq.get()
    iq.put("test")
    print(oq.get())
init()

如果我尝试导入它，我会得到以下结果：

>>> import shwrapper
Ready

这里卡住了。现在我发出一个键盘中断：

^CProcess Process-1:
Traceback (most recent call last):
  File "/usr/lib/python27.zip/multiprocessing/process.py", line 258, in _bootstrap
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "shwrapper2.py", line 17, in <module>
    init()
  File "shwrapper2.py", line 15, in init
    print(oq.get())
  File "/usr/lib/python27.zip/multiprocessing/queues.py", line 117, in get
KeyboardInterrupt
    self.run()
  File "/usr/lib/python27.zip/multiprocessing/process.py", line 114, in run
    self._target(*self._args, **self._kwargs)
  File "shwrapper2.py", line 6, in f
    oq.put(iq.get()+" out")
  File "/usr/lib/python27.zip/multiprocessing/queues.py", line 117, in get
    res = self._recv()
KeyboardInterrupt

如您所见，它卡在 f 中的 iq.get() 处。 iq-Queue 报告始终为空，无论我放入什么。另一方面，oq-Queue 按预期工作。有什么想法我可以在这里做什么？

我的系统是 Python 2.7.3，在 PowerPC 上的 Linux 2.6.29.6 上运行。

如果我在我的 Windows 计算机上的 Python 2.7.10 上运行它，一切正常。

我知道 Python 2.7.3 和 Linux 2.6.29 很古老，但制造商没有更新版本，而且系统有点锁定。

【问题讨论】：

感谢这篇文章。我在 Python 3.6 和 MacOS Mojave 中遇到了同样的问题。

标签： python-2.7 queue multiprocessing

【解决方案1】：

我终于通过使用multiprocessing.queues.SimpleQueue 而不是multiprocessing.Queue 来修复它。根据文档，在这方面应该没有区别，但在源代码中 SimpleQueue 使用 Locks 使其适用于多个进程。

【讨论】：

【解决方案2】：

尝试使用get_notwait 而不是get。例如：

from multiprocessing import Queue
q = Queue()
try:
    q.get_nowait()
except:
    print("queue is empty")

或者您可以将timeout 添加到get 方法。例如：

from multiprocessing import Queue
q = Queue()
try:
    q.get(timeout=1)
except:
    print("queue is empty.")

另外，您只能捕获Empty 异常而不是一般异常：

from queue import Empty
try:
    q.get_nowait()
except Empty:
    print("queue is empty")

【讨论】：