zoukankan      html  css  js  c++  java
  • 进程间通信IPC -- 管道, 队列

    进程间通信--IPC(Inter-Process Communication)

    管道

    from multiprocessing import Pipe
    con1,con2 = Pipe()
    管道是不安全的。
    管道是用于多进程之间通信的一种方式。
    如果在单进程中使用管道,那么就是con1收数据,就是con2发数据。
    如果是con1发数据,就是con2收数据

    如果在多进程中使用管道,那么就必须是父进程使用con1收,子进程就必须使用con2发
    父进程使用con1发,子进程就必须使用con2收
    父进程使用con2收,子进程就必须使用con1发
    父进程使用con2发,子进程就必须使用con1收
    在管道中有一个著名的错误叫做EOFError。是指,父进程中如果关闭了发送端,子进程还继续接收数据,那么就会引发EOFError
    #创建管道的类:
    Pipe([duplex]):在进程之间创建一条管道,并返回元组(conn1,conn2),其中conn1,conn2表示管道两端的连接对象,强调一点:必须在产生Process对象之前产生管道
    #参数介绍:
    dumplex:默认管道是全双工的,即conn1和conn2都是既能收数据也能发数据的,如果将duplex设成False,conn1只能用于接收,conn2只能用于发送。
    #主要方法:
        conn1.recv():接收conn2.send(obj)发送的对象。如果没有消息可接收,recv方法会一直阻塞。如果连接的另外一端已经关闭,那么recv方法会抛出EOFError。
        conn1.send(obj):通过连接发送对象。obj是与序列化兼容的任意对象
     #其他方法:
    conn1.close():关闭连接。如果conn1被垃圾回收,将自动调用此方法
    conn1.fileno():返回连接使用的整数文件描述符
    conn1.poll([timeout]):如果连接上的数据可用,返回True。timeout指定等待的最长时限。如果省略此参数,方法将立即返回结果。如果将timeout设成None,操作将无限期地等待数据到达。
     
    conn1.recv_bytes([maxlength]):接收c.send_bytes()方法发送的一条完整的字节消息。maxlength指定要接收的最大字节数。如果进入的消息,超过了这个最大值,将引发IOError异常,并且在连接上无法进行进一步读取。如果连接的另外一端已经关闭,再也不存在任何数据,将引发EOFError异常。
    conn.send_bytes(buffer [, offset [, size]]):通过连接发送字节数据缓冲区,buffer是支持缓冲区接口的任意对象,offset是缓冲区中的字节偏移量,而size是要发送字节数。结果数据以单条消息的形式发出,然后调用c.recv_bytes()函数进行接收    
     
    conn1.recv_bytes_into(buffer [, offset]):接收一条完整的字节消息,并把它保存在buffer对象中,该对象支持可写入的缓冲区接口(即bytearray对象或类似的对象)。offset指定缓冲区中放置消息处的字节位移。返回值是收到的字节数。如果消息长度大于可用的缓冲区空间,将引发BufferTooShort异常。
    
    介绍
    
    管道的介绍
    管道介绍

    应该特别注意管道端点的正确管理问题。如果是生产者或消费者中都没有使用管道的某个端点,就应将它关闭

    这也说明了为何在生产者中关闭了管道的输出端,在消费者中关闭管道的输入端

    如果忘记执行这些步骤,程序可能在消费者中的recv()操作上挂起。

    管道是由操作系统进行引用计数的,必须在所有进程中关闭管道后才能生成EOFError异常。

    因此,在生产者中关闭管道不会有任何效果,除非消费者也关闭了相同的管道端点。

    '''单进程下的管道'''
    '''
    from multiprocessing import Pipe
    
    c1,c2 = Pipe()
    
    c1.send('abc')
    print(c2.recv())
    c2.send(111)
    print(c1.recv())
    '''
    
    '''多进程下的管道'''
    
    from multiprocessing import Pipe, Process
    
    
    def func(c):
        c = c1, c2
        c1.close()  # 子进程使用c2和父进程通信
        while 1:
            try:
                print(c2.recv())#当主进程的con1发数据时,子进程要死循环的去接收。
            except EOFError:  # 如果主进程的con1发完数据并关闭con1,子进程的con2继续接收时,就会报错,使用try的方式,获取错误
                c2.close() # 获取到错误,就是指子进程已经把管道中所有数据都接收完了,所以用这种方式去关闭管道
                break
    
    
    if __name__ == '__main__':
        c1, c2 = Pipe()  # 建立管道
        p = Process(target=func, args=((c1, c2),))  # 建立进程
        p.start()  # 开始进程
        c2.close()  # 父进程中,使用c1和子进程通信不需要c2 就提前关闭
        for i in range(10):  # 创建数据
            c1.send(i)  # 发送数据
        c1.close()  # 生产完数据,关闭父进程这一段管道
    管道例子

    队列

    概念:创建共享的进程队列,Queue是多进程安全的队列,可以使用Queue实现多进程之间的数据传递。

    Queue([maxsize]) 
    创建共享的进程队列。
    参数 :maxsize是队列中允许存在的最大元素个数。如果省略此参数,则无大小限制。
    底层队列使用管道和锁定实现。
    Queue([maxsize]) 
    
    q = Queue([maxsize])
    
    q.get( [ block [ ,timeout ] ] ) 
    返回q中的一个项目。如果q为空,此方法将阻塞,直到队列中有项目可用为止。block用于控制阻塞行为,默认为True. 如果设置为False,将引发Queue.Empty异常(定义在Queue模块中)。timeout是可选超时时间,用在阻塞模式中。如果在指定的时间间隔内没有项目可用,将引发Queue.Empty异常。
    
    q.get_nowait( ) 
    同q.get(False)方法。
    
    q.put(item [, block [,timeout ] ] ) 
    将item放入队列。如果队列已满,此方法将阻塞至有空间可用为止。block控制阻塞行为,默认为True。如果设置为False,将引发Queue.Empty异常(定义在Queue库模块中)。timeout指定在阻塞模式中等待可用空间的时间长短。超时后将引发Queue.Full异常。
    
    q.qsize() 
    返回队列中目前项目的正确数量。此函数的结果并不可靠,因为在返回结果和在稍后程序中使用结果之间,队列中可能添加或删除了项目。在某些系统上,此方法可能引发NotImplementedError异常。
    
    
    q.empty() 
    如果调用此方法时 q为空,返回True。如果其他进程或线程正在往队列中添加项目,结果是不可靠的。也就是说,在返回和使用结果之间,队列中可能已经加入新的项目。
    
    q.full() 
    如果q已满,返回为True. 由于线程的存在,结果也可能是不可靠的(参考q.empty()方法)。。
    
    方法介绍
    
    Queue中的方法介绍
    用法
    q.close() 
    关闭队列,防止队列中加入更多数据。调用此方法时,后台线程将继续写入那些已入队列但尚未写入的数据,但将在此方法完成时马上关闭。如果q被垃圾收集,将自动调用此方法。关闭队列不会在队列使用者中生成任何类型的数据结束信号或异常。例如,如果某个使用者正被阻塞在get()操作上,关闭生产者中的队列不会导致get()方法返回错误。
    
    q.cancel_join_thread() 
    不会再进程退出时自动连接后台线程。这可以防止join_thread()方法阻塞。
    
    q.join_thread() 
    连接队列的后台线程。此方法用于在调用q.close()方法后,等待所有队列项被消耗。默认情况下,此方法由不是q的原始创建者的所有进程调用。调用q.cancel_join_thread()方法可以禁止这种行为。
    
    其他方法(了解)
    需要你了解的几个方法

    from multiprocessing import JoinableQueue
    task_done() 每消费一个数据,就返回一个标识
    join() 接收task_done返回的标识,以便可以知道队列中的数据什么时候被消费完了

  • 相关阅读:
    poj3278 Catch That Cow
    poj2251 Dungeon Master
    poj1321 棋盘问题
    poj3083 Children of the Candy Cor
    jvm基础知识—垃圾回收机制
    jvm基础知识1
    java面试基础必备
    java soket通信总结 bio nio aio的区别和总结
    java scoket aIO 通信
    java scoket Blocking 阻塞IO socket通信四
  • 原文地址:https://www.cnblogs.com/niuli1987/p/9518257.html
Copyright © 2011-2022 走看看