#### 進程間通信---Queue Process之間有時需要通信,操作系統提供了很多機制來時間進程間的通信. ##### 1.Queue的使用 可以使用multiprocessing模塊的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()），若括號中沒有指定最大可接收的消息數量，或數量為負值，那么就代表可接受的消息數量沒有上限（直到內存的盡頭）； * Queue.qsize()：返回當前隊列包含的消息數量； * Queue.empty()：如果隊列為空，返回True，反之False ； * Queue.full()：如果隊列滿了，返回True,反之False； * Queue.get([block[, timeout]])：獲取隊列中的一條消息，然后將其從列隊中移除，block默認值為True； 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 '所有數據都寫入并且讀完' ~~~ 運行結果:  ##### 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完全執行完成后，再用reader去讀取 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 ~~~