Python生產者消費者模型

一、消費模式

生產者消費者模式 是Controlnet網絡中特有的一種傳輸數據的模式。用于兩個CPU之間傳輸數據，即使是不同類型同一廠家的CPU也可以通過設置來使用。

二、傳輸原理

• 類似與點對點傳送，又略有不同，一個生產者可以對應N個消費者，但是一個消費者只能對應一個生產者；
• 每個生產者消費者對應一個地址，占一個網絡節(jié)點，屬于預定性數據，在網絡中優(yōu)先級最高；
• 此模式如果在網絡中設置過多會影響網絡傳輸速度，一般用在傳輸比較重要的信息上，比如設備的啟動、停止、故障、急停等等；
• 在Controlnet網絡中節(jié)點數是有限制的，最高節(jié)點數為99。
• 如果兩個控制器之前建立了多個生產者消費者的連接，只要一個失敗，則所有的均失敗，將數據整合到用戶自定義結構或數組中 ，兩個控制器中只保留一個連接。
• 生產者消費者信息可以通過以太網和Controlnet傳輸，但是同時只能通過一種途徑傳輸；
• 建立標簽時必須建立在全局變量里面，不能建立在局部變量里標簽的大小不能超過500B；
• 如果生產者幾個數據傳輸到到同一個控制器的的幾個消費者中，將幾個數據合并在一個用戶自定義標簽中，可以減少連接數，但合并后的數據將會會用相同的RPI。
• 生產者消費者標簽只能用DINT和REAL，或它們的數組，或用戶自定義結構數據，因為對外操作數據必須是32位的，如果有SINT和INT的數據要傳輸，必須將它們組合在用戶自定義結構中傳送，生產者和消費者的標簽數據格式必須一致，才能確保數據的準確性，如果數據打包后超過了 32位，那么生產者和消費者雙方必須使用一個復制緩沖指令，以獲得數據的同步，例如Control Logix中的CPS指令。
• 如果生產者要發(fā)送的32位數據，與非Control Logix的對方設備的數據結構不匹配，例如對方是16位的數據，為避免偏差，改為用戶自定義結構。
• 消費者的 RPI必須大于等于網絡刷新時間NUT，如果幾個消費者請求同一個生產者，則會以最小最快的RPI為準。

三、實現方式

方法一：

import threading,queue,time
# 創(chuàng)建一個隊列，隊列最大長度為2
q = queue.Queue(maxsize=2)
def product():
    while True:
        # 生產者往隊列塞數據
        q.put('money')
        print('生產了money， 生產時間：', time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
def consume():
    while True:
        time.sleep(0.5)
        # 消費者取出數據
        data = q.get()
        print('消費了%s， 消費時間%s' % (data, time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())))
t = threading.Thread(target=product)
t1 = threading.Thread(target=consume)
t.start()
t1.start()

缺點：

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

方法二：

from multiprocessing import JoinableQueue,Process
import time
def producer(q):
    for i in range(4):
        time.sleep(0.5)
        f = '生產者：已經生產'
        q.put(f)
        print(f)
    q.join()  # 一直阻塞，等待消耗完所有的數據后才釋放
def consumer(name, q):
    while True:
        food = q.get()
        print('\033[消費者：消費了%s\033' % name)
        time.sleep(0.5)
        q.task_done()  # 每次消耗減1
if __name__ == '__main__':
    q = JoinableQueue()  # 創(chuàng)建隊列
    # 模擬生產者隊列
    p1 = Process(target=producer, args=(q, ))
    p1.start()
    # 模擬消費者隊列
    c1 = Process(target=consumer, args=('money', q))
    c1.daemon = True  # 守護進程：主進程結束，子進程也會結束
    c1.start()
    p1.join()  # 阻塞主進程，等到p1子進程結束才往下執(zhí)行

優(yōu)點：參考地址

• 使用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方法也會結束。

到此這篇關于python多進程中的生產者和消費者模型詳解的文章就介紹到這了,更多相關python生產者和消費者模型內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！