程序間通訊-Queue

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

1. Queue的使⽤

可以使⽤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:
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

1）如果block使⽤預設值，且沒有設定timeout（單位秒），訊息列隊如果為空，此時程式將被阻塞（停在讀取狀態），直到從訊息列隊讀到訊息為止，如果設定了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（單位秒），訊息列隊如果已經沒有空間可寫⼊，此時程式將被阻塞（停在寫⼊狀態），直到從訊息列隊騰出空間為⽌，如果設定了timeout，則會等待timeout秒，若還沒空間，則丟擲"Queue.Full"異常；
2）如果block值為False，訊息列隊如果沒有空間可寫⼊，則會⽴刻丟擲"Queue.Full"異常；

Queue.put_nowait(item)：相當Queue.put(item, False)；

2. Queue例項

我們以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('所有資料都寫⼊並且讀完')

執行的結果是：

Put A to queue...
Put B to queue...
Put C to queue...
Get A from queue.
Get B from queue.
Get C from queue.
所有資料都寫⼊並且讀完
所有資料都寫⼊並且讀完

3. 程序池中的Queue

如果要使⽤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完全執⾏完po.apply(writer,(q,))

    po.apply(writer,(q,))
    po.apply(reader,(q,))
    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