Python程序中的进程操作-进程间通信(multiprocess.Queue)

目录

• 一、进程间通讯
• 二、队列
  • 2.1 概念介绍---multiprocessing.Queue
  • 2.2 方法介绍
  • 2.3 其他方法（了解）
• 三、Queue队列-实例演示
  • 3.1 但看队列用法
  • 3.2 子进程发送数据给父进程
• 四、生产者消费者模型
  • 4.1 为什么要使用生产者和消费者模式
  • 4.2 什么是生产者消费者模式
  • 4.3 基于队列实现生产者消费者模型
• 五、 JoinableQueue队列
  • 5.1 JoinableQueue的使用
  • 5.2 通过JoinableQueue队列实现生产者消费者模型

一、进程间通讯

IPC。就是多个进程间相互通讯。进程间通讯并不只是单个语言的问题。而是每个语言都有的。实现进程间通讯有很多。C语言中有：管道：pipe（匿名管道），有名管道、信号、共享内存、消息队列、信号量等等

Python中这里通过队列实现进程间通讯

二、队列

2.1 概念介绍---multiprocessing.Queue

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

Queue([maxsize])创建共享的进程队列。
参数 ：maxsize是队列中允许的最大项数。如果省略此参数，则无大小限制。

底层队列使用管道和锁定实现。

2.2 方法介绍

Queue([maxsize])：创建共享的进程队列。maxsize是队列中允许的最大项数。如果省略此参数，则无大小限制。底层队列使用管道和锁定实现。另外，还需要运行支持进程以便队列中的数据传输到底层管道中。
Queue的实例q具有以下方法：

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()方法）。

2.3 其他方法（了解）

q.close() ：关闭队列，防止队列中加入更多数据。调用此方法时，后台进程将继续写入那些已入队列但尚未写入的数据，但将在此方法完成时马上关闭。如果q被垃圾收集，将自动调用此方法。关闭队列不会在队列使用者中生成任何类型的数据结束信号或异常。例如，如果某个使用者正被阻塞在get()操作上，关闭生产者中的队列不会导致get()方法返回错误。

q.cancel_join_thread() ：不会再进程退出时自动连接后台进程。这可以防止join_thread()方法阻塞。

q.join_thread() ：连接队列的后台进程。此方法用于在调用q.close()方法后，等待所有队列项被消耗。默认情况下，此方法由不是q的原始创建者的所有进程调用。调用q.cancel_join_thread()方法可以禁止这种行为。

三、Queue队列-实例演示

3.1 但看队列用法

'''
ipc 进程间通讯。使用队列实现
管道:pipe 基于共享的内存空间
队列:pipe+锁 Queue
put：放（可以设置阻塞非阻塞，和等待时间）
get：取（可以设置阻塞非阻塞，和等待时间）
'''
### 语法
q = Queue(3) # 创建队列, 可以设置最大值
## put 放
q.put('你好')     # 往队列中放入值，可以设置阻塞和等待时间，默认：满了再放就会阻塞等待
q.put([1,2,4])
q.put(2)

### 不设置等待时间，队列满了继续放
# q.put(5)    # 阻塞，队列满了，等待队列空了就放进去

## get 拿
print(q.get())  # 获取队列中的内容，可以设置阻塞和等待时间，默认：拿不到内容就会阻塞等待
print(q.get())
print(q.get())

### 不设置等待时间
# print(q.get())  # 阻塞，一直等待获取队列内容

### 设置等待时间
# print(q.get(timeout=2))     # 等两秒，等不到就报错
# q.put(88,timeout=2)   # 不会报错，因为队列中没内容


### 全都设置非阻塞模式
q = Queue(3) # 创建队列, 可以设置最大值
q.put('你好', block=False)     # 设置非阻塞，如果满了再放就会报错
q.put([1,2,4], block=False)
q.put(2, block=False)
# q.put(4, block=False)       # 报错，队列满了

# q.put_nowait('666') # 等同 block = False，报错，队列满了

3.2 子进程发送数据给父进程

import time
from multiprocessing import Process, Queue

def f(q):
    q.put([time.asctime(), 'from Eva', 'hello'])  #调用主函数中p进程传递过来的进程参数 put函数为向队列中添加一条数据。

if __name__ == '__main__':
    q = Queue() #创建一个Queue对象
    p = Process(target=f, args=(q,)) #创建一个进程
    p.start()
    print(q.get())
    p.join()

上面是一个queue的简单应用，使用队列q对象调用get函数来取得队列中最先进入的数据。 接下来看一个稍微复杂一些的例子：批量生产数据放入队列再批量获取结果。

四、生产者消费者模型

在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产进程和消费进程的工作能力来提高程序的整体处理数据的速度。

4.1 为什么要使用生产者和消费者模式

在进程世界里，生产者就是生产数据的进程，消费者就是消费数据的进程。在多进程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。

4.2 什么是生产者消费者模式

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，消费者不找生产者要数据，而是直接从阻塞队列里取，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

4.3 基于队列实现生产者消费者模型

ef producer(q,name,food):
    '''生产者'''
    for i in range(10):
        print(f'{name}生产了{food}{i}')
        res = f'{food}{i}'
        q.put(res)


def consumer(q,name):
    '''消费者'''
    while 1:
        res = q.get()
        if res == None:
            break
        print(f'{name}买了{res}')


if __name__ == '__main__':
    q = Queue()   # 使用队列
    p1 = Process(target=producer,args=(q,"xc","意大利面"))
    c1 = Process(target=consumer,args=(q,"haha"))
    p2 = Process(target=producer, args=(q, "xc", "牛肉"))
    c2 = Process(target=consumer, args=(q, "xixi"))
    p3 = Process(target=producer, args=(q, "xc", "可乐"))
    p1.start()
    c1.start()
    p2.start()
    c2.start()
    p3.start()

    p1.join()   # 等待生产者结束
    p2.join()   # 等待生产者结束
    p3.join()   # 等待生产者结束

    q.put(None) # 有几个消费者就要发几个None，让 子进程收到None就结束
    q.put(None) # 有几个消费者就要发几个None，让 子进程收到None就结束

结束信号None，不一定要由生产者发，主进程里同样可以发，但主进程需要等生产者结束后才应该发送该信号。 有几个消费者就要发几个None，让 子进程收到None就结束

五、 JoinableQueue队列

JoinableQueue队列就对Queue队列的改良版，加入了类似信号量的机制。

5.1 JoinableQueue的使用

语法：

1. 实例化对象：q = JoinableQueue()
2. 向队列中放入内容，相当于 信号量+1 操作: q.put(xxx)
3. 从队列中取出内容：q.get()
4. 任务结束，相当于 信号量-1 操作:q.task_done()
5. 当这个 信号量不为0时，会阻塞等待，计数器为0后通过。

实例：

# JoinableQueue的使用
q = JoinableQueue()

q.put('1')  # +1
q.put('2')  # +1

print(q.get())
q.task_done()   # -1	
print(q.get())
q.task_done()  # -1
q.join() #计数器不为0会阻塞等待 计数器为0后通过

5.2 通过JoinableQueue队列实现生产者消费者模型

def producer(q, name, food):
    '''生产者'''
    for i in range(10):
        print(f'{name}生产了{food}{i}')
        res = f'{food}{i}'
        q.put(res)  # +1


def consumer(q, name):
    '''消费者'''
    while 1:
        res = q.get()
        if res == None:
            break
        print(f'{name}买了{res}')
        q.task_done()


if __name__ == '__main__':
    q = JoinableQueue()     # 使用加了join的队列
    p1 = Process(target=producer, args=(q, "xc", "意大利面"))
    c1 = Process(target=consumer, args=(q, "haha"))
    p2 = Process(target=producer, args=(q, "xc", "牛肉"))
    c2 = Process(target=consumer, args=(q, "xixi"))
    p3 = Process(target=producer, args=(q, "xc", "可乐"))
    # 把消费者变成守护进程，主进程结束，子进程就结束
    c1.daemon = True    # 进程结束，子进程就结束
    c2.daemon = True    # 进程结束，子进程就结束

    p1.start()
    c1.start()
    p2.start()
    c2.start()
    p3.start()

    p1.join()  # 等待生产者结束
    p2.join()  # 等待生产者结束
    p3.join()  # 等待生产者结束

上述代码中，是通过将子进程变为守护进程，在消费者的子进程中每次收到消息都task_done，然后等待主进程结束，直接结束子进程。