程序（四）：程序間通訊 —— Queue（佇列）和Pipe（管道）
目錄
程序間通訊
IPC(Inter-Process Communication)
佇列
概念介紹
建立共享的程序佇列，Queue是多程序安全的佇列，可以使用Queue實現多程序之間的資料傳遞。
●Queue([maxsize]) 建立共享的程序佇列。
●引數 ：maxsize是佇列中允許的最大項數。如果省略此引數，則無大小限制。
●底層佇列使用管道和鎖定實現。
方法介紹
Queue([maxsize])
建立共享的程序佇列。maxsize是佇列中允許的最大項數。如果省略此引數，則無大小限制。底層佇列使用管道和鎖定實現。另外，還需要執行支援執行緒以便佇列中的資料傳輸到底層管道中。
Queue的例項q具有以下方法：
q.get( [ block [ ,timeout ] ] )
返回q中的一個專案。如果q為空，此方法將阻塞，直到佇列中有專案可用為止。block用於控制阻塞行為，預設為True. 如果設定為False，將引發Queue.Empty異常（定義在Queue模組中）。timeout是可選超時時間，用在阻塞模式中。如果在制定的時間間隔內沒有專案變為可用，將引發Queue.Empty異常。
q.get_nowait( )
同q.get(False)方法。
q.put(item [, block [,timeout ] ] )
將item放入佇列。如果佇列已滿，此方法將阻塞至有空間可用為止。block控制阻塞行為，預設為True。如果設定為False，將引發Queue.Empty異常（定義在Queue庫模組中）。timeout指定在阻塞模式中等待可用空間的時間長短。超時後將引發Queue.Full異常。
q.qsize()
返回佇列中目前專案的正確數量。此函式的結果並不可靠，因為在返回結果和在稍後程式中使用結果之間，佇列中可能新增或刪除了專案。在某些系統上，此方法可能引發NotImplementedError異常。
q.empty()
如果呼叫此方法時 q為空，返回True。如果其他程序或執行緒正在往佇列中新增專案，結果是不可靠的。也就是說，在返回和使用結果之間，佇列中可能已經加入新的專案。
q.full()
如果q已滿，返回為True. 由於執行緒的存在，結果也可能是不可靠的（參考q.empty（）方法）。。
其他方法(瞭解)
q.close()
關閉佇列，防止佇列中加入更多資料。呼叫此方法時，後臺執行緒將繼續寫入那些已入佇列但尚未寫入的資料，但將在此方法完成時馬上關閉。如果q被垃圾收集，將自動呼叫此方法。關閉佇列不會在佇列使用者中生成任何型別的資料結束訊號或異常。例如，如果某個使用者正被阻塞在get（）操作上，關閉生產者中的佇列不會導致get（）方法返回錯誤。
q.cancel_join_thread()
不會再程序退出時自動連線後臺執行緒。這可以防止join_thread()方法阻塞。
q.join_thread()
連線佇列的後臺執行緒。此方法用於在呼叫q.close()方法後，等待所有佇列項被消耗。預設情況下，此方法由不是q的原始建立者的所有程序呼叫。呼叫q.cancel_join_thread()方法可以禁止這種行為。
程式碼例項
《單看佇列用法》
'''
multiprocessing模組支援程序間通訊的兩種主要形式:管道和佇列
都是基於訊息傳遞實現的,但是佇列介面
'''
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) # 如果佇列已經滿了，程式就會停在這裡，等待資料被別人取走，再將資料放入佇列。
# 如果佇列中的資料一直不被取走，程式就會永遠停在這裡。
try:
q.put_nowait(3) # 可以使用put_nowait，如果佇列滿了不會阻塞，但是會因為佇列滿了而報錯。
except: # 因此我們可以用一個try語句來處理這個錯誤。這樣程式不會一直阻塞下去，但是會丟掉這個訊息。
print('佇列已經滿了')
# 因此，我們再放入資料之前，可以先看一下佇列的狀態，如果已經滿了，就不繼續put了。
print(q.full()) #滿了
print(q.get())
print(q.get())
print(q.get())
# print(q.get()) # 同put方法一樣，如果佇列已經空了，那麼繼續取就會出現阻塞。
try:
q.get_nowait(3) # 可以使用get_nowait，如果佇列滿了不會阻塞，但是會因為沒取到值而報錯。
except: # 因此我們可以用一個try語句來處理這個錯誤。這樣程式不會一直阻塞下去。
print('佇列已經空了')
print(q.empty()) #空了
上面這個例子還沒有加入程序通訊，只是先來看看佇列為我們提供的方法，以及這些方法的使用和現象。
《子程序傳送資料給父程序》
import time
from multiprocessing import Process, Queue
def f(q):
q.put([time.asctime(), 'from Eva', 'hello']) #呼叫主函式中p程序傳遞過來的程序引數 put函式為向佇列中新增一條資料。
if __name__ == '__main__':
q = Queue() #建立一個Queue物件
p = Process(target=f, args=(q,)) #建立一個程序
p.start()
print(q.get())
p.join()
上面是一個queue的簡單應用，使用佇列q物件呼叫get函式來取得佇列中最先進入的資料。 接下來看一個稍微複雜一些的例子：
《批量生產資料放入佇列再批量獲取結果 x》
import os
import time
import multiprocessing
# 向queue中輸入資料的函式
def inputQ(queue):
info = str(os.getpid()) + '(put):' + str(time.asctime())
queue.put(info)
# 向queue中輸出資料的函式
def outputQ(queue):
info = queue.get()
print ('%s%s\033[32m%s\033[0m'%(str(os.getpid()), '(get):',info))
# Main
if __name__ == '__main__':
multiprocessing.freeze_support()
record1 = [] # store input processes
record2 = [] # store output processes
queue = multiprocessing.Queue(3)
# 輸入程序
for i in range(10):
process = multiprocessing.Process(target=inputQ,args=(queue,))
process.start()
record1.append(process)
# 輸出程序
for i in range(10):
process = multiprocessing.Process(target=outputQ,args=(queue,))
process.start()
record2.append(process)
for p in record1:
p.join()
for p in record2:
p.join()
生產者消費者模型
在併發程式設計中使用生產者和消費者模式能夠解決絕大多數併發問題。該模式通過平衡生產執行緒和消費執行緒的工作能力來提高程式的整體處理資料的速度。
為什麼要使用生產者和消費者模式
線上程世界裡，生產者就是生產資料的執行緒，消費者就是消費資料的執行緒。在多執行緒開發當中，如果生產者處理速度很快，而消費者處理速度很慢，那麼生產者就必須等待消費者處理完，才能繼續生產資料。同樣的道理，如果消費者的處理能力大於生產者，那麼消費者就必須等待生產者。為了解決這個問題於是引入了生產者和消費者模式。
什麼是生產者消費者模式
生產者消費者模式是通過一個容器來解決生產者和消費者的強耦合問題。生產者和消費者彼此之間不直接通訊，而通過阻塞佇列來進行通訊，所以生產者生產完資料之後不用等待消費者處理，直接扔給阻塞佇列，消費者不找生產者要資料，而是直接從阻塞佇列裡取，阻塞佇列就相當於一個緩衝區，平衡了生產者和消費者的處理能力。
基於佇列實現生產者消費者模型
基於佇列實現生產者消費者模型
此時的問題是主程序永遠不會結束，原因是：生產者p在生產完後就結束了，但是消費者c在取空了q之後，則一直處於死迴圈中且卡在q.get()這一步。
解決方式無非是讓生產者在生產完畢後，往佇列中再發一個結束訊號，這樣消費者在接收到結束訊號後就可以break出死迴圈。
《基於佇列實現生產者消費者模型》
from multiprocessing import Process,Queue
import time,random,os
def consumer(q):
while True:
res=q.get()
time.sleep(random.randint(1,3))
print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res))
def producer(q):
for i in range(10):
time.sleep(random.randint(1,3))
res='包子%s' %i
q.put(res)
print('\033[44m%s 生產了 %s\033[0m' %(os.getpid(),res))
if __name__ == '__main__':
q=Queue()
#生產者們:即廚師們
p1=Process(target=producer,args=(q,))
#消費者們:即吃貨們
c1=Process(target=consumer,args=(q,))
#開始
p1.start()
c1.start()
print('主')
此時的問題是主程序永遠不會結束，原因是：生產者p在生產完後就結束了，但是消費者c在取空了q之後，則一直處於死迴圈中且卡在q.get()這一步。
解決方式無非是讓生產者在生產完畢後，往佇列中再發一個結束訊號，這樣消費者在接收到結束訊號後就可以break出死迴圈。
《改良版——生產者消費者模型》
from multiprocessing import Process,Queue
import time,random,os
def consumer(q):
while True:
res=q.get()
if res is None:break #收到結束訊號則結束
time.sleep(random.randint(1,3))
print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res))
def producer(q):
for i in range(10):
time.sleep(random.randint(1,3))
res='包子%s' %i
q.put(res)
print('\033[44m%s 生產了 %s\033[0m' %(os.getpid(),res))
q.put(None) #傳送結束訊號
if __name__ == '__main__':
q=Queue()
#生產者們:即廚師們
p1=Process(target=producer,args=(q,))
#消費者們:即吃貨們
c1=Process(target=consumer,args=(q,))
#開始
p1.start()
c1.start()
print('主')
注意：結束訊號None，不一定要由生產者發，主程序裡同樣可以發，但主程序需要等生產者結束後才應該傳送該訊號
《主程序在生產者生產完畢後傳送結束訊號None》
from multiprocessing import Process,Queue
import time,random,os
def consumer(q):
while True:
res=q.get()
if res is None:break #收到結束訊號則結束
time.sleep(random.randint(1,3))
print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res))
def producer(q):
for i in range(2):
time.sleep(random.randint(1,3))
res='包子%s' %i
q.put(res)
print('\033[44m%s 生產了 %s\033[0m' %(os.getpid(),res))
if __name__ == '__main__':
q=Queue()
#生產者們:即廚師們
p1=Process(target=producer,args=(q,))
#消費者們:即吃貨們
c1=Process(target=consumer,args=(q,))
#開始
p1.start()
c1.start()
p1.join()
q.put(None) #傳送結束訊號
print('主')
但上述解決方式，在有多個生產者和多個消費者時，我們則需要用一個很low的方式去解決
《多個消費者的例子：有幾個消費者就需要傳送幾次結束訊號》
from multiprocessing import Process,Queue
import time,random,os
def consumer(q):
while True:
res=q.get()
if res is None:break #收到結束訊號則結束
time.sleep(random.randint(1,3))
print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res))
def producer(name,q):
for i in range(2):
time.sleep(random.randint(1,3))
res='%s%s' %(name,i)
q.put(res)
print('\033[44m%s 生產了 %s\033[0m' %(os.getpid(),res))
if __name__ == '__main__':
q=Queue()
#生產者們:即廚師們
p1=Process(target=producer,args=('包子',q))
p2=Process(target=producer,args=('骨頭',q))
p3=Process(target=producer,args=('泔水',q))
#消費者們:即吃貨們
c1=Process(target=consumer,args=(q,))
c2=Process(target=consumer,args=(q,))
#開始
p1.start()
p2.start()
p3.start()
c1.start()
p1.join() #必須保證生產者全部生產完畢,才應該傳送結束訊號
p2.join()
p3.join()
q.put(None) #有幾個消費者就應該傳送幾次結束訊號None
q.put(None) #傳送結束訊號
print('主')
JoinableQueue([maxsize])
建立可連線的共享程序佇列。這就像是一個Queue物件，但佇列允許專案的使用者通知生產者專案已經被成功處理。通知程序是使用共享的訊號和條件變數來實現的。
方法介紹
JoinableQueue的例項p除了與Queue物件相同的方法之外，還具有以下方法：
●q.task_done()
使用者使用此方法發出訊號，表示q.get()返回的專案已經被處理。如果呼叫此方法的次數大於從佇列中刪除的專案數量，將引發ValueError異常。
●q.join()
生產者將使用此方法進行阻塞，直到佇列中所有專案均被處理。阻塞將持續到為佇列中的每個專案均呼叫q.task_done()方法為止。
下面的例子說明如何建立永遠執行的程序，使用和處理佇列上的專案。生產者將專案放入佇列，並等待它們被處理。
《JoinableQueue佇列實現消費之生產者模型》
from multiprocessing import Process,JoinableQueue
import time,random,os
def consumer(q):
while True:
res=q.get()
time.sleep(random.randint(1,3))
print('\033[45m%s 吃 %s\033[0m' %(os.getpid(),res))
q.task_done() #向q.join()傳送一次訊號,證明一個數據已經被取走了
def producer(name,q):
for i in range(10):
time.sleep(random.randint(1,3))
res='%s%s' %(name,i)
q.put(res)
print('\033[44m%s 生產了 %s\033[0m' %(os.getpid(),res))
q.join() #生產完畢，使用此方法進行阻塞，直到佇列中所有專案均被處理。
if __name__ == '__main__':
q=JoinableQueue()
#生產者們:即廚師們
p1=Process(target=producer,args=('包子',q))
p2=Process(target=producer,args=('骨頭',q))
p3=Process(target=producer,args=('泔水',q))
#消費者們:即吃貨們
c1=Process(target=consumer,args=(q,))
c2=Process(target=consumer,args=(q,))
c1.daemon=True
c2.daemon=True
#開始
p_l=[p1,p2,p3,c1,c2]
for p in p_l:
p.start()
p1.join()
p2.join()
p3.join()
print('主')
#主程序等--->p1,p2,p3等---->c1,c2
#p1,p2,p3結束了,證明c1,c2肯定全都收完了p1,p2,p3發到佇列的資料
#因而c1,c2也沒有存在的價值了,不需要繼續阻塞在程序中影響主程序了。應該隨著主程序的結束而結束,所以設定成守護程序就可以了。
管道(瞭解)
介紹：
#建立管道的類：
Pipe([duplex]):在程序之間建立一條管道，並返回元組（conn1,conn2）,其中conn1，conn2表示管道兩端的連線物件，強調一點：必須在產生Process物件之前產生管道
#引數介紹：
dumplex:預設管道是全雙工的，如果將duplex射成False，conn1只能用於接收，conn2只能用於傳送。
#主要方法：
conn1.recv():接收conn2.send(obj)傳送的物件。如果沒有訊息可接收，recv方法會一直阻塞。如果連線的另外一端已經關閉，那麼recv方法會丟擲EOFError。
conn1.send(obj):通過連線傳送物件。obj是與序列化相容的任意物件
#其他方法：
conn1.close():關閉連線。如果conn1被垃圾回收，將自動呼叫此方法
conn1.fileno():返回連線使用的整數檔案描述符
conn1.poll([timeout]):如果連線上的資料可用，返回True。timeout指定等待的最長時限。如果省略此引數，方法將立即返回結果。如果將timeout射成None，操作將無限期地等待資料到達。
conn1.recv_bytes([maxlength]):接收c.send_bytes()方法傳送的一條完整的位元組訊息。maxlength指定要接收的最大位元組數。如果進入的訊息，超過了這個最大值，將引發IOError異常，並且在連線上無法進行進一步讀取。如果連線的另外一端已經關閉，再也不存在任何資料，將引發EOFError異常。
conn.send_bytes(buffer [, offset [, size]])：通過連線傳送位元組資料緩衝區，buffer是支援緩衝區介面的任意物件，offset是緩衝區中的位元組偏移量，而size是要傳送位元組數。結果資料以單條訊息的形式發出，然後呼叫c.recv_bytes()函式進行接收
conn1.recv_bytes_into(buffer [, offset]):接收一條完整的位元組訊息，並把它儲存在buffer物件中，該物件支援可寫入的緩衝區介面（即bytearray物件或類似的物件）。offset指定緩衝區中放置訊息處的位元組位移。返回值是收到的位元組數。如果訊息長度大於可用的緩衝區空間，將引發BufferTooShort異常。
《pipe初使用》
from multiprocessing import Process, Pipe
def f(conn):
conn.send("Hello The_Third_Wave")
conn.close()
if __name__ == '__main__':
parent_conn, child_conn = Pipe()
p = Process(target=f, args=(child_conn,))
p.start()
print(parent_conn.recv())
p.join()
應該特別注意管道端點的正確管理問題。如果是生產者或消費者中都沒有使用管道的某個端點，就應將它關閉。這也說明了為何在生產者中關閉了管道的輸出端，在消費者中關閉管道的輸入端。如果忘記執行這些步驟，程式可能在消費者中的recv（）操作上掛起。管道是由作業系統進行引用計數的，必須在所有程序中關閉管道後才能生成EOFError異常。因此，在生產者中關閉管道不會有任何效果，除非消費者也關閉了相同的管道端點。
#引發EOFError
from multiprocessing import Process, Pipe
def f(parent_conn,child_conn):
#parent_conn.close() #不寫close將不會引發EOFError
while True:
try:
print(child_conn.recv())
except EOFError:
child_conn.close()
if __name__ == '__main__':
parent_conn, child_conn = Pipe()
p = Process(target=f, args=(parent_conn,child_conn,))
p.start()
child_conn.close()
parent_conn.send('hello')
parent_conn.close()
p.join()
#pipe實現生產者消費者模型
from multiprocessing import Process,Pipe
def consumer(p,name):
produce, consume=p
produce.close()
while True:
try:
baozi=consume.recv()
print('%s 收到包子:%s' %(name,baozi))
except EOFError:
break
def producer(seq,p):
produce, consume=p
consume.close()
for i in seq:
produce.send(i)
if __name__ == '__main__':
produce,consume=Pipe()
c1=Process(target=consumer,args=((produce,consume),'c1'))
c1.start()
seq=(i for i in range(10))
producer(seq,(produce,consume))
produce.close()
consume.close()
c1.join()
print('主程序')
#多個消費之之間的競爭問題帶來的資料不安全問題
from multiprocessing import Process,Pipe,Lock
def consumer(p,name,lock):
produce, consume=p
produce.close()
while True:
lock.acquire()
baozi=consume.recv()
lock.release()
if baozi:
print('%s 收到包子:%s' %(name,baozi))
else:
consume.close()
break
def producer(p,n):
produce, consume=p
consume.close()
for i in range(n):
produce.send(i)
produce.send(None)
produce.send(None)
produce.close()
if __name__ == '__main__':
produce,consume=Pipe()
lock = Lock()
c1=Process(target=consumer,args=((produce,consume),'c1',lock))
c2=Process(target=consumer,args=((produce,consume),'c2',lock))
p1=Process(target=producer,args=((produce,consume),10))
c1.start()
c2.start()
p1.start()
produce.close()
consume.close()
c1.join()
c2.join()
p1.join()
print('主程序')