Python進程間通信 multiProcessing Queue隊列實現(xiàn)詳解
一、進程間通信
IPC(Inter-Process Communication)
IPC機制：實現(xiàn)進程之間通訊
管道：pipe 基于共享的內(nèi)存空間
隊列：pipe+鎖的概念--->queue
二、隊列(Queue)
2.1 概念-----multiProcess.Queue
創(chuàng)建共享的進程隊列，Queue是多進程安全的隊列，可以使用Queue實現(xiàn)多進程之間的數(shù)據(jù)傳遞。
Queue([maxsize])創(chuàng)建共享的進程隊列。
參數(shù) ：maxsize是隊列中允許的最大項數(shù)。如果省略此參數(shù)，則無大小限制。
底層隊列使用管道和鎖定實現(xiàn)。
2.2 Queue方法使用
2.2.1 q.get的使用：
是從隊列里面取值并且把隊列面的取出來的值刪掉，沒有參數(shù)的情況下就是是默認(rèn)一直等著取值
就算是隊列里面沒有可取的值的時候，程序也不會結(jié)束，就會卡在哪里，一直等著
from multiprocessing import Queue
q = Queue() # 生成一個隊列對象
# put方法是往隊列里面放值
q.put('Cecilia陳')
q.put('xuchen')
q.put('喜陳')
# get方法是從隊列里面取值
print(q.get())
print(q.get())
print(q.get())
q.put(5)
q.put(6)
print(q.get())
Cecilia陳
xuchen
喜陳
5
2.2.2 Queue(參數(shù)) +參數(shù)的使用：
Queue加參數(shù)以后，參數(shù)是數(shù)值
參數(shù)實幾就表示實例化的這個Queue隊列可以放幾個值
當(dāng)隊列已經(jīng)滿的時候，再放值,程序會阻塞，但不會結(jié)束
from multiprocessing import Queue
q = Queue(3)
q.put('Cecilia陳')
q.put('xuchen')
q.put('喜陳')
print(q.full()) # 判斷隊列是否滿了 返回的是True/False
q.put(2) # 當(dāng)隊列已經(jīng)滿的時候，再放值,程序會阻塞，但不會結(jié)束
True 隊列已經(jīng)滿了
2.2.3 q.put(參數(shù)1，參數(shù)2，參數(shù)3，參數(shù)4)：
q.put(self, obj, block=True, timeout=None)
self ：put就相當(dāng)于是Queue里的一個方法，這個時候q.put就相當(dāng)于是隊列對象q來調(diào)用對象的綁定方法，這個參數(shù)可以省略即可
obj：是我們需要往隊列里面放的值
block=True ：隊列如果滿了的話，再往隊列里放值的話會等待，程序不會結(jié)束
timeout=None：是再block這個參數(shù)的基礎(chǔ)上的，當(dāng)block的值為真的時候，timeout是用來等待多少秒，如果再這個時間里，隊列一直是滿的，那么程序就會報錯并結(jié)束（Queue.Full異常）
from multiprocessing import Queue
q = Queue(3)
q.put('zhao',block=True,timeout=2)
q.put('zhao',block=True,timeout=2)
q.put('zhao',block=True,timeout=2)
q.put('zhao',block=True,timeout=5) # 此時程序?qū)Φ却?秒以后報錯了
2.2.4 q.get(參數(shù)1，參數(shù)2，參數(shù)3，參數(shù)4):
q.get(self,block=True, timeout=None)
self ：get就相當(dāng)于是Queue里的一個方法，這個時候q.get就相當(dāng)于是隊列對象q來調(diào)用對象的綁定方法，這個參數(shù)可以省略即可
block=True ：從隊列q對象里面取值，如果娶不到值的話，程序不會結(jié)束
timeout=None：是再block這個參數(shù)的基礎(chǔ)上的，當(dāng)block的值為真的時候，timeout是用來等待多少秒,如果再這個時間里，get取不到隊列里面的值的話，那么程序就會報錯并結(jié)束（queue.Empty異常）
from multiprocessing import Queue
q = Queue()
q.put('Cecilia陳')
print(q.get())
q.get(block=True,timeout=2) # 此時程序會等待2秒后，報錯了，隊列里面沒有值了
2.2.5 block=False：
如果block的值是False的話，那么put方法再隊列是滿的情況下，不會等待阻塞，程序直接報錯（Queue.Full異常）結(jié)束
如果block的值是False的話，那么get方法再隊列里面沒有值的情況下，再去取的時候，不會等待阻塞，程序直接報錯（queue.Empty異常）結(jié)束
1.put()的block=False
from multiprocessing import Queue
q = Queue(2)
q.put('Cecilia陳')
q.put('喜陳')
print(q.full())
q.put('xichen',block=False) # 隊列已經(jīng)滿了，我不等待了，直接報錯
2.get()的block=Flase
from multiprocessing import Queue
q = Queue(2)
q.put('Cecilia陳')
q.put('喜陳')
print(q.get())
print(q.get())
print(q.get(block=False)) # 隊列已經(jīng)沒有值了，我不等待了，直接報錯
2.2.6 put_nowait()/get_nowait()
1.put_nowait() 相當(dāng)于bolok=False，隊列滿的時候，再放值的時候，程序不等待，不阻塞，直接報錯
from multiprocessing import Queue
q = Queue(2)
q.put('Cecilia陳')
q.put('喜陳')
print(q.full())
q.put_nowait('xichen') # 程序不等待，不阻塞，直接報錯
2.get_nowait() 相當(dāng)于bolok=False,當(dāng)隊列里沒有值的時候，再取值的時候，程序不等待，不阻塞，程序直接報錯
from multiprocessing import Queue
q = Queue(2)
q.put('Cecilia陳')
q.put('喜陳')
print(q.get())
print(q.get())
print(q.full())
q.get_nowait()# 再取值的時候，程序不等待，不阻塞，程序直接報錯
三、代碼實例
3.1 單看隊列的存取數(shù)據(jù)用法
這個例子還沒有加入進程通信，只是先來看看隊列為我們提供的方法，以及這些方法的使用和現(xiàn)象。
'''
multiprocessing模塊支持進程間通信的兩種主要形式:管道和隊列
都是基于消息傳遞實現(xiàn)的,但是隊列接口
'''
from multiprocessing import Queue
q=Queue(3)
#put ,get ,put_nowait,get_nowait,full,empty
q.put(3)
q.put(3)
q.put(3)
# q.put(3) # 如果隊列已經(jīng)滿了，程序就會停在這里，等待數(shù)據(jù)被別人取走，再將數(shù)據(jù)放入隊列。
# 如果隊列中的數(shù)據(jù)一直不被取走，程序就會永遠(yuǎn)停在這里。
try:
q.put_nowait(3) # 可以使用put_nowait，如果隊列滿了不會阻塞，但是會因為隊列滿了而報錯。
except: # 因此我們可以用一個try語句來處理這個錯誤。這樣程序不會一直阻塞下去，但是會丟掉這個消息。
print('隊列已經(jīng)滿了')
# 因此，我們再放入數(shù)據(jù)之前，可以先看一下隊列的狀態(tài)，如果已經(jīng)滿了，就不繼續(xù)put了。
print(q.full()) #滿了
print(q.get())
print(q.get())
print(q.get())
# print(q.get()) # 同put方法一樣，如果隊列已經(jīng)空了，那么繼續(xù)取就會出現(xiàn)阻塞。
try:
q.get_nowait(3) # 可以使用get_nowait，如果隊列滿了不會阻塞，但是會因為沒取到值而報錯。
except: # 因此我們可以用一個try語句來處理這個錯誤。這樣程序不會一直阻塞下去。
print('隊列已經(jīng)空了')
print(q.empty()) #空了
3.2 子進程向父進程發(fā)送數(shù)據(jù)
這是一個queue的簡單應(yīng)用，使用隊列q對象調(diào)用get函數(shù)來取得隊列中最先進入的數(shù)據(jù)。
from multiprocessing import Process, Queue def f(q,name,age): q.put(name,age) #調(diào)用主函數(shù)中p進程傳遞過來的進程參數(shù) put函數(shù)為向隊列中添加一條數(shù)據(jù)。 if __name__ == '__main__': q = Queue() #創(chuàng)建一個Queue對象 p = Process(target=f, args=(q,'Cecilia陳',18)) #創(chuàng)建一個進程 p.start() print(q.get()) p.join()
['Cecilia陳', 18]
四、生產(chǎn)者消費者模型
生產(chǎn)者: 生產(chǎn)數(shù)據(jù)的任務(wù)
消費者: 處理數(shù)據(jù)的任務(wù)
生產(chǎn)者--隊列(盆)-->消費者
生產(chǎn)者可以不停的生產(chǎn),達到了自己最大的生產(chǎn)效率,消費者可以不停的消費,也達到了自己最大的消費效率.
生產(chǎn)者消費者模型大大提高了生產(chǎn)者生產(chǎn)的效率和消費者消費的效率.
補充: queue不適合傳大文件,通產(chǎn)傳一些消息.
在并發(fā)編程中使用生產(chǎn)者和消費者模式能夠解決絕大多數(shù)并發(fā)問題。該模式通過平衡生產(chǎn)線程和消費線程的工作能力來提高程序的整體處理數(shù)據(jù)的速度。
4.1 為什么要使用生產(chǎn)者和消費者模型
在線程世界里，生產(chǎn)者就是生產(chǎn)數(shù)據(jù)的線程，消費者就是消費數(shù)據(jù)的線程。在多線程開發(fā)當(dāng)中，如果生產(chǎn)者處理速度很快，而消費者處理速度很慢，那么生產(chǎn)者就必須等待消費者處理完，才能繼續(xù)生產(chǎn)數(shù)據(jù)。同樣的道理，如果消費者的處理能力大于生產(chǎn)者，那么消費者就必須等待生產(chǎn)者。為了解決這個問題于是引入了生產(chǎn)者和消費者模式。
4.2 什么是生產(chǎn)者消費者模型
生產(chǎn)者消費者模式是通過一個容器來解決生產(chǎn)者和消費者的強耦合問題。生產(chǎn)者和消費者彼此之間不直接通訊，而通過阻塞隊列來進行通訊，所以生產(chǎn)者生產(chǎn)完數(shù)據(jù)之后不用等待消費者處理，直接扔給阻塞隊列，消費者不找生產(chǎn)者要數(shù)據(jù)，而是直接從阻塞隊列里取，阻塞隊列就相當(dāng)于一個緩沖區(qū)，平衡了生產(chǎn)者和消費者的處理能力。
4.3 基于Queue隊列實現(xiàn)的生產(chǎn)者消費者模型
from multiprocessing import Queue,Process
# 生產(chǎn)者
def producer(q,name,food):
for i in range(3):
print(f'{name}生產(chǎn)了{(lán)food}{i}')
res = f'{food}{i}'
q.put(res)
# 消費者
def consumer(q,name):
while True:
res = q.get(timeout=5)
print(f'{name}吃了{(lán)res}')
if __name__ == '__main__':
q = Queue() # 為的是讓生產(chǎn)者和消費者使用同一個隊列，使用同一個隊列進行通訊
p1 = Process(target=producer,args=(q,'Cecilia陳','巧克力'))
c1 = Process(target=consumer,args=(q,'Tom'))
p1.start()
c1.start()
此時的問題是主進程永遠(yuǎn)不會結(jié)束，原因是：生產(chǎn)者p在生產(chǎn)完后就結(jié)束了，但是消費者c在取空了q之后，則一直處于死循環(huán)中且卡在q.get()這一步。
解決方式無非是讓生產(chǎn)者在生產(chǎn)完畢后，往隊列中再發(fā)一個結(jié)束信號，這樣消費者在接收到結(jié)束信號后就可以break出死循環(huán)。
4.4 改良版----生產(chǎn)者消費者模型
注意：結(jié)束信號None，不一定要由生產(chǎn)者發(fā)，主進程里同樣可以發(fā)，但主進程需要等生產(chǎn)者結(jié)束后才應(yīng)該發(fā)送該信號
from multiprocessing import Queue,Process
def producer(q,name,food):
for i in range(3):
print(f'{name}生產(chǎn)了{(lán)food}{i}')
res = f'{food}{i}'
q.put(res)
q.put(None) # 當(dāng)生產(chǎn)者結(jié)束生產(chǎn)的的時候，我們再隊列的最后再做一個表示，告訴消費者，生產(chǎn)者已經(jīng)不生產(chǎn)了，讓消費者不要再去隊列里拿東西了
def consumer(q,name):
while True:
res = q.get(timeout=5)
if res == None:break # 判斷隊列拿出的是不是生產(chǎn)者放的結(jié)束生產(chǎn)的標(biāo)識，如果是則不取，直接退出，結(jié)束程序
print(f'{name}吃了{(lán)res}')
if __name__ == '__main__':
q = Queue() # 為的是讓生產(chǎn)者和消費者使用同一個隊列，使用同一個隊列進行通訊
p1 = Process(target=producer,args=(q,'Cecilia陳','巧克力'))
c1 = Process(target=consumer,args=(q,'Tom'))
p1.start()
c1.start()
4.5 主進程在生產(chǎn)者生產(chǎn)結(jié)束以后，發(fā)送結(jié)束信號
使用這個方法的話，是很low的，有幾個消費者就要在主進程中向隊列中put幾個結(jié)束信號
from multiprocessing import Queue,Process
import time,random
def producer(q,name,food):
for i in range(3):
print(f'{name}生產(chǎn)了{(lán)food}{i}')
time.sleep((random.randint(1,3)))
res = f'{food}{i}'
q.put(res)
# q.put(None) # 當(dāng)生產(chǎn)者結(jié)束生產(chǎn)的的時候，我們再隊列的最后再做一個表示，告訴消費者，生產(chǎn)者已經(jīng)不生產(chǎn)了，讓消費者不要再去隊列里拿東西了
def consumer(q,name):
while True:
res = q.get(timeout=5)
if res == None:break # 判斷隊列拿出的是不是生產(chǎn)者放的結(jié)束生產(chǎn)的標(biāo)識，如果是則不取，直接退出，結(jié)束程序
time.sleep((random.randint(1, 3)))
print(f'{name}吃了{(lán)res}')
if __name__ == '__main__':
q = Queue() # 為的是讓生產(chǎn)者和消費者使用同一個隊列，使用同一個隊列進行通訊
# 多個生產(chǎn)者進程
p1 = Process(target=producer,args=(q,'Cecilia陳','巧克力'))
p2 = Process(target=producer,args=(q,'xichen','冰激凌'))
p3 = Process(target=producer,args=(q,'喜陳','可樂'))
# 多個消費者進程
c1 = Process(target=consumer,args=(q,'Tom'))
c2 = Process(target=consumer,args=(q,'jack'))
# 告訴操作系統(tǒng)啟動生產(chǎn)者進程
p1.start()
p2.start()
p3.start()
# 告訴操作系統(tǒng)啟動消費者進程
c1.start()
c2.start()
p1.join()
p2.join()
p3.join()
q.put(None) # 幾個消費者put幾次
q.put(None)
五、JoinableQueue方法
創(chuàng)建可連接的共享進程隊列。這就像是一個Queue對象，但隊列允許項目的使用者通知生產(chǎn)者項目已經(jīng)被成功處理。通知進程是使用共享的信號和條件變量來實現(xiàn)的。
5.1 方法介紹
JoinableQueue的實例p除了與Queue對象相同的方法之外，還具有以下方法：
q.task_done()：使用者使用此方法發(fā)出信號，表示q.get()返回的項目已經(jīng)被處理。如果調(diào)用此方法的次數(shù)大于從隊列中刪除的項目數(shù)量，將引發(fā)ValueError異常。
q.join()：生產(chǎn)者將使用此方法進行阻塞，直到隊列中所有項目均被處理。阻塞將持續(xù)到為隊列中的每個項目均調(diào)用q.task_done()方法為止。
5.2 joinableQueue隊列實現(xiàn)生產(chǎn)者消費者模型
from multiprocessing import Queue,Process,JoinableQueue
import time,random
def producer(q,name,food):
for i in range(3):
print(f'{name}生產(chǎn)了{(lán)food}{i}')
# time.sleep((random.randint(1,3)))
res = f'{food}{i}'
q.put(res)
# q.put(None) # 當(dāng)生產(chǎn)者結(jié)束生產(chǎn)的的時候，我們再隊列的最后再做一個表示，告訴消費者，生產(chǎn)者已經(jīng)不生產(chǎn)了，讓消費者不要再去隊列里拿東西了
q.join()
def consumer(q,name):
while True:
res = q.get(timeout=5)
# if res == None:break # 判斷隊列拿出的是不是生產(chǎn)者放的結(jié)束生產(chǎn)的標(biāo)識，如果是則不取，直接退出，結(jié)束程序
# time.sleep((random.randint(1, 3)))
print(f'{name}吃了{(lán)res}')
q.task_done()#向q.join()發(fā)送一次信號,證明一個數(shù)據(jù)已經(jīng)被取走了
if __name__ == '__main__':
q = JoinableQueue() # 為的是讓生產(chǎn)者和消費者使用同一個隊列，使用同一個隊列進行通訊
# 多個生產(chǎn)者進程
p1 = Process(target=producer,args=(q,'Cecilia陳','巧克力'))
p2 = Process(target=producer,args=(q,'xichen','冰激凌'))
p3 = Process(target=producer,args=(q,'喜陳','可樂'))
# 多個消費者進程
c1 = Process(target=consumer,args=(q,'Tom'))
c2 = Process(target=consumer,args=(q,'jack'))
# 告訴操作系統(tǒng)啟動生產(chǎn)者進程
p1.start()
p2.start()
p3.start()
# 把生產(chǎn)者設(shè)為守護進程
c1.daemon = True
c2.daemon = True
# 告訴操作系統(tǒng)啟動消費者進程
c1.start()
c2.start()
p1.join()
p2.join()
p3.join() # 等待生產(chǎn)者生產(chǎn)完畢
print('主進程')
### 分析
# 生產(chǎn)者生產(chǎn)完畢--這是主進程最后一行代碼結(jié)束--q.join()消費者已經(jīng)取干凈了,沒有存在的意義了
# 這是主進程最后一行代碼結(jié)束,消費者已經(jīng)取干凈了,沒有存在的意義了.守護進程的概念.
5.3 測試joinableQueue
from multiprocessing import Process,Queue,JoinableQueue
q = JoinableQueue()
q.put('zhao') # 放隊列里一個任務(wù)
q.put('qian')
print(q.get())
q.task_done() # 完成了一次任務(wù)
print(q.get())
q.task_done() # 完成了一次任務(wù)
q.join() #計數(shù)器不為0的時候 阻塞等待計數(shù)器為0后通過
# 想象成一個計數(shù)器 :put +1 task_done -1
以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。
相關(guān)文章
PyQt是一個創(chuàng)建GUI應(yīng)用程序的工具包。它是Python編程語言和Qt庫的成功融合。Qt庫是最強大的庫之一。PyQt是由Phil?Thompson?開發(fā)。接下來通過本文給大家介紹下使用pyqt彈出消息提示框的問題,需要的朋友可以參考下2022-01-01
pycharm2021激活碼是一個可以輕松幫助用戶免費激活pycharm2021.1軟件的文件，雖然說pycharm現(xiàn)在只是推出了2021.1的EAP版，但是如果你想先率先體驗一波，那么就可以利用小編提供的這個激活碼來進行使用啦，并這個激活碼是永久有效的2021-03-03
這篇文章主要介紹了python?logging模塊詳解及其日志定時清理方式,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2025-04-04
解決List.append()?在?Python?中不起作用的問題
在?Python?中，我們通常使用?List.append()?方法向列表末尾添加元素，然而，在某些情況下，你可能會遇到?List.append()?方法不起作用的問題，本文將詳細(xì)討論這個問題并提供解決方法，需要的朋友可以參考下2023-06-06
TensorFlow打印tensor值的實現(xiàn)方法
今天小編就為大家分享一篇TensorFlow打印tensor值的實現(xiàn)方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2018-07-07
這篇文章主要介紹了Python實現(xiàn)打印http請求信息方式,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2024-06-06
這篇文章主要為大家詳細(xì)介紹了將Python項目打包成exe并附帶下載功能的操作流程,文中的示例代碼講解詳細(xì),具有一定的借鑒價值,感興趣的小伙伴可以了解下2023-12-12
詳解Python常用標(biāo)準(zhǔn)庫之os模塊與shutil模塊
os系統(tǒng)模塊與shutil文件操作模塊是Python常用的標(biāo)準(zhǔn)庫，本文將通過示例詳細(xì)講解一下二者的使用，感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下2022-06-06
雪花算法是在一個項目體系中生成全局唯一ID標(biāo)識的一種方式，偶然間看到了Python使用雪花算法不盡感嘆真的是太便捷了。本文就來聊聊這具體的實現(xiàn)方法，感興趣的可以了解一下2022-11-11
這篇文章主要為大家詳細(xì)介紹了Ubuntu安裝Jupyter Notebook教程，具有一定的參考價值，感興趣的小伙伴們可以參考一下2017-10-10