建站
咨詢
售后
建站咨詢
新聞中心
進程間的通信-Queue
1. Queue的使用
可以使用multiprocessing模塊的Queue實現(xiàn)多進程之間的數(shù)據(jù)傳遞，Queue本身是一個消息列隊程序，首先用一個小實例來演示一下Queue的工作原理：
#-*- coding:utf-8 -*-
from multiprocessing import Queue
#創(chuàng)建一個Queue對象，最多可接受三條put消息
q = Queue(3)
q.put("消息1")
q.put("消息2")
print(q.full())
q.put("消息3")
print(q.full())
try:
q.put("消息4",True,2)
except :
print("消息隊列已滿，現(xiàn)有消息數(shù)量：%s"%q.qsize())
try:
q.put_nowait("消息5")
except :
print("消息隊列已滿，現(xiàn)有消息數(shù)量：%s"%q.qsize())
#推薦方式，先判斷消息隊列是否已滿，在寫入
if not q.full():
q.put_nowait("消息6")
#讀取消息時，先判斷消息隊列是否為空，在讀取
if not q.empty():
for i in range(q.qsize()):
print(q.get_nowait())運行結果為：
False True 消息隊列已滿，現(xiàn)有消息數(shù)量：3 消息隊列已滿，現(xiàn)有消息數(shù)量：3 消息1 消息2 消息3
說明
初始化Queue()對象時（例如：q=Queue()），若括號中沒有指定可接收的消息數(shù)量，或數(shù)量為負值，那么就代表可接受的消息數(shù)量沒有上限（直到內存的盡頭）；
Queue.qsize()：返回當前隊列包含的消息數(shù)量；
Queue.empty()：如果隊列為空，返回True，反之False ；
Queue.full()：如果隊列滿了，返回True,反之False；
Queue.get([block[, timeout]])：獲取隊列中的一條消息，然后將其從列隊中移除，block默認值為True；
1）如果block使用默認值，且沒有設置timeout（單位秒），消息列隊如果為空，此時程序將被阻塞（停在讀取狀態(tài)），直到從消息列隊讀到消息為止，如果設置了timeout，則會等待timeout秒，若還沒讀取到任何消息，則拋出"Queue.Empty"異常；
2）如果block值為False，消息列隊如果為空，則會立刻拋出"Queue.Empty"異常；
Queue.get_nowait()：相當Queue.get(False)；
Queue.put(item,[block[, timeout]])：將item消息寫入隊列，block默認值為True；
1）如果block使用默認值，且沒有設置timeout（單位秒），消息列隊如果已經(jīng)沒有空間可寫入，此時程序將被阻塞（停在寫入狀態(tài)），直到從消息列隊騰出空間為止，如果設置了timeout，則會等待timeout秒，若還沒空間，則拋出"Queue.Full"異常；
2）如果block值為False，消息列隊如果沒有空間可寫入，則會立刻拋出"Queue.Full"異常；
Queue.put_nowait(item)：相當Queue.put(item, False)；
2. Queue實例
我們以Queue為例，在父進程中創(chuàng)建兩個子進程，一個往Queue里寫數(shù)據(jù)，一個從Queue里讀數(shù)據(jù)：
from multiprocessing import Process
from multiprocessing import Queue
import os
import time
import random
#寫數(shù)據(jù)進程執(zhí)行的代碼
def write(q):
for value in ["A","B","C"]:
print("Put %s to Queue "%value)
q.put(value)
time.sleep(random.random())
#讀取數(shù)據(jù)進程的代碼
def read(q):
while True:
if not q.empty():
value = q.get(True)
print("Get %s from Queue "%value)
time.sleep(random.random())
else:
break
if __name__ == '__main__':
#父進程創(chuàng)建Queue，并傳遞給各個子進程
q = Queue()
pw = Process(target = write,args=(q,))
pr = Process(target = read,args=(q,))
#啟動子進程pw，寫入
pw.start()
#等待pw結束
pw.join()
#啟動子進程pr，讀取
pr.start()
pr.join()
print("所有數(shù)據(jù)都寫入并且讀完")運行結果為：
Put A to Queue Put B to Queue Put C to Queue Get A from Queue Get B from Queue Get C from Queue
所有數(shù)據(jù)都寫入并且讀完。
3. 進程池中的Queue
如果要使用Pool創(chuàng)建進程，就需要使用multiprocessing.Manager()中的Queue()，而不是multiprocessing.Queue()，否則會得到一條如下的錯誤信息：
RuntimeError: Queue objects should only be shared between processes through inheritance.
#coding=utf-8
from multiprocessing import Manager
from multiprocessing import Pool
import os
import time
import random
def reader(q):
print("reader啟動(%d),父進程為(%d)"%(os.getpid(),os.getppid()))
for i in range(q.qsize()):
print("reader從Queue獲取到的消息時：%s"%q.get(True))
def writer(q):
print("writer啟動(%d),父進程為(%d)"%(os.getpid(),os.getppid()))
for i in "Se7eN_HOU":
q.put(i)
if __name__ == '__main__':
print("-------(%d) Start-------"%os.getpid())
#使用Manager中的Queue來初始化
q = Manager().Queue()
po = Pool()
#使用阻塞模式創(chuàng)建進程，這樣就不需要在reader中使用死循環(huán)了，可以讓writer完全執(zhí)行完成后，再用reader去讀取
po.apply(writer,(q,))
po.apply(reader,(q,))
po.close()
po.join()
print("-------(%d) End-------"%os.getpid())運行結果為：
-------(880) Start------- writer啟動(7744),父進程為(880) reader啟動(7936),父進程為(880) reader從Queue獲取到的消息時：S reader從Queue獲取到的消息時：e reader從Queue獲取到的消息時：7 reader從Queue獲取到的消息時：e reader從Queue獲取到的消息時：N reader從Queue獲取到的消息時：_ reader從Queue獲取到的消息時：H reader從Queue獲取到的消息時：O reader從Queue獲取到的消息時：U -------(880) End-------
相關推薦：
python中的進程池是什么
當前文章：創(chuàng)新互聯(lián)Python教程：Python如何進行進程間的通信
分享地址：http://m.fisionsoft.com.cn/article/djpidgp.html
