程序間通訊 Queue

process之間有時需要通訊，作業系統提供了很多機制來實現程序間的通訊。

可以使用multiprocessing模組的queue實現多程序之間的資料傳遞，queue本身是乙個訊息列隊程式，首先用乙個小例項來演示一下queue的工作原理：

#coding=utf-8
from multiprocessing import queue
q=queue(3) #初始化乙個queue物件，最多可接收三條put訊息
q.put("訊息1") 
q.put("訊息2")
print(q.full())  #false
q.put("訊息3")
print(q.full()) #true
#因為訊息列隊已滿下面的try都會丟擲異常，第乙個try會等待2秒後再丟擲異常，第二個try會立刻丟擲異常
try:
q.put("訊息4",true,2)
except:
print("訊息列隊已滿，現有訊息數量:%s"%q.qsize())
try:
q.put_nowait("訊息4")
except:
print("訊息列隊已滿，現有訊息數量:%s"%q.qsize())
#推薦的方式，先判斷訊息列隊是否已滿，再寫入
if not q.full():
q.put_nowait("訊息4")
#讀取訊息時，先判斷訊息列隊是否為空，再讀取
if not q.empty():
for i in range(q.qsize()):
print(q.get_nowait())

執行結果:

false true 訊息列隊已滿，現有訊息數量:3 訊息列隊已滿，現有訊息數量:3 訊息1訊息2

訊息3

說明

初始化queue()物件時（例如：q=queue()），若括號中沒有指定最大可接收的訊息數量，或數量為負值，那麼就代表可接受的訊息數量沒有上限（直到記憶體的盡頭）；

1）如果block使用預設值，且沒有設定timeout（單位秒），訊息列隊如果為空，此時程式將被阻塞（停在讀取狀態），直到從訊息列隊讀到訊息為止，如果設定了timeout，則會等待timeout秒，若還沒讀取到任何訊息，則丟擲"queue.empty"異常；

2）如果block值為false，訊息列隊如果為空，則會立刻丟擲"queue.empty"異常；

1）如果block使用預設值，且沒有設定timeout（單位秒），訊息列隊如果已經沒有空間可寫入，此時程式將被阻塞（停在寫入狀態），直到從訊息列隊騰出空間為止，如果設定了timeout，則會等待timeout秒，若還沒空間，則丟擲"queue.full"異常；

2）如果block值為false，訊息列隊如果沒有空間可寫入，則會立刻丟擲"queue.full"異常；

我們以queue為例，在父程序中建立兩個子程序，乙個往queue裡寫資料，乙個從queue裡讀資料：

from multiprocessing import process, queue
import os, time, random
# 寫資料程序執行的**:
def write(q):
for value in ['a', 'b', 'c']:
print 'put %s to queue...' % value
q.put(value)
time.sleep(random.random())
# 讀資料程序執行的**:
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__':
# 父程序建立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()
# pr程序裡是死迴圈，無法等待其結束，只能強行終止:
print ''
print '所有資料都寫入並且讀完'

執行結果：

如果要使用pool建立程序，就需要使用multiprocessing.manager()中的queue()，而不是multiprocessing.queue()，否則會得到一條如下的錯誤資訊：

runtimeerror: queue objects should only be shared between processes through inheritance.

下面的例項演示了程序池中的程序如何通訊：

#coding=utf-8
#修改import中的queue為manager
from multiprocessing import manager,pool
import os,time,random
def reader(q):
print("reader啟動(%s),父程序為(%s)"%(os.getpid(),os.getppid()))
for i in range(q.qsize()):
print("reader從queue獲取到訊息：%s"%q.get(true))
def writer(q):
print("writer啟動(%s),父程序為(%s)"%(os.getpid(),os.getppid()))
for i in "dongge":
q.put(i)
if __name__=="__main__":
print("(%s) start"%os.getpid())
q=manager().queue() #使用manager中的queue來初始化
po=pool()
#使用阻塞模式建立程序，這樣就不需要在reader中使用死迴圈了，可以讓writer完全執行完成後，再用reader去讀取
po.close()
po.join()
print("(%s) end"%os.getpid())

執行結果:

(21156) start
writer啟動(21162),父程序為(21156)
reader啟動(21162),父程序為(21156)
reader從queue獲取到訊息：d
reader從queue獲取到訊息：o
reader從queue獲取到訊息：n
reader從queue獲取到訊息：g
reader從queue獲取到訊息：g
reader從queue獲取到訊息：e
(21156) end

程序間通訊 Queue

程序間通訊 Queue

程序間通訊 Queue

python程序間通訊 Queue

相關推薦