进程间通信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返回的标识,以便可以知道队列中的数据什么时候被消费完了

原文地址:https://www.cnblogs.com/niuli1987/p/9518257.html