1、基于普通队列的生产者和消费者模型:
from multiprocessing import Process, Queue
import time
import os
import random
"""
主进程永远不会结束,原因是:生产者p在生产完后就结束了,但是消费者c在取空了q之后,则一直处于死循环中且卡在q.get()这一步。
解决方式无非是让生产者在生产完毕后,往队列中再发一个结束信号,这样消费者在接收到结束信号后就可以break出死循环。
"""
def produce(q):
for i in range(10):
q.put("%s在生产%s" % (os.getpid(), i))
time.sleep(random.randint(1,3))
print("%s在生产%s" % (os.getpid(), i))
# time.sleep(0.1)
# q.put(None)
def consumer(q):
while True:
time.sleep(random.randint(1, 3))
res = q.get()
print("%s在消费 %s" % (os.getpid(), res))
if res == None:
break
if __name__ == "__main__":
q = Queue()
p = Process(target=produce, args=(q,))
p1 = Process(target=produce, args=(q,))
p2 = Process(target=consumer, args=(q,))
p3 = Process(target=consumer, args=(q,))
p.start()
p1.start()
p2.start()
p3.start()
p.join() #必须保证生产者全部生产完毕,才应该发送结束信号
p1.join()
q.put(None) #发送结束信号
q.put(None)
2、基于JoinableQueue队列的生产者消费者模型:
from multiprocessing import Process, JoinableQueue
import time
import random
import os
"""
流程:
主进程先设置p.join()等待生产者结束 ----》生产者设置q.join()等待队列中的数据全部被取走 -----》消费者设置q.task_done()每取一次
数据向q.join() 发送一次信号直到数据全部被取走------》全部取走后q.join()失效生产者结束-----》生产者结束后主进程p.join()
失效主进程结束----》c1.daemon设置消费者守护进程主进程一结束消费者就结束。
"""
def produce(q):
for i in range(10):
q.put("%s在生产%s" % (os.getpid(), i))
time.sleep(random.randint(1,3))
print("%s在生产%s" % (os.getpid(), i))
q.join() #全部生产完之后设置堵塞,等待队列里的数据全部被取出
def consumer(q):
while True:
time.sleep(random.randint(1, 3))
res = q.get()
print("%s在消费 %s" % (os.getpid(), res))
q.task_done() #向q.join()发送一次信号,证明一个数据已经被取走了
if __name__ == "__main__":
q = JoinableQueue()
p = Process(target=produce, args=(q,))
p1 = Process(target=produce, args=(q,))
c1 = Process(target=consumer, args=(q,))
c2 = Process(target=consumer, args=(q,))
c1.daemon = True #设置p1为守护进程主进程一结束c1结束
c2.daemon = True
p.start()
p1.start()
c1.start()
c2.start()
p.join() #主进程在这堵塞防止一开始主进程就结束带动c1结束,主进程等待p进程结束
p1.join()