Python 队列的使用适用于线程，但(显然)不适用于多处理

Question

许多有关多处理使用的教程似乎并没有完全解决为什么下面的技术适用于线程但不适用于多处理。

为什么这不适用于多处理，以及我尝试要做的事情的实现是什么？谢谢!

线程实现(工作正常，对我来说很有意义):

from threading import Thread
from Queue import Queue
from time import sleep    

"""threading functions"""
def producer_thread(n):
    for x in range(10):
        thread_q.put(n)

def consumer_thread():
    while True:
        item = thread_q.get()
        print item

if __name__ == '__main__':
    thread_q = Queue()

    """works fine"""
    p_thread = Thread(target=producer_thread, args=(10,))
    c_thread = Thread(target=consumer_thread)
    c_thread.daemon=True
    p_thread.start(); c_thread.start()
    p_thread.join()
    """prevents c_thread daemon process from cancelling prematurely"""
    sleep(.001)

输出:

10
10
10
10
10
10
10
10
10
10

多处理实现(似乎与线程相同，但根本不起作用):

from multiprocessing import Process, freeze_support
from Queue import Queue

"""multiprocessing functions"""
def producer_process(n):
    for x in range(10):
        process_q.put(n)

def consumer_process():
    while True:
        item = process_q.get()
        print item
#            
if __name__ == '__main__':
    freeze_support()
    process_q = Queue()        
    """computer explodes"""
    p_process = Process(target=producer_process, args=(10,))
    c_process = Process(target=consumer_process)
    c_process.daemon=True
    p_process.start(); c_process.start()
    p_process.join()

输出:

Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "C:\Users\J\Anaconda\lib\multiprocessing\forking.py", line 381, in main
    self = load(from_parent)
  File "C:\Users\J\Anaconda\lib\pickle.py", line 1378, in load
    return Unpickler(file).load()
  File "C:\Users\J\Anaconda\lib\pickle.py", line 858, in load
    dispatch[key](self)
  File "C:\Users\J\Anaconda\lib\pickle.py", line 1090, in load_global
    klass = self.find_class(module, name)
  File "C:\Users\J\Anaconda\lib\pickle.py", line 1126, in find_class
    klass = getattr(mod, name)
AttributeError: 'module' object has no attribute 'get_successors'
Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "C:\Users\J\Anaconda\lib\multiprocessing\forking.py", line 381, in main
    self = load(from_parent)
  File "C:\Users\J\Anaconda\lib\pickle.py", line 1378, in load
    return Unpickler(file).load()
  File "C:\Users\J\Anaconda\lib\pickle.py", line 858, in load
    dispatch[key](self)
  File "C:\Users\J\Anaconda\lib\pickle.py", line 1090, in load_global
    klass = self.find_class(module, name)
  File "C:\Users\J\Anaconda\lib\pickle.py", line 1126, in find_class
    klass = getattr(mod, name)
AttributeError: 'module' object has no attribute 'get_successors'
Process Process-33:
Traceback (most recent call last):
  File "C:\Users\J\Anaconda\lib\multiprocessing\process.py", line 258, in _bootstrap
    self.run()
  File "C:\Users\J\Anaconda\lib\multiprocessing\process.py", line 114, in run
    self._target(*self._args, **self._kwargs)
  File "C:\Users\J\Documents\Python Scripts\producer_consumer_test.py", line 18, in consumer
    item = q.get()
NameError: global name 'q' is not defined
Process Process-32:
Traceback (most recent call last):
  File "C:\Users\J\Anaconda\lib\multiprocessing\process.py", line 258, in _bootstrap
    self.run()
  File "C:\Users\J\Anaconda\lib\multiprocessing\process.py", line 114, in run
    self._target(*self._args, **self._kwargs)
  File "C:\Users\J\Documents\Python Scripts\producer_consumer_test.py", line 14, in producer
    q.put(n)
NameError: global name 'q' is not defined
Process Process-34:
Traceback (most recent call last):
  File "C:\Users\J\Anaconda\lib\multiprocessing\process.py", line 258, in _bootstrap
    self.run()
  File "C:\Users\J\Anaconda\lib\multiprocessing\process.py", line 114, in run
    self._target(*self._args, **self._kwargs)
  File "C:\Users\J\Documents\Python Scripts\producer_consumer_test.py", line 14, in producer
    q.put(n)
NameError: global name 'q' is not defined
Process Process-35:
Traceback (most recent call last):
  File "C:\Users\J\Anaconda\lib\multiprocessing\process.py", line 258, in _bootstrap
    self.run()
  File "C:\Users\J\Anaconda\lib\multiprocessing\process.py", line 114, in run
    self._target(*self._args, **self._kwargs)
  File "C:\Users\J\Documents\Python Scripts\producer_consumer_test.py", line 18, in consumer
    item = q.get()
NameError: global name 'q' is not defined
Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "C:\Users\J\Anaconda\lib\multiprocessing\forking.py", line 381, in main
    self = load(from_parent)
  File "C:\Users\J\Anaconda\lib\pickle.py", line 1378, in load
    return Unpickler(file).load()
  File "C:\Users\J\Anaconda\lib\pickle.py", line 858, in load
    dispatch[key](self)
  File "C:\Users\J\Anaconda\lib\pickle.py", line 1090, in load_global
    klass = self.find_class(module, name)
  File "C:\Users\J\Anaconda\lib\pickle.py", line 1126, in find_class
    klass = getattr(mod, name)
AttributeError: 'module' object has no attribute 'consumer'
Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "C:\Users\J\Anaconda\lib\multiprocessing\forking.py", line 381, in main
    self = load(from_parent)
  File "C:\Users\J\Anaconda\lib\pickle.py", line 1378, in load
    return Unpickler(file).load()
  File "C:\Users\J\Anaconda\lib\pickle.py", line 858, in load
    dispatch[key](self)
  File "C:\Users\J\Anaconda\lib\pickle.py", line 1090, in load_global
    klass = self.find_class(module, name)
  File "C:\Users\J\Anaconda\lib\pickle.py", line 1126, in find_class
    klass = getattr(mod, name)
AttributeError: 'module' object has no attribute 'producer'

Answer 1

导入队列适用于多线程应用:https://docs.python.org/2/library/queue.html不适用于多进程应用程序。

来自多处理导入队列适用于多进程应用程序:https://docs.python.org/2/library/multiprocessing.html#exchanging-objects-between-processes

根据文档 multiprocessing.Queue “是 Queue.Queue 的近乎克隆”

除了 multiprocessing.Queue 之外，还有 JoinableQueue，它具有 task_done() 和 join() 方法，以备您需要时使用。

在您的示例中，我认为您不需要 JoinableQueue。你尝试过这个吗:

from multiprocessing import (Process, Queue, freeze_support)

def producer(q, n):
    for x in range(n):
        q.put(x)
    q.put("end")


def consumer(q):
    while True:
        item = q.get()
        if item == "end":
            break
        print item

if __name__ == '__main__':
    freeze_support()
    q = Queue()
    c = Process(target=consumer, args=(q,))
    c.start()
    p = Process(target=producer, args=(q, 10))
    p.start()
    c.join()

在 Linux 和 Windows 中测试。