multiprocessing在python中的高级应用-IPC 之 Queue

Dear 丶 2022-08-04 13:42 211阅读 0赞

multiprocessing模块支持进程间通信的两种主要形式：管道和队列。这两种方法都使用了消息传递实现的，但队列接口有意模仿线程程序中常见的队列用法。  
有关Queue编程实例[可以查看微博内容][Link 1]。

**Queue(\[maxsize\])**  
创建共享的进程队列。maxsize是队列中允许的最大项数。如果省略此参数，则无大小限制。底层队列使用管道和锁定实现。另外，还需要运行支持线程以便队列中的数据传输到底层管道中。  
Queue的实例q具有以下方法：

**q.cancel\_join\_thread()**  
不会再进程退出时自动连接后台线程。这可以防止join\_thread()方法阻塞。

**q.close()**  
关闭队列，防止队列中加入更多数据。调用此方法时，后台线程将继续写入那些已入队列但尚未写入的数据，但将在此方法完成时马上关闭。如果q被垃圾收集，将自动调用此方法。关闭队列不会在队列使用者中生成任何类型的数据结束信号或异常。例如，如果某个使用者正被阻塞在get（）操作上，关闭生产者中的队列不会导致get（）方法返回错误。

**q.empty()**  
如果调用此方法时 q为空，返回True。如果其他进程或线程正在往队列中添加项目，结果是不可靠的。也就是说，在返回和使用结果之间，队列中可能已经加入新的项目。

**q.full()**  
如果q已满，返回为True. 由于线程的存在，结果也可能是不可靠的（参考q.empty（）方法）。。

**q.get( \[ block \[ ,timeout \] \] )**  
返回q中的一个项目。如果q为空，此方法将阻塞，直到队列中有项目可用为止。block用于控制阻塞行为，默认为True. 如果设置为False，将引发Queue.Empty异常（定义在Queue模块中）。timeout是可选超时时间，用在阻塞模式中。如果在制定的时间间隔内没有项目变为可用，将引发Queue.Empty异常。

**q.get\_nowait( )**  
同q.get(False)方法。

**q.join\_thread()**  
连接队列的后台线程。此方法用于在调用q.close()方法后，等待所有队列项被消耗。默认情况下，此方法由不是q的原始创建者的所有进程调用。调用q.cancel\_join\_thread()方法可以禁止这种行为。

**q.put(item \[, block \[,timeout \] \] )**  
将item放入队列。如果队列已满，此方法将阻塞至有空间可用为止。block控制阻塞行为，默认为True。如果设置为False，将引发Queue.Empty异常（定义在Queue库模块中）。timeout指定在阻塞模式中等待可用空间的时间长短。超时后将引发Queue.Full异常。

**q.qsize()**  
返回队列中目前项目的正确数量。此函数的结果并不可靠，因为在返回结果和在稍后程序中使用结果之间，队列中可能添加或删除了项目。在某些系统上，此方法可能引发NotImplementedError异常。

**JoinableQueue(\[maxsize\])**  
创建可连接的共享进程队列。这就像是一个Queue对象，但队列允许项目的使用者通知生产者项目已经被成功处理。通知进程是使用共享的信号和条件变量来实现的。  
JoinableQueue的实例p除了与Queue对象相同的方法之外，还具有以下方法：

**q.task\_done()**  
使用者使用此方法发出信号，表示q.get()返回的项目已经被处理。如果调用此方法的次数大于从队列中删除的项目数量，将引发ValueError异常。

**q.join()**  
生产者将使用此方法进行阻塞，直到队列中所有项目均被处理。阻塞将持续到为队列中的每个项目均调用q.task\_done()方法为止。  
下面的例子说明如何建立永远运行的进程，使用和处理队列上的项目。生产者将项目放入队列，并等待它们被处理。

import multiprocessing
    
    def consumer(input_q):
        while True:
            item=input_q.get()
            #处理项目
            print item #此处替换为有用的工作
            #发出信号通知任务完成
            input_q.task_done()
    def producer(sequence,output_q):
        for item in sequence:
            #将项目放入队列
            output_q.put(item)      
    #建立进程
    if __name__=='__main__':
        q=multiprocessing.JoinableQueue()
        #运行使用者进程
        cons_p=multiprocessing.Process(target=consumer,args=(q,))
        cons_p.daemon=True   #定义该进程为后台运行 True
        cons_p.start()
        #生产项目，sequence代表要发送给使用者的项目序列
        #在时间中，这可能是生成器的输出或通过一些其他方式生产出来
        sequence=[1,2,3,4]#range(5)[1:5]
        producer(sequence,q)
        #等待所有项目被处理
        q.join()

如果需要，可以在同一个队列中放置多个进程，也可以从同一个队列中获取多个进程。例如，如果要构造使用者进程池，可以编写下面这样的代码：

#这里只是对main函数作出了修改 
    if __name__=='__main__':
        q=multiprocessing.JoinableQueue()
        #创建一些使用者进程
        cons_p1=multiprocessing.Process(target=consumer,args=(q,))
        cons_p1.daemon=True
        cons_p1.start()
    
        cons_p2=multiprocessing.Process(target=consumer,args=(q,))
        cons_p2.daemon=True
        cons_p2.start()
        #生产项目。sequence代表要发送给使用者的项目序列
        #在时间中，这可能是生成器的输出或者是通过一些其他方式生产出来的
        sequence=[1,2,3,4]
        producer(sequence,q)
        #等待所有项目被处理
        q.join()

在某些应用程序中 生产者需要通知使用者 他们不再生产任何项目而且应该关闭。为此 编写的代码中应该使用标志（sentinel）-指示完成的特殊值。下面这个例子使用None作为标志说明这个概念：

def consumer(imput_q):
        while True:
            item=imput_q.get()
            if item is None:
                break
            #处理项目
            print item
        #关闭
        print "Coonsumer done"
    def producer(sequence,output_q):
        for item in sequence:
            #吧项目放入队列
            output_q.put(item)
    if __name__=="__main__":
        q=multiprocessing.Queue()
        #启动使用者进程
        cons_p=multiprocessing.Process(target=consumer,args=(q,))
        cons_p.start()
        #生产项目
        sequence=[1,2,3,4]
        producer(sequence,q)
        #在队列上安置标志 发出完成信号
        q.put(None)
        #等待使用者进程关闭
        cons_p.join()

如果像上面这个例子中那样使用标志，一定要在队列上为每个使用者上都安置标志。例如 如果有三个使用者进程使用队列上的项目，那么生产者需要在队列上安置三个标志，才能让所有使用者都关闭。

[Link 1]: http://blog.csdn.net/winterto1990/article/details/48025881