摘要：上一篇文章進程專題進程池下一篇文章進程專題共享數(shù)據(jù)與同步模塊支持的進程間通信主要有兩種管道和隊列。隊列底層使用管道和鎖，同時運行支持線程講隊列中的數(shù)據(jù)傳輸?shù)降讓庸艿乐?，來實?xí)進程間通信。

上一篇文章：Python進程專題4：進程池Pool
下一篇文章：Python進程專題6：共享數(shù)據(jù)與同步

multiprocessing模塊支持的進程間通信主要有兩種：管道和隊列。一般來說，發(fā)送較少的大對象比發(fā)送大量的小對象要好。

底層使用管道和鎖，同時運行支持線程講隊列中的數(shù)據(jù)傳輸?shù)降讓庸艿乐?，來實?xí)進程間通信。

語法：

Queue([maxsize])
創(chuàng)建共享隊列。使用multiprocessing模塊的Queue實現(xiàn)多進程之間的數(shù)據(jù)傳遞。Queue本身是一個消息隊列，
maxsize是隊列運行的最大項數(shù)，如果不指定，則不限制大小。

常用方法

q.close()：關(guān)閉隊列，不再向隊列中添加數(shù)據(jù)，那些已經(jīng)進入隊列的數(shù)據(jù)會繼續(xù)處理。q被回收時將自動調(diào)用此方法。

q.empty():如果調(diào)用此方法時，隊列為null返回True，單由于此時其他進程或者線程正在添加或刪除項，
所以結(jié)果不可靠，而且有些平臺運行該方法會直接報錯，我的mac系統(tǒng)運行該方法，直接報錯。

q.full():如果調(diào)用此方法時，隊列已滿，返回True，同q.empty()方法，結(jié)果不可靠。

q.get([block,timeout])：返回q中的一個項，block如果設(shè)置為True,如果q隊列為空，該方法會阻塞（就是不往下運行了，處于等待狀態(tài)），
直到隊列中有項可用為止，如果同時頁設(shè)置了timeout，那么在改時間間隔內(nèi)，都沒有等到有用的項，就會引發(fā)Queue.Empty異常。
如果block設(shè)置為false，timeout沒有意義，如果隊列為空，將引發(fā)Queue.Empt異常。

q.get_nowait():等同于q.get(False)

q.put(item,block,timeout):將item放入隊列，如果此時隊列已滿：
如果block=True，timeout沒有設(shè)置，就會阻塞，直到有可用空間為止。
如果block=True，timeout也設(shè)置，就會阻塞到timeout，超過這個時間會報Queue.Full異常。
如果block=False，timeout設(shè)置無效，直接報Queue.Full異常。

q.put_nowait(item):等同于q.put(item,False)

q.qsize():返回當前隊列項的數(shù)量，結(jié)果不可靠，而且mac會直接報錯：NotImplementedError。

實例1：驗證：put方法會阻塞

實例：

#驗證：put方法會阻塞
from multiprocessing import Queue
queue=Queue(3)#初始化一個Queue隊列，可以接受3個消息
queue.put("我是第1條信息")
queue.put("我是第2條信息")
queue.put("我是第3條信息")

print("插入第4條信息之前")
queue.put("我是第4條信息")
print("插入第4條信息之后")

效果：程序會一直阻塞，最后一句輸永遠也不會輸出。

實例2：closse方法、get方法、put方法簡單使用：多進程訪問同一個Queue

代碼：

#closse方法、get方法、put方法簡單使用：多進程訪問同一個Queue
from multiprocessing import Queue,Process
import time,os
#參數(shù)q就是Queue實例
def mark(q,interval):
    time.sleep(interval)
    # 打印信息
    print("進程%d取出數(shù)據(jù)："%os.getpid()+queue.get(True))


if __name__=="__main__":
    queue = Queue(3)  # 初始化一個Queue隊列，可以接受3個消息
    queue.put("我是第1條信息")
    queue.put("我是第2條信息")
    queue.put("我是第3條信息")

    p1=Process(target=mark,args=(queue,1))
    p2=Process(target=mark,args=(queue,2))
    p3=Process(target=mark,args=(queue,3))

    p1.start()
    p2.start()
    p3.start()
    # 關(guān)閉隊列，不再插入信息
    queue.close()
    # 下面插入會導(dǎo)致異常
    # queue.put("我是第4條信息")

    # 打印第1條信息
    print("程序語句執(zhí)行完成")

效果

創(chuàng)建可連接的共享進程隊列，可以看做還是一個Queue，只不過這個Queue除了Queue特有功能外，允許項的消費者通知項的生產(chǎn)者，項已經(jīng)處理成功。該通知進程時使用共享的信號和條件變量來實現(xiàn)的。

JoinableQueue實例除了與Queue對象相同的方法外，還具有下列方法：

q.task_done():消費者使用此方法發(fā)送信號，表示q.get()返回的項已經(jīng)被處理。
注意??：如果調(diào)用此方法的次數(shù)大于隊列中刪除的項的數(shù)量，將引發(fā)ValueError異常。

q.join():生產(chǎn)者使用此方法進行阻塞，直到隊列中所有的項都被處理完成，即阻塞將持續(xù)到隊列中的每一項均調(diào)用q.task_done()方法為止。

代碼實例：

#利用JoinableQueue實現(xiàn)生產(chǎn)者與消費者，并且加入了哨兵，來監(jiān)聽生產(chǎn)者的要求
from multiprocessing import JoinableQueue,Process
import time

#參數(shù)q為JoinableQueue隊列實例
def mark(q):
    #循環(huán)接受信息,一直運行，這也下面為什么要將它設(shè)為后臺進程的原因，必須保證當主線程退出時，它可以退出
    while True:
        value = q.get()
        print(value)  # 實際開發(fā)過程中，此處一般用來進行有用的處理
        # 消費者發(fā)送信號：任務(wù)完成（此處實例的任務(wù)就是打印一下下）
        q.task_done()
        #我來方便看出效果，特意停留1s
        time.sleep(1)

        #使用哨兵，監(jiān)聽生產(chǎn)者的消息，此處通過判斷value是否為None來判斷傳遞的消息
        if value==None:
            #執(zhí)行哨兵計劃后，后面的語句都不會輸出
            break




if __name__=="__main__":
    #實例化JoinableQueue
    q=JoinableQueue()

    #定義消費者進程
    p=Process(target=mark,args=(q,))
    #將消費者線程設(shè)置為后臺進程，隨創(chuàng)建它的進程（此處是主進程）的終止而終止
    #也就是當它的創(chuàng)建進程（此處是主現(xiàn)場）意外退出時，它也會跟隨一起退出。
    #并且后臺進程無法創(chuàng)建新的進程
    p.daemon=True

    #啟動消費者進程
    p.start()

    #模擬生產(chǎn)者，生產(chǎn)多個項
    for xx in range(5):
        print(xx)
        #當xx==3時去執(zhí)行哨兵計劃
        if xx==3:
            print("我用哨兵計劃了")
            q.put(None)
            print("哨兵計完美執(zhí)行")
        q.put("第%d條消息"%xx)


    #等待所有項都處理完成再退出，由于使用了哨兵計劃，隊列沒有完全執(zhí)行，所以會一直卡在這個位置
    q.join()

    print("程序真正退出了")

效果：

除了使用隊列來進行進程間通信，還可以使用管道來進行消息傳遞。

語法：

(connection1,connection2)=Pipe([duplex])
在進程間創(chuàng)建一條管道，并返回元祖(connection1,connection2),其中connection1、connection2表示兩端的Connection對象。
默認情況下，duplex=True，此時管道是雙向的，如果設(shè)置duplex=false，connection1只能用于接收，connection2只能用于發(fā)送。
注意：必須在多進程創(chuàng)建之前創(chuàng)建管道。

常用方法：

connection.close() :關(guān)閉連接，當connection被垃圾回收時，默認會調(diào)用該方法。

connection.fileno() :返回連接使用的整數(shù)文件描述符

connection.poll([timeout]):如果連接上的數(shù)據(jù)可用，返回True，timeout為等待的最長時間，如果不指定，該方法將立刻返回結(jié)果。
如果指定為None，該方法將會無限等待直到數(shù)據(jù)到達。

connection.send(obj):通過連接發(fā)送對象，obj是與序列號兼容的任意對象。

connection.send_bytes(buffer[,offset,size]):通過連接發(fā)送字節(jié)數(shù)據(jù)緩沖區(qū)，buffer是支持緩沖區(qū)的任何對象。
offset是緩沖區(qū)的字節(jié)偏移量，而size是要發(fā)送的字節(jié)數(shù)。

connection.recv():接收connection.send()方法返回的對象。如果連接的另一端已經(jīng)關(guān)閉，再也不會存在任何數(shù)據(jù)，
該方法將引起EOFError異常。

connection.recv_bytes([maxlength]):接收connection.send_bytes()方法發(fā)送的一條完整字節(jié)信息，maxlength為可以接受的
最大字節(jié)數(shù)。如果消息超過這個最大數(shù)，將引發(fā)IOError異常，并且在連接上無法進一步讀取。如果連接的另一端已經(jīng)關(guān)閉，
再也不會有任何數(shù)據(jù)，該方法將引發(fā)EOFError異常。

connection.recv_bytes_into(buffer[,offset]):接收一條完整的字節(jié)信息，兵把它保存在buffer對象中，
該對象支持可寫入的緩沖區(qū)接口（就是bytearray對象或類似對象）。
offset指定緩沖區(qū)放置消息的字節(jié)偏移量。返回值是收到的字節(jié)數(shù)。如果消息長度大于可用的緩沖區(qū)空間，將引發(fā)BufferTooShort異常。

實例1：理解管道的生產(chǎn)者與消費者

示意圖：

代碼：

#理解管道的生產(chǎn)者與消費者
from multiprocessing import Pipe, Process
import time

def mark(pipe):
    #接受參數(shù)
    output_p, input_p = pipe
    print("mark方法內(nèi)部調(diào)用input_p.close()")
    #消費者（子進程）此實例只接收，所以把輸入關(guān)閉
    input_p.close()

    while True:
        try:
            item = output_p.recv()
        except EOFError:
            print("報錯了")
            break
        print(item)
        time.sleep(1)
    print("mark執(zhí)行完成")


if __name__ == "__main__":
    #必須在多進程創(chuàng)建之前，創(chuàng)建管道,該管道是雙向的
    (output_p, input_p) = Pipe()#創(chuàng)建管道
    #創(chuàng)建一個進程，并把管道兩端都作為參數(shù)傳遞過去
    p = Process(target=mark, args=((output_p, input_p),))
    #啟動進程
    p.start()
    #生產(chǎn)者（主進程）此實例只輸入，所以關(guān)閉輸出（接收端）
    output_p.close()

    for item in list(range(5)):
        input_p.send(item)
    print("主方法內(nèi)部調(diào)用input_p.close()()")
    #關(guān)閉生產(chǎn)者（主進程）的輸入端
    input_p.close()

效果圖：

實例2：利用管道實現(xiàn)多進程協(xié)作：子線程計算結(jié)果，返回給主線程

代碼：

#利用管道實現(xiàn)多進程協(xié)作：子線程計算結(jié)果，返回給主線程
from multiprocessing import Pipe, Process

def mark(pipe):
    #接受參數(shù)
    server_p, client_p = pipe
    #消費者（子進程）此實例只接收，所以把輸入關(guān)閉
    client_p.close()

    while True:
        try:
            x,y = server_p.recv()
        except EOFError:
            print("報錯了")
            break
        result=x+y
        server_p.send(result)
    print("mark執(zhí)行完成")


if __name__ == "__main__":
    #必須在多進程創(chuàng)建之前，創(chuàng)建管道,該管道是雙向的
    (server_p, client_p) = Pipe()#創(chuàng)建管道
    #創(chuàng)建一個進程，并把管道兩端都作為參數(shù)傳遞過去
    p = Process(target=mark, args=((server_p, client_p),))
    #啟動進程
    p.start()
    #生產(chǎn)者（主進程）此實例只輸入，所以關(guān)閉輸出（接收端）
    server_p.close()

    #發(fā)送數(shù)據(jù)
    client_p.send((4,5))
    #打印接受到的數(shù)據(jù)
    print(client_p.recv())

    client_p.send(("Mark", "大帥哥"))
    # 打印接受到的數(shù)據(jù)
    print(client_p.recv())





    #關(guān)閉生產(chǎn)者（主進程）的輸入端
    client_p.close()

結(jié)果：

9
Mark大帥哥
報錯了
mark執(zhí)行完成