消息队列

1. 通信原理：在内存中建立队列模型，进程通过队列将消息存入，或者从队列取出完成进程间通信。

2. 实现方法

   from multiprocessing import Queue

   • q = Queue(maxsize=0)
     #功能: 创建队列对象
     #参数：最多存放消息个数
     #返回值：队列对象

   • q.put(data,[block,timeout])
     功能：向队列存入消息
     参数：data 要存入的内容
     block 设置是否阻塞 False为非阻塞
     timeout 超时检测

   • q.get([block,timeout])

     功能：从队列取出消息
     参数：block 设置是否阻塞 False为非阻塞
     timeout 超时检测
     返回值： 返回获取到的内容

   • q.full() 判断队列是否为满

   • q.empty() 判断队列是否为空

   • q.qsize() 获取队列中消息个数

   • q.close() 关闭队列

管道通信(Pipe)

1. 通信原理：在内存中开辟管道空间，生成管道操作对象，多个进程使用同一个管道对象进行读写即可实现通

2. .实现方法

   from multiprocessing import Pipe

   • fd1,fd2 = Pipe(duplex = True)
     功能: 创建管道
     参数：默认表示双向管道
     如果为False 表示单向管道
     返回值：表示管道两端的读写对象
     　　如果是双向管道均可读写
     　　如果是单向管道fd1只读 fd2只写

   • fd.recv()
     功能 ： 从管道获取内容
     返回值：获取到的数据

   • fd.send(data)

     功能： 向管道写入内容
     参数： 要写入的数据

   • 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异常。

from multiprocessing import Process, Pipe
def f(conn):conn.send([12, {"name":"yuan"}, 'hello'])response=conn.recv()print("response",response)conn.close()print("q_ID2:",id(conn))if __name__ == '__main__':parent_conn, child_conn = Pipe() #双向管道print("q_ID1:",id(child_conn))#将一个管道对象作为参数，传递给另外一个线程，达到通讯的目的p = Process(target=f, args=(child_conn,))p.start()print(parent_conn.recv())   # prints "[42, None, 'hello']"parent_conn.send("儿子你好!")p.join()
#打印
# q_ID1: 47103568
# [12, {'name': 'yuan'}, 'hello']
# response
# 儿子你好!
# q_ID2: 119641040

Manager

由 Manager() 返回的管理器对象控制一个服务器进程，该进程保存Python对象并允许其他进程使用代理操作它们。

Manager() 返回的管理器支持类型： list 、 dict 、 Namespace 、 Lock 、 RLock 、 Semaphore 、 BoundedSemaphore 、 Condition 、 Event 、 Barrier 、 Queue 、 Value 和 Array 。

from multiprocessing import Process, Managerdef f(d, l):d[1] = '1'd['2'] = 2d[0.25] = Nonel.reverse()if __name__ == '__main__':with Manager() as manager:d = manager.dict()l = manager.list(range(10))p = Process(target=f, args=(d, l))p.start()p.join()print(d)print(l)
#打印
#{0.25: None, 1: '1', '2': 2}
#[9, 8, 7, 6, 5, 4, 3, 2, 1, 0]