进程间通信（管道）

一、关于创建管道的类

（1）创建管道的类

• Pipe([duplex])
  • 在进程之间创建一条管道，并返回元组（conn1,conn2）,其中conn1，conn2表示管道两端的连接对象
  • 强调一点：必须在产生Process对象之前产生管道
  • duplex
    • 默认管道是全双工的，如果将duplex射成False，conn1只能用于接收，conn2只能用于发送。

（2）主要方法

• conn1.recv()
  • 接收conn2.send(obj)发送的对象
  • 如果没有消息可以接收recv就会一直处于阻塞状态
  • 如果连接的另一端已经关闭，那么recv方法就会抛出EOFError异常、
• conn1.senf(obj)
  • 通过连接发送对象。obj是与序列化兼容的任意对象

（3）次要方法

• 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异常
  • 基于管道实现进程间通信（与队列的方式是类似的，队列就是管道加锁实现的）

三、代码实现

[1]基于管道实现进程间通信

from multiprocessing import Process, Pipe


def producer(p_conn, name):
    left_conn, right_conn = p_conn
    right_conn.close()
    for i in range(5):
        data = f'{name}大厨烹饪的美食{i + 1}'
        left_conn.send(data)
    left_conn.close()


def customer(p_conn):
    left_conn, right_conn = p_conn
    left_conn.close()
    while True:
        try:
            data = right_conn.recv()
            print(f'消费者品尝了{data}')
        except EOFError:
            print('品尝结束！')
            right_conn.close()
            break


def main():
    # 创建管道必须在创建子进程之前
    left_conn, right_conn = Pipe()
    p_list = []
    # 创建消费者子进程
    customers = Process(target=customer, args=((left_conn, right_conn),))
    # 创建生产者子进程
    xanadu = Process(target=producer, args=((left_conn, right_conn), 'Xanadu',))
    p_list.append(xanadu)
    bridge = Process(target=producer, args=((left_conn, right_conn), 'bridge',))
    p_list.append(bridge)
    p_list.append(customers)
    p_list_new = []
    for p in p_list:
        p.start()
        p_list_new.append(p)
    for p in p_list_new:
        p.join()

    print('这是主进程！')


if __name__ == '__main__':
    main()
    
    
"""
消费者品尝了Xanadu大厨烹饪的美食1
消费者品尝了Xanadu大厨烹饪的美食2
消费者品尝了Xanadu大厨烹饪的美食3
消费者品尝了Xanadu大厨烹饪的美食4
消费者品尝了Xanadu大厨烹饪的美食5
消费者品尝了bridge大厨烹饪的美食1
消费者品尝了bridge大厨烹饪的美食2
消费者品尝了bridge大厨烹饪的美食3
消费者品尝了bridge大厨烹饪的美食4
消费者品尝了bridge大厨烹饪的美食5
"""

[2]管道通信特别注意

• 生产者和消费者都没有使用管道的某个端点，就应该将其关闭，
• 如在生产者中关闭管道的右端，在消费者中关闭管道的左端。
• 如果忘记执行这些步骤，程序可能再消费者中的recv()操作上挂起。
• 管道是由操作系统进行引用计数的,必须在所有进程中关闭管道后才能生产EOFError异常。
• 因此在生产者中关闭管道不会有任何效果，付费消费者中也关闭了相同的管道端点。