python 进程间通信 消费者生产者模型 队列

Posted 2020-12-31 北寒-

队列    from multiprocessing import Queue

 队列： 先进先出（First In First Out   简称 FIFO）///  栈：先进后出（First In Last Out       简称 FILO）

Queue 方法介绍 :

 

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

 

其他方法:

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

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

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

from multiprocessing import Queue

q = Queue(4)
q.put(1)
q.put(2)
q.put(3)
q.put(4)
# q.put(5)  #如果是put，因为已经放满数据，所以程序会阻塞在put，等待取出数据
# q.put_nowait(5)   # 如果是put_nowait() ，不会阻塞，直接放入队列数据，队列满则报异常
# try :
#     q.put_nowait(5)  # 在此处用try直接处理异常。此时数据不会放入到队列，会被直接丢弃
# except :
#     print(‘队列满了‘)

print(q.get())
print(q.get())
print(q.get())
print(q.get())
# print(q.get())  # 此处和上边一样，因为队列中已空，所以程序会阻塞在get，等待放入数据
# print(q.get_nowait())  # 不会阻塞，直接从队列中获取数据，获取不到则报错
try :
    q.get_nowait()
except :
    print(‘队列已空‘)

进程间通信的队列

from multiprocessing import Process, Queue
import time


def func(q):
    time.sleep(1)
    q.put(‘我是中国人‘)


if __name__ == ‘__main__‘:
    q = Queue(5)
    p = Process(target=func,args=(q,))
    p.start()  #子进程和父进程同时运行,不一定队列中有数据
    print(q.get()) # 一定不会报错,get()是阻塞获取数据如果队列没有就等着
    # print(q.get_nowait()) #数据为空是获取不到会报错

from multiprocessing import Process,Queue
import time,os
# import random
def p_func(q):
    q.put(os.getpid())

def g_func(q):
    print(q.get())
if __name__ == ‘__main__‘:
    q = Queue(5)
    l_put = []
    l_get = []
    for i in range(10):
        p_put = Process(target=p_func,args=(q,))
        p_put.start()
        l_put.append(p_put)

    for i in range(10):
        p_get = Process(target=g_func, args=(q,))
        p_get.start()
        l_get.append(p_get)

    [i.join() for i in l_get]
    [i.join() for i in l_put]

生产者消费者模型

主要是为解耦
借助队列来实现生产者消费者模型