1.1 队列实现进程间通信

1.1.1 Queue常用类

1.1.2 Queue常用实例q常用方法

1.1.3 JoinableQueue

1.1.4 JoinableQueue 实例除Queue实例方法处的方法

1.1.5 JoinableQueue实现进程间通信示例

import multiprocessing


def consumer(input_q): 
    """ 消费者
    """
    while True:
        item = input_q.get()
        # 处理项
        print(f'处理input_q: {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
    cons_p.start()

    # 运行生成者进程
    sequence = (1,2,3,4)
    producer(sequence, q)

    # 等待所有项被处理完成
    q.join()

1.1.6 JoinableQueue实现多个消费都进程池间通信示例

from msilib import sequence
import multiprocessing


def consumer(input_q): 
    """ 消费者
    """
    while True:
        item = input_q.get()
        # 处理项
        print(f'处理input_q: {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_1 = multiprocessing.Process(target=consumer, args=(q, ))
    # 进程1处理项
    cons_p_1.daemon = True
    cons_p_1.start()

    cons_p_2 = multiprocessing.Process(target=consumer, args=(q, ))
    # 进程2处理项
    cons_p_2.daemon = True
    cons_p_2.start()

    # 运行生产者
    sequence = (1,2,3,4)
    producer(sequence, q)

    # 等待所有项被处理
    q.join()

1.1.7 哨兵模式实现多进程间通信示例

from ast import arg
from audioop import mul
from msilib import sequence
import multiprocessing
import sqlite3
from tkinter.messagebox import NO


def consumer(input_q):
    while True:
        item = input_q.get()
        # 处理哨兵
        if item is None:
            break
        # 处理项
        print(f'处理input_q: {item}')
    # 关闭
    print('consumer 处理完成！')


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

注意： 如果使用哨兵，一定要在队列上为每个消费都都安置哨兵，才能保证所有消费者进程都能关闭。

1.2 管道实现进程间通信

1.2.1 管道类

1.2.2 Pipe()返回的Connection对象实例c具有的方法和属性

1.2.3 管道实现生产者消费都模型示例

import multiprocessing


def consumer(pipe):
    output_p, input_p = pipe
    input_p.close()  # 关闭管道的输入端
    while True:
        try:
            item = output_p.recv()
        except EOFError:
            break
        # TODO 可替换为有用的工作
        print(item)
    # 关闭
    print('consumer done.')


def producer(sequence, input_p):
    for item in sequence:
        input_p.send(item)


if __name__ == '__main__':
    # 创建管道
    (output_p, input_p) = multiprocessing.Pipe()

    # 启动消费者进程
    cons_p = multiprocessing.Process(target=consumer, args=((output_p, input_p),))
    cons_p.start()

    # 关闭生产者中的输出管道
    output_p.close()

    # 生产项
    sequence = (1,2,3,4)
    producer(sequence, input_p)

    # 半闭输入管道，表示输入完成
    input_p.close()

    # 等待消费者进程结束
    cons_p.join()

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

1.2.4 管道双向通信示例

import multiprocessing
from unittest import result


def adder(pipe):
    server_p, client_p = pipe
    client_p.close()
    while True:
        try:
            x,y = server_p.recv()
        except EOFError:
            break
        result = x + y  # TODO 可用实例业务代替
        server_p.send(result)
    # 关闭
    print('Server done.')


if __name__ == '__main__':
    # 生成管道
    (server_p, client_p) = multiprocessing.Pipe()
    
    # 启动服务器进程
    adder_p = multiprocessing.Process(target=adder, args=((server_p, client_p),))
    adder_p.start()
    
    # 关闭客户端中的服务器管道
    server_p.close()
    
    # 客户端发出请求
    client_p.send((3,4))
    print(client_p.recv())

    client_p.send(('Hello', 'World'))
    print(client_p.recv())

    # 完成，关闭管道
    client_p.close()

    # 等待消费者进程完成
    adder_p.join()