事件Event实现消费者模型

标签：实现   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