Python 2.6 multiprocessing.Queue与线程兼容？

Question

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

  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

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

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

谢谢!

布赖恩

Answer 1

您应该将队列对象作为目标的参数传递。

多处理文档中的示例：

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()

队列是线程和进程安全的。