快捷導(dǎo)航

Python生產(chǎn)者與消費者模型中的優(yōu)勢介紹

更新時間：2023年03月25日 11:14:20 作者：丘山Ivan

這篇文章主要介紹了python多進(jìn)程中的生產(chǎn)者和消費者模型優(yōu)勢，生產(chǎn)者是指生產(chǎn)數(shù)據(jù)的任務(wù)，消費者是指消費數(shù)據(jù)的任務(wù)。當(dāng)生產(chǎn)者的生產(chǎn)能力遠(yuǎn)大于消費者的消費能力，生產(chǎn)者就需要等消費者消費完才能繼續(xù)生產(chǎn)新的數(shù)據(jù)

生產(chǎn)者消費者模型具體來講，就是在一個系統(tǒng)中，存在生產(chǎn)者和消費者兩種角色，他們通過內(nèi)存緩沖區(qū)進(jìn)行通信，生產(chǎn)者生產(chǎn)消費者需要的資料，消費者把資料做成產(chǎn)品，從而消耗掉生產(chǎn)的數(shù)據(jù)。達(dá)到供需平衡，不能生產(chǎn)多了浪費，也不能需要消耗資源的時候沒有。

multiprocessing-Queue實現(xiàn)

from  multiprocessing import Process,Queue  #多進(jìn)程組件,隊列
import time,random  
#生產(chǎn)者方法
def producer(name,food,q):
    for i in  range(4):
        time.sleep(random.randint(1,3)) #模擬獲取數(shù)據(jù)時間
        f = '%s生產(chǎn)的%s%s'%(name,food,i)
        print(f)
        q.put(f) #添加進(jìn)隊列
#消費者方法
def consumer(q,name):
    while True:
        food = q.get() #如果獲取不到，會一直阻塞進(jìn)程不會結(jié)束子進(jìn)程
        # 當(dāng)隊列中的數(shù)據(jù)是None的時候結(jié)束while循環(huán)
        if food is None:
            print('%s獲取到一個空'%name)
            break
        f = '\033[31m%s消費了%s\033[0m' % (name, food)
        print(f)
        time.sleep(random.randint(1,3)) # 模擬消耗數(shù)據(jù)時間
if __name__ == '__main__':
    q = Queue()  # 創(chuàng)建隊列
    # 模擬生產(chǎn)者 生產(chǎn)數(shù)據(jù)
    p = Process(target=producer, args=('p', '包子', q)) #創(chuàng)建進(jìn)程
    p.start() #啟動進(jìn)程
    p1 = Process(target=producer, args=('p1', '燒餅', q))
    p1.start()
   #模擬消費者消費數(shù)據(jù)
    c = Process(target=consumer, args=(q, 'c'))
    c.start()
    c1 = Process(target=consumer, args=(q, 'c1'))
    c1.start()
    p.join()#阻塞主進(jìn)程 直到p和p1 子進(jìn)程結(jié)束后才執(zhí)行q.put() 方法
    p1.join()#阻塞主進(jìn)程 直到p和p1 子進(jìn)程結(jié)束后才執(zhí)行q.put() 方法
    #為了確保生產(chǎn)者生產(chǎn)完所有數(shù)據(jù)后，
    #最后一個是None,方便結(jié)束子進(jìn)程中的while循環(huán),
    #否則會一直等待隊列中加入新數(shù)據(jù)。
    q.put(None)
    q.put(None)

使用Queue組件實現(xiàn)的缺點就是，實現(xiàn)了多少個消費者consumer進(jìn)程，就需要在最后往隊列中添加多少個None標(biāo)識，方便生產(chǎn)完畢結(jié)束消費者consumer進(jìn)程。否則，p.get() 不到任務(wù)會阻塞子進(jìn)程，因為while循環(huán)，直到隊列q中有新的任務(wù)加進(jìn)來，才會再次執(zhí)行。而我們的生產(chǎn)者只能生產(chǎn)這么多東西，所以相當(dāng)于程序卡死。

multiprocessing-JoinableQueue實現(xiàn)

from multiprocessing import JoinableQueue,Process
import time,random
#生產(chǎn)者方法
def producer(name,food,q):
    for i in range(4):
        time.sleep(random.randint(1, 2))
        f = '%s生產(chǎn)的%s%s'%(name,food,i)
        q.put(f)
        print(f)
    q.join()  #一直阻塞，等待消耗完所有的數(shù)據(jù)后才釋放
#消費者方法
def consumer(name,q):
    while True:
        food = q.get()
        print('\033[31m%s消費了%s\033[0m' % (name, food))
        time.sleep(random.randint(4,8))
        q.task_done() #每次消耗減1
if __name__ == '__main__':
    q = JoinableQueue()  #創(chuàng)建隊列
    #模擬生產(chǎn)者隊列
    p1 = Process(target=producer,args=('p1','包子',q))
    p1.start()
    p2 = Process(target=producer,args=('p2','燒餅',q))
    p2.start()
    #模擬消費者隊列
    c1 = Process(target=consumer,args=('c1',q))
    c1.daemon = True #守護(hù)進(jìn)程：主進(jìn)程結(jié)束，子進(jìn)程也會結(jié)束
    c1.start()
    c2 = Process(target=consumer,args=('c2',q))
    c2.daemon = True
    c2.start()
    p1.join() #阻塞主進(jìn)程，等到p1子進(jìn)程結(jié)束才往下執(zhí)行
    p2.join()
    # q.task_done() 每次消耗隊列中的 任務(wù)數(shù)減1
    # q.join() 一直阻塞，等待隊列中的任務(wù)數(shù)消耗完才釋放
    # 因為有 q.join 所有一直會等待 c1,c2 消耗完畢。才會執(zhí)行 p.join 后面的代碼
    # 因為 c1 c2 是守護(hù)進(jìn)程，所以到這一步主進(jìn)程代碼執(zhí)行完畢，主進(jìn)程會釋放死掉，
    # 所以 c1 c2 也會跟隨 主進(jìn)程釋放死掉。

使用JoinableQueue組件，是因為JoinableQueue中有兩個方法：task_done()和join() 。首先說join()和Process中的join()的效果類似，都是阻塞當(dāng)前進(jìn)程，防止當(dāng)前進(jìn)程結(jié)束。但是JoinableQueue的join()是和task_down()配合使用的。
Process中的join()是等到子進(jìn)程中的代碼執(zhí)行完畢，就會執(zhí)行主進(jìn)程join()下面的代碼。而JoinableQueue中的join()是等到隊列中的任務(wù)數(shù)量為0的時候才會執(zhí)行q.join()下面的代碼，否則會一直阻塞。
task_down()方法是每獲取一次隊列中的任務(wù)，就需要執(zhí)行一次。直到隊列中的任務(wù)數(shù)為0的時候，就會執(zhí)行JoinableQueue的join()后面的方法了。所以生產(chǎn)者生產(chǎn)完所有的數(shù)據(jù)后，會一直阻塞著。不讓p1和p2進(jìn)程結(jié)束。等到消費者get()一次數(shù)據(jù)，就會執(zhí)行一次task_down()方法，從而隊列中的任務(wù)數(shù)量減1,當(dāng)數(shù)量為0后，執(zhí)行JoinableQueue的join()后面代碼,從而p1和p2進(jìn)程結(jié)束。
因為p1和p2添加了join()方法，所以當(dāng)子進(jìn)程中的consumer方法執(zhí)行完后，才會往下執(zhí)行。從而主進(jìn)程結(jié)束。因為這里把消費者進(jìn)程c1和c2 設(shè)置成了守護(hù)進(jìn)程，主進(jìn)程結(jié)束的同時，c1和c2 進(jìn)程也會隨之結(jié)束，進(jìn)程都結(jié)束了。所以消費者consumer方法也會結(jié)束。