建站
咨詢
售后
建站咨詢
新聞中心
進(jìn)程間的通信-Queue
1. Queue的使用
可以使用multiprocessing模塊的Queue實(shí)現(xiàn)多進(jìn)程之間的數(shù)據(jù)傳遞，Queue本身是一個(gè)消息列隊(duì)程序，首先用一個(gè)小實(shí)例來演示一下Queue的工作原理：
#-*- coding:utf-8 -*-
from multiprocessing import Queue
#創(chuàng)建一個(gè)Queue對(duì)象，最多可接受三條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("消息隊(duì)列已滿，現(xiàn)有消息數(shù)量：%s"%q.qsize())
try:
q.put_nowait("消息5")
except :
print("消息隊(duì)列已滿，現(xiàn)有消息數(shù)量：%s"%q.qsize())
#推薦方式，先判斷消息隊(duì)列是否已滿，在寫入
if not q.full():
q.put_nowait("消息6")
#讀取消息時(shí)，先判斷消息隊(duì)列是否為空，在讀取
if not q.empty():
for i in range(q.qsize()):
print(q.get_nowait())運(yùn)行結(jié)果為：
False True 消息隊(duì)列已滿，現(xiàn)有消息數(shù)量：3 消息隊(duì)列已滿，現(xiàn)有消息數(shù)量：3 消息1 消息2 消息3
說明
初始化Queue()對(duì)象時(shí)（例如：q=Queue()），若括號(hào)中沒有指定可接收的消息數(shù)量，或數(shù)量為負(fù)值，那么就代表可接受的消息數(shù)量沒有上限（直到內(nèi)存的盡頭）；
Queue.qsize()：返回當(dāng)前隊(duì)列包含的消息數(shù)量；
Queue.empty()：如果隊(duì)列為空，返回True，反之False ；
Queue.full()：如果隊(duì)列滿了，返回True,反之False；
Queue.get([block[, timeout]])：獲取隊(duì)列中的一條消息，然后將其從列隊(duì)中移除，block默認(rèn)值為True；
1）如果block使用默認(rèn)值，且沒有設(shè)置timeout（單位秒），消息列隊(duì)如果為空，此時(shí)程序?qū)⒈蛔枞ㄍＴ谧x取狀態(tài)），直到從消息列隊(duì)讀到消息為止，如果設(shè)置了timeout，則會(huì)等待timeout秒，若還沒讀取到任何消息，則拋出"Queue.Empty"異常；
2）如果block值為False，消息列隊(duì)如果為空，則會(huì)立刻拋出"Queue.Empty"異常；
Queue.get_nowait()：相當(dāng)Queue.get(False)；
Queue.put(item,[block[, timeout]])：將item消息寫入隊(duì)列，block默認(rèn)值為True；
1）如果block使用默認(rèn)值，且沒有設(shè)置timeout（單位秒），消息列隊(duì)如果已經(jīng)沒有空間可寫入，此時(shí)程序?qū)⒈蛔枞ㄍＴ趯懭霠顟B(tài)），直到從消息列隊(duì)騰出空間為止，如果設(shè)置了timeout，則會(huì)等待timeout秒，若還沒空間，則拋出"Queue.Full"異常；
2）如果block值為False，消息列隊(duì)如果沒有空間可寫入，則會(huì)立刻拋出"Queue.Full"異常；
Queue.put_nowait(item)：相當(dāng)Queue.put(item, False)；
2. Queue實(shí)例
我們以Queue為例，在父進(jìn)程中創(chuàng)建兩個(gè)子進(jìn)程，一個(gè)往Queue里寫數(shù)據(jù)，一個(gè)從Queue里讀數(shù)據(jù)：
from multiprocessing import Process
from multiprocessing import Queue
import os
import time
import random
#寫數(shù)據(jù)進(jìn)程執(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ù)進(jìn)程的代碼
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__':
#父進(jìn)程創(chuàng)建Queue，并傳遞給各個(gè)子進(jìn)程
q = Queue()
pw = Process(target = write,args=(q,))
pr = Process(target = read,args=(q,))
#啟動(dòng)子進(jìn)程pw，寫入
pw.start()
#等待pw結(jié)束
pw.join()
#啟動(dòng)子進(jìn)程pr，讀取
pr.start()
pr.join()
print("所有數(shù)據(jù)都寫入并且讀完")運(yùn)行結(jié)果為：
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. 進(jìn)程池中的Queue
如果要使用Pool創(chuàng)建進(jìn)程，就需要使用multiprocessing.Manager()中的Queue()，而不是multiprocessing.Queue()，否則會(huì)得到一條如下的錯(cuò)誤信息：
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òng)(%d),父進(jìn)程為(%d)"%(os.getpid(),os.getppid()))
for i in range(q.qsize()):
print("reader從Queue獲取到的消息時(shí)：%s"%q.get(True))
def writer(q):
print("writer啟動(dòng)(%d),父進(jìn)程為(%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)建進(jìn)程，這樣就不需要在reader中使用死循環(huán)了，可以讓writer完全執(zhí)行完成后，再用reader去讀取
po.apply(writer,(q,))
po.apply(reader,(q,))
po.close()
po.join()
print("-------(%d) End-------"%os.getpid())運(yùn)行結(jié)果為：
-------(880) Start------- writer啟動(dòng)(7744),父進(jìn)程為(880) reader啟動(dòng)(7936),父進(jìn)程為(880) reader從Queue獲取到的消息時(shí)：S reader從Queue獲取到的消息時(shí)：e reader從Queue獲取到的消息時(shí)：7 reader從Queue獲取到的消息時(shí)：e reader從Queue獲取到的消息時(shí)：N reader從Queue獲取到的消息時(shí)：_ reader從Queue獲取到的消息時(shí)：H reader從Queue獲取到的消息時(shí)：O reader從Queue獲取到的消息時(shí)：U -------(880) End-------
相關(guān)推薦：
python中的進(jìn)程池是什么
新聞名稱：創(chuàng)新互聯(lián)Python教程：Python如何進(jìn)行進(jìn)程間的通信
標(biāo)題鏈接：http://www.5511xx.com/article/djpidgp.html
