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

multiprocessing模块支持进程间通信的两种主要形式:管道和队列。这两种方法都使用了消息传递实现的,但队列接口有意模仿线程程序中常见的队列用法。

有关Queue编程实例可以查看微博内容

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

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

版权声明:本文为博主原创文章,未经博主允许不得转载。

时间: 2024-08-29 08:20:37

multiprocessing在python中的高级应用-IPC 之 Queue的相关文章

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

作为使用队列的另一种形式,还可以使用管道在进程回见执行消息传递. Pipe( [ duplex]) 在进程之间创建一条管道,并返回元组(conn1,conn2),其中conn1和conn2是表示管道两端的Connection对象.默认情况下,管道是双向的.如果将duplex置为False,conn1只能用于接收,而conn2只能用于发送.必须在创建和启动使用管道的Process对象之前调用Pipe()方法. Pipe()方法返回的Connection对象的实例c具有以下方法和属性. c.clos

multiprocessing在python中的高级应用-进程

本篇主要讲解multiprocessing中的重要模块-进程. Process([group [,target [,name [,args [,kwargs]]]]]) 这个类表示运行在一个子进程中的任务,应该使用关键字参数来指定构造函数中的参数.target是当前进程启动时执行的可调用对象,args是传递给target的位置参数的元组,而kwargs是传递给target的关键字参数的字典.如果省略args和kwargs参数,将不带参数调用target.name是为进程指定描述性名称额字符串.g

multiprocessing在python中的高级应用-共享数据与同步

通常,进程之间彼此是完全孤立的,唯一的通信方式是队列或管道.但可以使用两个对象来表示共享数据.其实,这些对象使用了共享内存(通过mmap模块)使访问多个进程成为可能. Value( typecode, arg1, - argN, lock ) 在共享内容中常见ctypes对象.typecode要么是包含array模块使用的相同类型代码(如'i','d'等)的字符串,要么是来自ctypes模块的类型对象(如ctypes.c_int.ctypes.c_double等).所有额外的位置参数arg1,

multiprocessing在python中的高级应用-进程池

下面的类可以创建进程池,可以吧各种数据处理任务都提交给进程池.进程池提供的功能有点类似于列表解析和功能性编程操作(如映射-规约)提供的功能. Pool( [ numprocess [, initializer [, initargs] ] ] ) 创建工作进程池. numprocess是要创建的进程数.如果省略此参数,将使用cpu_count()的值.[这里简单介绍一下: from multiprocessing import cpu_count print(cpu_count()) #获得电脑

Python中的高级数据结构(转)

add by zhj: Python中的高级数据结构 数据结构 数据结构的概念很好理解,就是用来将数据组织在一起的结构.换句话说,数据结构是用来存储一系列关联数据的东西.在Python中有四种内建的数据 结构,分别是List.Tuple.Dictionary以及Set.大部分的应用程序不需要其他类型的数据结构,但若是真需要也有很多高级数据结构可供 选择,例如Collection.Array.Heapq.Bisect.Weakref.Copy以及Pprint.本文将介绍这些数据结构的用法,看 看它

Python中的高级数据结构详解

这篇文章主要介绍了Python中的高级数据结构详解,本文讲解了Collection.Array.Heapq.Bisect.Weakref.Copy以及Pprint这些数据结构的用法,需要的朋友可以参考下 数据结构 数据结构的概念很好理解,就是用来将数据组织在一起的结构.换句话说,数据结构是用来存储一系列关联数据的东西.在Python中有四种内建的数据结构,分别是List.Tuple.Dictionary以及Set.大部分的应用程序不需要其他类型的数据结构,但若是真需要也有很多高级数据结构可供选择

Python中的高级数据结构

数据结构 数据结构的概念很好理解,就是用来将数据组织在一起的结构.换句话说,数据结构是用来存储一系列关联数据的东西.在Python中有四种内建的数据结构,分别是List.Tuple.Dictionary以及Set.大部分的应用程序不需要其他类型的数据结构,但若是真需要也有很多高级数据结构可供选择,例如Collection.Array.Heapq.Bisect.Weakref.Copy以及Pprint.本文将介绍这些数据结构的用法,看看它们是如何帮助我们的应用程序的. 关于四种内建数据结构的使用方

XPath在python中的高级应用

XPath在python的爬虫学习中,起着举足轻重的地位,对比正则表达式 re两者可以完成同样的工作,实现的功能也差不多,但XPath明显比re具有优势,在网页分析上使re退居二线. XPath介绍: 是什么? 全称为XML Path Language 一种小型的查询语言 说道XPath是门语言,不得不说它所具备的优点: 1) 可在XML中查找信息 2) 支持HTML的查找 3) 通过元素和属性进行导航 python开发使用XPath条件: 由于XPath属于lxml库模块,所以首先要安装库lx

python 中的高级函数sorted()

Python内置的 sorted()函数可对list进行排序: >>>sorted([36, 5, 12, 9, 21]) [5, 9, 12, 21, 36] 但 sorted()也是一个高阶函数,它可以接收一个比较函数来实现自定义排序,比较函数的定义是,传入两个待比较的元素 x, y,如果 x 应该排在 y 的前面,返回 -1,如果 x 应该排在 y 的后面,返回 1.如果 x 和 y 相等,返回 0. 因此,如果我们要实现倒序排序,只需要编写一个reversed_cmp函数: de