在multiprocessing.Queue中放入multiprocessing.Queue会爆炸

Question

下面的代码抛出一个异常，并在python2.7和3.3中打印123。

from multiprocessing import Queue

class Pool(object):
    def __init__(self):
        self.q = Queue()

p = Pool()
p.q.put(p)
print(123)

这实际上是某种类型的竞争条件，可以在这里看到：

yuv@yuvpad2:~/$ python3.3 t.py
123
Traceback (most recent call last):
  File "/home/yuv/Downloads/Python-3.3.0/Lib/multiprocessing/queues.py", line 249, in _feed
yuv@yuvpad2:~/$ 

完整的错误是RuntimeError: Queue objects should only be shared between processes through inheritance，而回溯根本没有解释它是如何/在哪里发生的。问题的根源是队列中的对象不能引用队列。我的实际用例实际上是一个worker对象和一个池对象，其中一个worker向池的Queue报告它已经完成了工作。所以我想让worker把自己发送回worker Queue。

我不使用Queue.Queue的原因是因为在Python2.7中有一个bug，使得queue.get()忽略Ctrl-C，这很烦人。

有没有一种方法可以干净利落地完成这个模式？

The real problem code is on codepad

Answer 1

这个错误抱怨的是：

p.q.put(p)

在这里，您试图将一个引用Queue的对象放入队列中。队列用于进程之间的通信，其工作方式是在另一个进程中对您尝试放入的任何内容进行酸洗，然后取消对它的酸洗-但是酸洗Queue是不可能的，甚至没有意义。

这就是为什么当你尝试pickle一个队列时，你会得到你提到的错误：

>>> from multiprocessing import Queue
>>> q = Queue()
>>> import pickle
>>> pickle.dumps(q)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/usr/lib/python3.2/multiprocessing/queues.py", line 77, in __getstate__
    assert_spawning(self)
  File "/usr/lib/python3.2/multiprocessing/forking.py", line 51, in assert_spawning
    ' through inheritance' % type(self).__name__
RuntimeError: Queue objects should only be shared between processes through inheritance

如果您想使用队列在进程之间共享数据，一种方法是这样做：

class Worker(multiprocessing.Process):
    queue = multiprocessing.Queue()
    def run(self):
        print(self.queue.get())
        ...

有关更多示例，请查看docs

Answer 2

我猜有几种方法可以做到这一点，但如果不确切知道你不想做什么，就很难推荐一种。

我想最简单的方法是使用2个不同的队列来达到这个目的。一个用于新来的工人，另一个用于已完成的工人。

Answer 3

回溯没有显示代码中的问题的原因是multiprocessing.Queue类启动了一个后台线程，而异常是在该线程中生成的。我得到了下面的回溯。

Traceback (most recent call last):
  File "/usr/lib/python2.7/multiprocessing/queues.py", line 266, in _feed
    send(obj)
  File "/usr/lib/python2.7/multiprocessing/queues.py", line 77, in __getstate__
    assert_spawning(self)
  File "/usr/lib/python2.7/multiprocessing/forking.py", line 51, in assert_spawning
    ' through inheritance' % type(self).__name__
RuntimeError: Queue objects should only be shared between processes through inheritance

我怀疑...which是由线路启动的.

p.q.put(p)

...in您似乎正在将包含Queue对象的Pool对象放入Queue中，这是不允许的，因此出现错误。

如果您想要一个有用的解决方案，它将有助于明确您试图实现的目标。