python之路------進程間通訊，隊列和管道

進程間通訊

　　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函數來取得隊列中最早進入的數據。 接下來看一個稍微複雜一些的例子：

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()

批量生產數據放入隊列再批量獲取結果 x

生產者消費者模型

　　在併發編程中使用生產者和消費者模式可以解決絕大多數併發問題。該模式經過平衡生產線程和消費線程的工做能力來提升程序的總體處理數據的速度。

　　爲何要使用生產者和消費者模式

　　在線程世界裏，生產者就是生產數據的線程，消費者就是消費數據的線程。在多線程開發當中，若是生產者處理速度很快，而消費者處理速度很慢，那麼生產者就

　　必須等待消費者處理　完，才能繼續生產數據。一樣的道理，若是消費者的處理能力大於生產者，那麼消費者就必須等待生產者。爲了解決這個問題因而引入了生產者和消費者模式。

　　什麼是生產者消費者模式

　　生產者消費者模式是經過一個容器來解決生產者和消費者的強耦合問題。生產者和消費者彼此之間不直接通信，而經過阻塞隊列來進行通信，因此生產者生產完數據以後

　　不用等待消費者處理，直接扔給阻塞隊列，消費者不找生產者要數據，而是直接從阻塞隊列裏取，阻塞隊列就至關於一個緩衝區，平衡了生產者和消費者的處理能力。

　　基於隊列實現生產者消費者模型

基於隊列實現生產者消費者模型

　　此時的問題是主進程永遠不會結束，緣由是：生產者p在生產完後就結束了，可是消費者c在取空了q以後，則一直處於死循環中且卡在q.get()這一步。

　　解決方式無非是讓生產者在生產完畢後，往隊列中再發一個結束信號，這樣消費者在接收到結束信號後就能夠break出死循環。

改良版——生產者消費者模型

　　注意：結束信號None，不必定要由生產者發，主進程裏一樣能夠發，但主進程須要等生產者結束後才應該發送該信號

主進程在生產者生產完畢後發送結束信號None

　　但上述解決方式，在有多個生產者和多個消費者時，咱們則須要用一個很low的方式去解決

多個消費者的例子：有幾個消費者就須要發送幾回結束信號

　　JoinableQueue([maxsize]) 

　　建立可鏈接的共享進程隊列。這就像是一個Queue對象，但隊列容許項目的使用者通知生產者項目已經被成功處理。通知進程是使用共享的信號和條件變量來實現的。

JoinableQueue的實例p除了與Queue對象相同的方法以外，還具備如下方法：

q.task_done() 
使用者使用此方法發出信號，表示q.get()返回的項目已經被處理。若是調用此方法的次數大於從隊列中刪除的項目數量，將引起ValueError異常。

q.join() 
生產者將使用此方法進行阻塞，直到隊列中全部項目均被處理。阻塞將持續到爲隊列中的每一個項目均調用q.task_done()方法爲止。 
下面的例子說明如何創建永遠運行的進程，使用和處理隊列上的項目。生產者將項目放入隊列，並等待它們被處理。

JoinableQueue隊列實現消費之生產者模型

　　管道：

#建立管道的類：
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異常。

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異常。

　　所以，在生產者中關閉管道不會有任何效果，除非消費者也關閉了相同的管道端點。 

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()

引起EOFError

pipe實現生產者消費者模型

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('主進程')

多個消費之之間的競爭問題帶來的數據不安全問題

　　進程之間的數據共享

　　展望將來，基於消息傳遞的併發編程是大勢所趨

　　即使是使用線程，推薦作法也是將程序設計爲大量獨立的線程集合，經過消息隊列交換數據。

　　這樣極大地減小了對使用鎖定和其餘同步手段的需求，還能夠擴展到分佈式系統中。

　　但進程間應該儘可能避免通訊，即使須要通訊，也應該選擇進程安全的工具來避免加鎖帶來的問題。

　　之後咱們會嘗試使用數據庫來解決如今進程之間的數據共享問題。

進程間數據是獨立的，能夠藉助於隊列或管道實現通訊，兩者都是基於消息傳遞的
雖然進程間數據獨立，但能夠經過Manager實現數據共享，事實上Manager的功能遠不止於此

A manager object returned by Manager() controls a server process which holds Python objects and allows other processes to manipulate them using proxies.

A manager returned by Manager() will support types list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, 
Queue, Value and Array.

from multiprocessing import Manager,Process,Lock
def work(d,lock):
    with lock: #不加鎖而操做共享的數據,確定會出現數據錯亂
        d['count']-=1

if __name__ == '__main__':
    lock=Lock()
    with Manager() as m:
        dic=m.dict({'count':100})
        p_l=[]
        for i in range(100):
            p=Process(target=work,args=(dic,lock))
            p_l.append(p)
            p.start()
        for p in p_l:
            p.join()
        print(dic)

Manager例子

　　進程池和multiprocess.Pool模塊

　　進程池　

爲何要有進程池?進程池的概念。

　　在程序實際處理問題過程當中，忙時會有成千上萬的任務須要被執行，閒時可能只有零星任務。那麼在成千上萬個任務須要被執行的時候，咱們就須要去建立成千上萬個進程麼？

　　首先，建立進程須要消耗時間，銷燬進程也須要消耗時間。第二即使開啓了成千上萬的進程，操做系統也不能讓他們同時執行，這樣反而會影響程序的效率。

　　所以咱們不能無限制的根據任務開啓或者結束進程。那麼咱們要怎麼作呢？

　　在這裏，要給你們介紹一個進程池的概念，定義一個池子，在裏面放上固定數量的進程，有需求來了，就拿一個池中的進程來處理任務，等處處理完畢，進程並不關閉，

　　而是將進程再放回進程池中繼續等待任務。若是有不少任務須要執行，池中的進程數量不夠，任務就要等待以前的進程執行任務完畢歸來，拿到空閒進程才能繼續執行。

　　也就是說，池中進程的數量是固定的，那麼同一時間最多有固定數量的進程在運行。這樣不會增長操做系統的調度難度，還節省了開閉進程的時間，也必定程度上可以實現併發效果。

　　multiprocess.Pool模塊

　　概念介紹：

1 numprocess:要建立的進程數，若是省略，將默認使用cpu_count()的值
2 initializer：是每一個工做進程啓動時要執行的可調用對象，默認爲None
3 initargs：是要傳給initializer的參數組

1 p.apply(func [, args [, kwargs]]):在一個池工做進程中執行func(*args,**kwargs),而後返回結果。
2 '''須要強調的是：此操做並不會在全部池工做進程中並執行func函數。若是要經過不一樣參數併發地執行func函數，必須從不一樣線程調用p.apply()函數或者使用p.apply_async()'''
3 
4 p.apply_async(func [, args [, kwargs]]):在一個池工做進程中執行func(*args,**kwargs),而後返回結果。
5 '''此方法的結果是AsyncResult類的實例，callback是可調用對象，接收輸入參數。當func的結果變爲可用時，將理解傳遞給callback。callback禁止執行任何阻塞操做，
　　不然將接收其餘異步操做中的結果。'''
6    
7 p.close():關閉進程池，防止進一步操做。若是全部操做持續掛起，它們將在工做進程終止前完成
8 
9 P.jion():等待全部工做進程退出。此方法只能在close（）或teminate()以後調用

主要方法

1 方法apply_async()和map_async（）的返回值是AsyncResul的實例obj。實例具備如下方法
2 obj.get():返回結果，若是有必要則等待結果到達。timeout是可選的。若是在指定時間內尚未到達，將引起一場。若是遠程操做中引起了異常，它將在調用此方法時再次被引起。
3 obj.ready():若是調用完成，返回True
4 obj.successful():若是調用完成且沒有引起異常，返回True，若是在結果就緒以前調用此方法，引起異常
5 obj.wait([timeout]):等待結果變爲可用。
6 obj.terminate()：當即終止全部工做進程，同時不執行任何清理或結束任何掛起工做。若是p被垃圾回收，將自動調用此函數

其餘方法瞭解

　　代碼示例：

import time
import random
from multiprocessing import Pool,Process
def wahaha(i):
    print(i ** i)
    time.sleep(random.randint(1,5))
    return i ** i*'-'

if __name__ == '__main__':
    start = time.time()
    p = Pool(5)
    for i in range(101):
        p.apply_async(func=wahaha,args=(i,))   # 進程池異步提交了一個任務
    p.close()
    p.join()
    print(time.time() - start)


    start = time.time()
    p_lst = [Process(target=wahaha, args=(i,)) for i in range(101)]# 多進程提交任務
    for p in p_lst:p.start()
    for p in p_lst:p.join()
    print( time.time()- start)

多進程和進程池提交任務的效率對比

import time
import random
from multiprocessing import Pool,Process
def wahaha(i):
    print(i ** i)
    time.sleep(random.random())


if __name__ == '__main__':
    p = Pool(5)
    for i in range(101):
        p.apply_async(func=wahaha,args=(i,))   # 異步提交了一個任務
    p.close()
    p.join()

異步提交任務

import time
import random
from multiprocessing import Pool,Process
def wahaha(i):
    print(i ** i)
    time.sleep(random.random())


if __name__ == '__main__':
    p = Pool(5)
    for i in range(101):
        p.apply(func=wahaha,args=(i,))   # 同步提交了一個任務
    p.close()
    p.join()

同步提交任務

import time
import random
from multiprocessing import Pool,Process
def wahaha(i):
    print(i ** i)
    time.sleep(random.random())


if __name__ == '__main__':
    p = Pool(5)
    p.map(func=wahaha,iterable=range(101))

p.map進程池

import time
import random
from multiprocessing import Pool,Process
def wahaha(i):
    print(i ** i)
    time.sleep(random.random())
    return i ** i * '-'

if __name__ == '__main__':
    p = Pool(5)
    result_lst = []
    for i in range(101):
        r = p.apply_async(func=wahaha,args=(i,))   # 異步提交了一個任務
        result_lst.append(r)
    for r in result_lst:print(r.get())
    p.close()
    p.join()

異步進程池提交數據處理返回值

多進程和進程池的對比
對於純計算型的代碼 使用進程池更好 —— 真理
對於高IO的代碼 直接使用多進程更好 —— 相對論
結論 進程池比起多進程來講 節省了開啓進程回收進程資源的時間,給操做系統調度進程下降了難度

使用進程池提交任務
apply       # 同步提交任務 沒有多進程的優點

apply_async # 異步提交任務 經常使用，能夠經過get方法獲取返回值
close       # 關閉進程池，阻止往池中添加新的任務
join        # join依賴close，一個進程池必須先close再join

map #接收一個任務函數，和一個iterable。節省了for循環和close、join，是一種簡便的寫法

apply_async和map相比，操做複雜，可是能夠經過get方法獲取返回值

　　回調函數

　　須要回調函數的場景：進程池中任何一個任務一旦處理完了，就當即告知主進程：我好了額，你能夠處理個人結果了。主進程則調用一個函數去處理該結果，
該函數即回調函數。咱們能夠把耗時間（阻塞）的任務放到進程池中，而後指定回調函數（主進程負責執行），這樣主進程在執行回調函數時就省去了I/O的過程，
直接拿到的是任務的結果。

from multiprocessing import Process, Pool
import os


def wahaha(num):
    print('子', os.getpid())
    return num ** num


def call(argv):  # 回調函數用的是主進程中的資源
    print(os.getpid())
    print(argv)


if __name__ == '__main__':
    print('主', os.getpid())
    p = Pool(5)
    p.apply_async(func=wahaha, args=(50,), callback=call)  # callback是一個回調函數，接收一個函數地址
    p.close()
    p.join()

回調函數示例

　　注意：回調函數用的是主進程中的資源