生成者消費者模式

1. 解釋

   

   成都創(chuàng)新互聯(lián)公司專注于企業(yè)營銷型網(wǎng)站、網(wǎng)站重做改版、包河網(wǎng)站定制設計、自適應品牌網(wǎng)站建設、H5高端網(wǎng)站建設、購物商城網(wǎng)站建設、集團公司官網(wǎng)建設、成都外貿(mào)網(wǎng)站建設公司、高端網(wǎng)站制作、響應式網(wǎng)頁設計等建站業(yè)務，價格優(yōu)惠性價比高，為包河等各大城市提供網(wǎng)站開發(fā)制作服務。

   • 什么是生產(chǎn)者和消費者？

     兩個模塊（可能是函數(shù)、線程或進程），一個生產(chǎn)數(shù)據(jù)，一個消費數(shù)據(jù)，生產(chǎn)數(shù)據(jù)的模塊叫生產(chǎn)者，消費數(shù)據(jù)的模塊叫消費者。

   • 為什么要使用生產(chǎn)者-消費者模式？

     在多線程開發(fā)時，生產(chǎn)者可能產(chǎn)生數(shù)據(jù)的速度比較快，這時候生產(chǎn)者需要等消費者消費完了才能繼續(xù)生產(chǎn)數(shù)據(jù)，因為生產(chǎn)的數(shù)據(jù)沒地方放，如果消費者消費數(shù)據(jù)的速度比較快，消費者就要等待生產(chǎn)者生產(chǎn)數(shù)據(jù)，為了解決這種供需不平衡的問題，以達到生產(chǎn)者和消費者之間的一種“平衡”狀態(tài)，引入“生產(chǎn)者-消費者“模式

   • 什么是生產(chǎn)者消費者模式？

     一個生產(chǎn)者（線程或進程）生產(chǎn)數(shù)據(jù)，一個消費者（線程或進程）消費數(shù)據(jù)，它們之間通過一個緩沖區(qū)來進行通訊，生產(chǎn)者生產(chǎn)的數(shù)據(jù)放入緩沖區(qū)，消費者從緩沖區(qū)取來數(shù)據(jù)進行消費。

   • 生產(chǎn)者消費者模式有什么優(yōu)點？

     • 解耦：將生產(chǎn)者類和消費者類進行解耦，消除代碼之間的依賴性，簡化工作負載的管理
     • 復用：通過將生產(chǎn)者類和消費者類獨立開來，那么可以對生產(chǎn)者類和消費者類進行獨立的復用與擴展
     • 調(diào)整并發(fā)數(shù)：由于生產(chǎn)者和消費者的處理速度是不一樣的，可以調(diào)整并發(fā)數(shù)，給予慢的一方多的并發(fā)數(shù)，來提高任務的處理速度
     • 異步：對于生產(chǎn)者和消費者來說能夠各司其職，生產(chǎn)者只需要關心緩沖區(qū)是否還有數(shù)據(jù)，不需要等待消費者處理完；同樣的對于消費者來說，也只需要關注緩沖區(qū)的內(nèi)容，不需要關注生產(chǎn)者，通過異步的方式支持高并發(fā)，將一個耗時的流程拆成生產(chǎn)和消費兩個階段，這樣生產(chǎn)者因為執(zhí)行put()的時間比較短，而支持高并發(fā)
     • 支持分布式：生產(chǎn)者和消費者通過隊列進行通訊，所以不需要運行在同一臺機器上，在分布式環(huán)境中可以通過redis的list作為隊列，而消費者只需要輪詢隊列中是否有數(shù)據(jù)。同時還能支持集群的伸縮性，當某臺機器宕掉的時候，不會導致整個集群宕掉

2. 生產(chǎn)者消費者例子

   1. e.g.

      5個廚師做包子，三秒一個，10個食客吃包子，1秒一個，食客吃的又快人又多

      import queue
      import threading
      import time
      
      q = queue.Queue()
      
      
      def cook(i):
          while True:
              q.put(f'廚師{i}做de包子')
              time.sleep(2)
      
      def hangers_on(j):
          while True:
              print(f'食客{j}吃了{q.get()}')
              time.sleep(1)
      
      for i in range(5):
          threading.Thread(target=cook,args=(i,)).start()
      
      for j in range(10):
          threading.Thread(target=hangers_on,args=(j,)).start()
      
      

   2. 用協(xié)程實現(xiàn)同樣的例子

      q = asyncio.Queue()
      
      
      async def hangers_on(j):
          while True:
              print(f'食客{j}吃了{await q.get()}')
              await asyncio.sleep(3)
      
      async def cook(i):
          while True:
              print(f'廚師{i}做的包子')
              await q.put(f'廚師{i}做的包子')
              await asyncio.sleep(1)
      
      
      async def main():
      
      
          cook_tasks = [asyncio.create_task(cook(j)) for j in range(10)]
          hanges_tasks = [asyncio.create_task(hangers_on(i)) for i in range(10)]
          await asyncio.gather(*cook_tasks,*hanges_tasks,return_exceptions = True)
      
      
      asyncio.run(main())