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

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

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

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

multiprocessing-Queue實現

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

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

multiprocessing-JoinableQueue實現

from multiprocessing import JoinableQueue,Process
import time,random
#生產者方法
def producer(name,food,q):
    for i in range(4):
        time.sleep(random.randint(1, 2))
        f = '%s生產的%s%s'%(name,food,i)
        q.put(f)
        print(f)
    q.join()  #一直阻塞，等待消耗完所有的數據后才釋放
#消費者方法
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)建隊列
    #模擬生產者隊列
    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 #守護進程：主進程結束，子進程也會結束
    c1.start()
    c2 = Process(target=consumer,args=('c2',q))
    c2.daemon = True
    c2.start()
    p1.join() #阻塞主進程，等到p1子進程結束才往下執(zhí)行
    p2.join()
    # q.task_done() 每次消耗隊列中的 任務數減1
    # q.join() 一直阻塞，等待隊列中的任務數消耗完才釋放
    # 因為有 q.join 所有一直會等待 c1,c2 消耗完畢。才會執(zhí)行 p.join 后面的代碼
    # 因為 c1 c2 是守護進程，所以到這一步主進程代碼執(zhí)行完畢，主進程會釋放死掉，
    # 所以 c1 c2 也會跟隨 主進程釋放死掉。

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