快捷導(dǎo)航

python多進程下的生產(chǎn)者和消費者模型

更新時間：2020年05月07日 09:13:42 作者：李俊的博客

這篇文章主要介紹了python多進程下的生產(chǎn)者和消費者模型，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧

一、生產(chǎn)者消費者模型介紹

1.1 為什么需要使用生產(chǎn)者消費者模型

生產(chǎn)者是指生產(chǎn)數(shù)據(jù)的任務(wù)，消費者是指消費數(shù)據(jù)的任務(wù)。當生產(chǎn)者的生產(chǎn)能力遠大于消費者的消費能力，生產(chǎn)者就需要等消費者消費完才能繼續(xù)生產(chǎn)新的數(shù)據(jù)，同理，如果消費者的消費能力遠大于生產(chǎn)者的生產(chǎn)能力，消費者就需要等生產(chǎn)者生產(chǎn)完數(shù)據(jù)才能繼續(xù)消費，這種等待會造成效率的低下，為了解決這種問題就引入了生產(chǎn)者消費者模型。

1.2 如何實現(xiàn)生產(chǎn)者消費者模型

進程間引入隊列可以實現(xiàn)生產(chǎn)者消費者模型，通過使用隊列無需考慮鎖的概念，因為進程間的通信是通過隊列來實現(xiàn)的；

生產(chǎn)者生產(chǎn)的數(shù)據(jù)往隊列里面寫，消費者消費數(shù)據(jù)直接從隊列里面取，這樣就對實現(xiàn)了生產(chǎn)者和消費者之間的解耦。

生產(chǎn)者 -- > 隊列 <--消費者

二、Queue實現(xiàn)生產(chǎn)者消費者模型

2.1 消費者生產(chǎn)者模型代碼

from multiprocessing import Process, Queue
import time
 
# 消費者方法
def consumer(q, name):
  while True:
    res = q.get()
    # if res is None: break
    print("%s 吃了 %s" % (name, res))
 
# 生產(chǎn)者方法
def producer(q, name, food):
  for i in range(3):
    time.sleep(1) # 模擬生產(chǎn)西瓜的時間延遲
    res = "%s %s" % (food, i)
    print("%s 生產(chǎn)了 %s" % (name, res))
    # 把生產(chǎn)的vegetable放入到隊列中
    q.put(res)
 
if __name__ == "__main__":
  #創(chuàng)建隊列
  q = Queue()
  # 創(chuàng)建生產(chǎn)者
  p1 = Process(target=producer, args=(q, "kelly", "西瓜"))
  c1 = Process(target=consumer, args=(q, "peter",))
  p1.start()
  c1.start()
 
  # p1.join()
  # q.put(None)
  print("主進程")

2.2 執(zhí)行結(jié)果

2.2.1 直接執(zhí)行上面的代碼的結(jié)果

直接執(zhí)行會出現(xiàn)一個問題就是生產(chǎn)者生產(chǎn)完了，沒有向消費者發(fā)送一個停止的信號，所以消費者一直會一直阻塞在q.get()，導(dǎo)致程序無法退出。

為了解決上面的問題，讓消費者消費完了生產(chǎn)者的數(shù)據(jù)之后自動退出，就需要在生產(chǎn)者進程介紹的時候往隊列里面put一個結(jié)束信號，消費者拿到這個信號，就退出消費進程。

主要是兩個地方修改，把下方代碼的注釋打開就可以實現(xiàn)消費者消費完接收到生產(chǎn)者的結(jié)束信號就退出消費者進程了。

def consumer():
  if res is None: break
 
if __name__ == "__main__":
p1.join() 
q.put(None)

2.2.2 把注釋打開后的運行結(jié)果

把注釋打開后，消費者拿到了生產(chǎn)者發(fā)送的結(jié)束信號，可以正常退出程序了。

但如果有n個消費者，就需要發(fā)送n個結(jié)束信號，這種方式就不是那么簡潔，像下面的代碼這樣：

from multiprocessing import Process, Queue
import time
 
 
# 消費者方法
def consumer(q, name):
  while True:
    res = q.get()
    if res is None: break
    print("%s 吃了 %s" % (name, res))
 
 
# 生產(chǎn)者方法
def producer(q, name, food):
  for i in range(3):
    time.sleep(1) # 模擬生產(chǎn)西瓜的時間延遲
    res = "%s %s" % (food, i)
    print("%s 生產(chǎn)了 %s" % (name, res))
    # 把生產(chǎn)的vegetable放入到隊列中
    q.put(res)
 
 
if __name__ == "__main__":
  # 創(chuàng)建隊列
  q = Queue()
  # 創(chuàng)建生產(chǎn)者
  p1 = Process(target=producer, args=(q, "kelly", "西瓜"))
  p2 = Process(target=producer, args=(q, "kelly2", "香蕉"))
  c1 = Process(target=consumer, args=(q, "peter",))
  c2 = Process(target=consumer, args=(q, "peter2",))
  c3 = Process(target=consumer, args=(q, "peter3",))
  p1.start()
  p2.start()
  c1.start()
  c2.start()
  c3.start()
 
  p1.join()
  p2.join()
  q.put(None)
  q.put(None)
  q.put(None)
  print("主進程")

其實我們現(xiàn)在就是生產(chǎn)者生產(chǎn)完數(shù)據(jù)之后想往隊列里面發(fā)送一個結(jié)束信號，python語言提供了另外一種隊列JoinableQueue([maxsize])來解決這種問題

三、JoinableQueue實現(xiàn)生產(chǎn)者消費者模型

3.1 JoinableQueue方法介紹

JoinableQueue([maxsize]) ： A queue type which also supports join() and task_done() methods

q.task_done()：消費者使用此方法發(fā)出信號，表示q.get()的返回項目已經(jīng)被處理。

q.join()：生產(chǎn)者調(diào)用此方法進行阻塞，直到隊列中所有的項目均被處理；阻塞將持續(xù)到隊列中的每個項目均調(diào)用q.task_done（）方法為止。

3.2 JoinableQueue實現(xiàn)生產(chǎn)者消費者模型源碼

from multiprocessing import Process,JoinableQueue
import time
 
 
# 消費者方法
def consumer(q, name):
  while True:
    res = q.get()
    if res is None: break
    print("%s 吃了 %s" % (name, res))
    q.task_done() # 發(fā)送信號給q.join(),表示已經(jīng)從隊列中取走一個值并處理完畢了
 
 
# 生產(chǎn)者方法
def producer(q, name, food):
  for i in range(3):
    time.sleep(1) # 模擬生產(chǎn)西瓜的時間延遲
    res = "%s %s" % (food, i)
    print("%s 生產(chǎn)了 %s" % (name, res))
    # 把生產(chǎn)的vegetable放入到隊列中
    q.put(res)
  q.join() # 等消費者把自己放入隊列的所有元素取完之后才結(jié)束
 
 
if __name__ == "__main__":
  # q = Queue()
  q = JoinableQueue()
  # 創(chuàng)建生產(chǎn)者
  p1 = Process(target=producer, args=(q, "kelly", "西瓜"))
  p2 = Process(target=producer, args=(q, "kelly2", "藍莓"))
  # 創(chuàng)建消費者
  c1 = Process(target=consumer, args=(q, "peter",))
  c2 = Process(target=consumer, args=(q, "peter2",))
  c3 = Process(target=consumer, args=(q, "peter3",))
 
  c1.daemon = True
  c2.daemon = True
  c3.daemon = True
 
  p_l = [p1, p2, c1, c2, c3]
  for p in p_l:
    p.start()
  
  p1.join()
  p2.join()
  # 1.主進程等待p1,p2進程結(jié)束才繼續(xù)執(zhí)行
  # 2.由于q.join()的存在,生產(chǎn)者只有等隊列中的元素被消費完才會結(jié)束
  # 3.生產(chǎn)者結(jié)束了,就代表消費者已經(jīng)消費完了,也可以結(jié)束了,所以可以把消費者設(shè)置為守護進程(隨著主進程的退出而退出)
 
  print("主進程")

3.3 運行結(jié)果

通過運行結(jié)果可以看出，生產(chǎn)者沒有手動發(fā)送結(jié)束信號給消費者，而是通過JoinableQueue隊列的方式也實現(xiàn)了生產(chǎn)者消費者模型。