码迷,mamicode.com
首页 > 其他好文 > 详细

事件Event实现消费者模型

时间:2019-01-21 01:09:14      阅读:180      评论:0      收藏:0      [点我收藏+]

标签:实现   proc   执行   方法   生产   ble   get   进程   完成   

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()方法停止阻塞
主进程代码执行完毕
守护进程(消费者的进程)结束

‘‘‘










事件Event实现消费者模型

标签:实现   proc   执行   方法   生产   ble   get   进程   完成   

原文地址:https://www.cnblogs.com/bozhengheng/p/10296805.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!