事件Event实现消费者模型

import time
import random
from multiprocessing import Process,Queue def consumer(q,name):
while True:
print('\033[31m%s吃了%s\033[0m'%(name,q.get()))  # 如果队列中没有数据了,q.get()方法会一直阻塞
time.sleep(random.randint(1, 3)) def producer(q,food,name): for i in range(10):
time.sleep(random.randint(1, 3))
f = '%s 生产的 %s%i' % (name, food, i)
print(f)
q.put(f) if __name__ == '__main__':
q = Queue(20)
p = Process(target=producer,args=(q,'包子','Egon'))
p1 = Process(target=producer, args=(q, '泔水', 'jinboss'))
c1 = Process(target=consumer, args=(q, 'alex'))
c2 = Process(target=consumer, args=(q, '娜扎'))
p.start()
p1.start()
c1.start()
c2.start()
  q.put(None)
  q.put(None)
这样是有问题的,因为对列中没有了数据,但是consumer函数还在一个q.get()取数据,但是producter已经生产完毕,所以或一值阻塞,
解决问题的方法:
在末尾添加 q.put(None) 然后在consumer中的取得的数据进行判断,为None就break。
但是这样很麻烦,有多少个consumer就要放几个None 使用JoinableQueue,可以解决这个问题
import time
import random
from multiprocessing import Process,JoinableQueue def consumer(name,q):
while True:
f = q.get()
print('\033[31m%s 吃了 %s\033[0m'%(name,f))
time.sleep(random.randint(1, 3))
q.task_done() # 每一次取得一个数据,处理完成后,返回的一个标志,就相当于一个计数器,每次-1 def producter(name,food,q):
for i in range(10):
f = '%s 生产了 %s%i'%(name,food,i)
print('%s 生产了 %s%i'%(name,food,i))
q.put(f)
time.sleep(random.randint(1,3))
q.join() # 阻塞 直到队列中的数据都被处理完成后,producter进程才结束
# 一般是代码执行完毕就结束,q.join(),感知队列数据的状态,处理完成后,结束
if __name__ == '__main__':
q = JoinableQueue() p1 = Process(target=producter,args=('Egon','包子',q))
p2 = Process(target=producter,args=('nazhe','包子',q))
c1 = Process(target=consumer,args=('alex',q))
c2 = Process(target=consumer,args=('jinboss',q))
p1.start()
p2.start()
c1.daemon = True  # 守护进程,随着主进程代码的执行完毕后,才结束
c2.daemon = True
c1.start()
c2.start()
p1.join()      # 感知一个子进程的结束,不结束就一直等着
p2.join()
'''
在消费者这一端:
每次处理一个数据
处理一个数据
发送一个记号:标志一个数据被处理完成
在生产者端:
每次生产一个数据
且每次生产的数据都放入队列中
每放入一个数据,就刻一个记号
当生产者全部生产完毕后
join信号:已经停止生产数据 ,进入阻塞状态
但是进程不会结束,且要等到之前存数据时,课上的记号被消费完,否则就一直阻塞
当数据都被处理完毕时,join阻塞结束 consumer 中把所有的数据消耗完
producter 端的join就会感知到,停止阻塞
所有的producter进程都结束之后
主进程中的p.join()方法停止阻塞
主进程代码执行完毕
守护进程(消费者的进程)结束 '''

上一篇:[C] zlstdint(让VC、TC等编译器自动兼容C99的整数类型)V1.0。支持Turbo C++ 3等DOS下的编译器


下一篇:201521123085 《Java程序设计》第14周学习总结