Python多进程队列(Queue)和生产者消费者模型解读

from multiprocessing import Process,Queue

q = Queue([maxsize])  # 创建一个队列,maxsize表示队列的大小，即可以存放几条数据，如果省略则可以存放无穷多条数据，取决于你内存空间的大小

q.put(data,block=True,timeout=3) 
# 往队列里面存数据data
# 如果block为True（默认值），并且timeout为正值，该方法会阻塞timeout指定的时间，直到该队列有剩余的空间。如果超时，会抛出Queue.Full异常。如果block为False，但该Queue已满，会立即抛出Queue.Full异常。

q.get(block=True,timeout=3) 
# 从队列里面去取数据，该方法了put方法一样，也是有block和timeout2个参数。用法同参建put方法。

q.get_nowait():# 等同于q.get(block=False)
q.put_nowait():# 等同于q.put(block=False)

q.empty() # 判断队列是否为空
q.full() # 判断队列是否已经满了
q.qsize() # 返回队列中存放数据的个数
# 由于多进程是并发执行的，随时都有可能向队列里面存放数据，也可能葱队列里面去取数据。故这个结果并不是很可靠。

q.cancel_join_thread() # 不会在进程退出时自动连接后台线程。可以防止join_thread()方法阻塞
q.close() # 关闭队列，防止队列中加入更多数据。调用此方法，后台线程将继续写入那些已经入队列但尚未写入的数据，但将在此方法完成时马上关闭。如果q被垃圾收集，将调用此方法。关闭队列不会在队列使用者中产生任何类型的数据结束信号或异常。例如，如果某个使用者正在被阻塞在get()操作上，关闭生产者中的队列不会导致get()方法返回错误。
q.join_thread() # 连接队列的后台线程。此方法用于在调用q.close()方法之后，等待所有队列项被消耗。默认情况下，此方法由不是q的原始创建者的所有进程调用。调用q.cancel_join_thread方法可以禁止这种行为

from multiprocessing import Process,Queue
import time,random,os

def get_data(q,name):
    while True:
        data=q.get()
        if data is None: # 获取结束信号，跳出死循环，不然等数据取玩了会一直卡在此处
            break
        time.sleep(random.randint(1,3))
        print('%s 取出了 %s' %(name,data))

def create_data(q,name):
    for i in range(5):
        time.sleep(random.randint(1,3))
        data='数据%s' %i
        q.put(data)
        print('%s 生产了 %s' %(name,data))

    q.put(None) # 结束信号，不一定是None可以自己定义，就是告诉消费者已经停止生产数据了

if __name__ == '__main__':
    q=Queue()
    p1=Process(target=create_data,args=(q,"生产者")) #生产者
    p1.start()
    c1=Process(target=get_data,args=(q,"消费者")) #消费者
    c1.start()

生产者 生产了 数据0
消费者 取出了 数据0
生产者 生产了 数据1
生产者 生产了 数据2
消费者 取出了 数据1
生产者 生产了 数据3
消费者 取出了 数据2
生产者 生产了 数据4
消费者 取出了 数据3
消费者 取出了 数据4

# JoinableQueue除了拥有和Queue一样的方法外，还有2个用来在生产者和消费者之间通信的方法
q.task_done() # 在消费者q.get()取出数据并处理后调用此方法告诉生产者，数据已经被处理了。每取出一个数据，就需要通过此方法发送一次信号给生产者，否则会抛出ValueError异常
q.join() # 生产者调用此方法进行阻塞，直到队列中所有的数据都被处理

from multiprocessing import Process,JoinableQueue
import time,random

def get_data(q,name):
    while True:
        data=q.get()
        if data is None: # 获取结束信号，跳出死循环，不然等数据取玩了会一直卡在此处
            break
        time.sleep(random.randint(1,5))
        print('%s 取出了 %s' %(name,data))
        q.task_done()

def create_data(q,name):
    for i in range(5):
        time.sleep(random.randint(1,5))
        data='数据%s' %i
        q.put(data)
        print('%s 生产了 %s' %(name,data))
    q.join() # 让生产者在此处阻塞，等待消费者处理完所有数据。

if __name__ == '__main__':
    q=JoinableQueue()
    p1=Process(target=create_data,args=(q,"生产者")) #生产者
    p1.start()
    c1=Process(target=get_data,args=(q,"消费者")) #消费者
    c1.daemon = True # 将消费者设置为守护进程，这样主进程结束后，该进程也会跟着结束
    c1.start()
    p1.join() # 此处让主进程等待生产者