Python生產(chǎn)者消費者模型

一、消費模式

生產(chǎn)者消費者模式 是Controlnet網(wǎng)絡(luò)中特有的一種傳輸數(shù)據(jù)的模式。用于兩個CPU之間傳輸數(shù)據(jù)，即使是不同類型同一廠家的CPU也可以通過設(shè)置來使用。

二、傳輸原理

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

三、實現(xiàn)方式

方法一：

import threading,queue,time
# 創(chuàng)建一個隊列，隊列最大長度為2
q = queue.Queue(maxsize=2)
def product():
    while True:
        # 生產(chǎn)者往隊列塞數(shù)據(jù)
        q.put('money')
        print('生產(chǎn)了money， 生產(chǎn)時間：', time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
def consume():
    while True:
        time.sleep(0.5)
        # 消費者取出數(shù)據(jù)
        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()

缺點：

    實現(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)于程序卡死。

方法二：

from multiprocessing import JoinableQueue,Process
import time
def producer(q):
    for i in range(4):
        time.sleep(0.5)
        f = '生產(chǎn)者：已經(jīng)生產(chǎn)'
        q.put(f)
        print(f)
    q.join()  # 一直阻塞，等待消耗完所有的數(shù)據(jù)后才釋放
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)建隊列
    # 模擬生產(chǎn)者隊列
    p1 = Process(target=producer, args=(q, ))
    p1.start()
    # 模擬消費者隊列
    c1 = Process(target=consumer, args=('money', q))
    c1.daemon = True  # 守護(hù)進(jìn)程：主進(jìn)程結(jié)束，子進(jìn)程也會結(jié)束
    c1.start()
    p1.join()  # 阻塞主進(jìn)程，等到p1子進(jìn)程結(jié)束才往下執(zhí)行

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

• 使用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é)束。

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

最新評論