Python 2.6 multiprocessing.Queue 与线程兼容？

Question

我正在试验 Python 2.6 中的新多处理模块。我正在创建多个进程，每个进程都有自己的 multiprocessor.JoinableQueue 实例。每个进程生成一个或多个共享 JoinableQueue 实例(通过每个线程的 __init__ 方法传入)的工作线程(threading.Thread 的子类)。它似乎通常可以工作，但偶尔会出现不可预测的失败并出现以下错误:

  File "C:\Documents and Settings\Brian\Desktop\testscript.py", line 49, in run
    self.queue.task_done()
  File "C:\Python26\lib\multiprocessing\queues.py", line 293, in task_done
    raise ValueError('task_done() called too many times')
ValueError: task_done() called too many times

我的 Queue get() 和 task_done() 调用彼此紧随其后，因此它们应该相等。有趣的是，这似乎只有在 get() 和 task_done() 之间完成的工作非常快时才会发生。插入一个小的 time.sleep(0.01) 似乎可以缓解这个问题。

知道发生了什么事吗？我可以使用带线程的多处理器队列而不是更传统的队列 (Queue.Queue) 吗？

谢谢!

-布赖恩

Answer 1

我还没有在 2.6 中尝试多处理，但我玩了很多 pyprocessing(在 2.5 中被称为)。

我可以看到您正在寻找许多进程，每个进程分别生成一组线程。

由于您使用的是多处理模块，我建议使用多进程而不是多线程方法，这样您会遇到更少的问题，例如死锁等。

创建队列对象。 http://pyprocessing.berlios.de/doc/queue-objects.html

要创建多进程环境，请使用池:http://pyprocessing.berlios.de/doc/pool-objects.html它将为您管理工作进程。然后，您可以将异步/同步应用于工作人员，如果需要，还可以为每个工作人员添加回调。但请记住回调是一个常见的代码块，它应该立即返回(如文档中所述)

一些附加信息:如果需要，创建一个经理 http://pyprocessing.berlios.de/doc/manager-objects.html管理对队列对象的访问。您必须为此共享队列对象。但优点是，一旦共享和管理，您就可以通过创建代理对象在整个网络上访问这个共享队列。这将使您能够在任何网络节点上将集中式共享队列对象的方法作为(显然) native 方法进行调用。

这是文档中的代码示例

可以在一台机器上运行管理器服务器并让客户端从其他机器上使用它(假设所涉及的防火墙允许)。运行以下命令为远程客户端可以使用的共享队列创建一个服务器:

>>> from processing.managers import BaseManager, CreatorMethod
>>> import Queue
>>> queue = Queue.Queue()
>>> class QueueManager(BaseManager):
...     get_proxy = CreatorMethod(callable=lambda:queue, typeid='get_proxy')
...
>>> m = QueueManager(address=('foo.bar.org', 50000), authkey='none')
>>> m.serve_forever()

一个客户端可以按如下方式访问服务器:

>>> from processing.managers import BaseManager, CreatorMethod
>>> class QueueManager(BaseManager):
...     get_proxy = CreatorMethod(typeid='get_proxy')
...
>>> m = QueueManager.from_address(address=('foo.bar.org', 50000), authkey='none')
>>> queue = m.get_proxy()
>>> queue.put('hello')

如果您坚持安全线程的东西，PEP371(多处理)引用此 http://code.google.com/p/python-safethread/