multiprocessing.Queue 作为 arg 池工人中止工人的执行

Question

我实际上很难相信我遇到了我遇到的问题，这似乎是 python 多处理模块中的一个大错误......无论如何我遇到的问题是每当我将 multiprocessing.Queue 作为参数传递给 multiprocessing.Pool 工作者时，池工作者永远不会执行其代码。即使是在 python docs 中找到的示例代码的略微修改版本的非常简单的测试中，我也能够重现此错误。

这是队列示例代码的原始版本：

from multiprocessing import Process, Queue

def f(q):
    q.put([42, None, 'hello'])


if __name__ == '__main__':
    q = Queue()
    p = Process(target=f, args=(q,))
    p.start()
    print(q.get())  # prints "[42, None, 'hello']"
    p.join()

这是我对队列示例代码的修改版本：

from multiprocessing import Queue, Pool

def f(q):
    q.put([42, None, 'hello'])

if __name__ == '__main__':
    q = Queue()
    p = Pool(1)
    p.apply_async(f,args=(q,))
    print(q.get()) # prints "[42, None, 'hello']"
    p.close()
    p.join()

我所做的只是让 p 成为大小为 1 的进程池，而不是 multiprocessing.Process 对象，结果是代码永远挂在 print 语句上，因为没有任何东西写入队列！当然，我以原始形式对此进行了测试，并且效果很好。我的操作系统是 windows 10，我的 python 版本是 3.5.x，有人知道为什么会这样吗？

更新：仍然不知道为什么此示例代码适用于 multiprocessing.Process 而不是 multiprocessing.Pool 但我找到了 work around 我很满意（Alex Martelli 的回答）。显然，您可以只创建一个 multiprocessing.Queues 的全局列表并传递每个进程和索引以供使用，我将避免使用托管队列，因为它们速度较慢。感谢客人向我展示链接。

Answer 1

问题

当您调用apply_async 时，它会返回一个AsyncResult 对象并将工作负载分配给单独的线程（另请参见this answer）。该线程遇到Queue对象不能为pickled的问题，因此请求的工作无法分发（并最终执行）。我们可以通过调用AsyncResult.get看到这一点：

r = p.apply_async(f,args=(q,))
r.get()

引发RuntimeError:

RuntimeError: Queue objects should only be shared between processes through inheritance

但是，这个RuntimeError 仅在您请求结果后才会在主线程中引发，因为它实际上发生在不同的线程中（因此需要一种传输方式）。

那么当你这样做时会发生什么

p.apply_async(f,args=(q,))

是目标函数f 永远不会被调用，因为它的一个参数（q）不能被腌制。因此q 永远不会收到一个项目并保持为空，因此在主线程中对q.get 的调用将永远阻塞。

解决方案

使用apply_async，您不必手动管理结果队列，但它们很容易以AsyncResult 对象的形式提供给您。所以你可以修改代码以简单地从目标函数返回：

from multiprocessing import Queue, Pool

def f():
    return [42, None, 'hello']

if __name__ == '__main__':
    q = Queue()
    p = Pool(1)
    result = p.apply_async(f)
    print(result.get())