?

一、管道

概念

管道可用于具有親緣關系進程間的通信，有名管道克服了管道沒有名字的限制，因此，除具有管道所具有的功能外，它還允許無親緣關系進程間的通信.

先畫一幅圖幫助大家理解下管道的基本原理

?

? 現有2個進程A和B，他們都在內存中開辟了空間，那么我們在內存中再開辟一個空間C，作用是連接這兩個進程的。對于進程來說內存空間是可以共享的（任何一個進程都可以使用內存，內存當中的空間是用地址來標記的，我們通過查找某一個地址就能找到這個內存）A進程可以不斷的向C空間輸送東西，B進程可以不斷的從C空間讀取東西，這就是進程間的通信? ?.

? 管道在信息傳輸上是以流的方式傳輸， 也就是你從A進程不斷的寫入，B進程源源不斷的讀出，A進程先寫入的就會被B進程先讀出，后寫進來的就會被后讀出，

Pipe僅僅適用于只有兩個進程一讀一寫的半雙工情況，也就是說信息是只向一個方向流動。單項通信叫做半雙工，雙向叫做全雙工.

單工：簡單的說就是一方只能發信息，另一方則只能收信息，通信是單向的。

半雙工：比單工先進一點，就是雙方都能發信息，但同一時間則只能一方發信息。

全雙工：比半雙工再先進一點，就是雙方不僅都能發信息，而且能夠同時發送。

實現機制：

? 管道是由內核管理的一個緩沖區，相當于我們放入內存中的一個紙條。管道的一端連接一個進程的輸出。這個進程會向管道中放入信息。管道的另一端連接一個進程的輸入，這個進程取出被放入管道的信息。一個緩沖區不需要很大，它被設計成為環形的數據結構，以便管道可以被循環利用。當管道中沒有信息的話，從管道中讀取的進程會等待，直到另一端的進程放入信息。當管道被放滿信息的時候，嘗試放入信息的進程會等待，直到另一端的進程取出信息。當兩個進程都終結的時候，管道也自動消失。

管道特點

管道是單向的、先進先出的、無結構的字節流，它把一個進程的輸出和另一個進程的輸入連接在一起。

• 寫進程在管道的尾端寫入數據，讀進程在管道的首端讀出數據。數據讀出后將從管道中移走，其它讀進程都不能再讀到這些數據。

• 管道提供了簡單的流控制機制。進程試圖讀一個空管道時，在數據寫入管道前，進程將一直阻塞。同樣，管道已經滿時，進程再試圖寫管道，在其它進程從管道中讀走數據之前，寫進程將一直阻塞。

匿名管道具有的特點：

• 只能用于具有親緣關系的進程之間的通信（也就是父子進程或者兄弟進程之間）。

• 一種半雙工的通信模式，具有固定的讀端和寫端。

• LINUX把管道看作是一種文件，采用文件管理的方法對管道進行管理，對于它的讀寫也可以使用普通的read()和write()等函數。但是它不是普通的文件，并不屬于其他任何文件系統，只存在于內核的內存空間中。

參數介紹

#創建管道的類：
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 Pipe,Process?def func(out_pipe, in_pipe):in_pipe.close()# 關閉復制過來的管道的輸入端while True:try :msg = out_pipe.recv() #子進程的管道端口接收主進程的數據print(msg)except EOFError:out_pipe.close()breakif __name__ == '__main__':out_pipe, in_pipe = Pipe()Process(target=func,args = (out_pipe, in_pipe)).start() #啟動子進程out_pipe.close() #關閉主進程的輸出管道端口for i in range(20):in_pipe.send('hello world!') #通過管道的端口向子進程寫入in_pipe.close()

　　

例子2

# 出現EOF錯誤的情況# 當pipe的輸入端被關閉，且無法接收到輸入的值，那么就會拋出EOFError。?from multiprocessing import Pipe, Process??def func(out_pipe, in_pipe):in_pipe.close()# 關閉復制過來的管道的輸入端while True:?msg = out_pipe.recv() # 子進程的管道端口接收主進程的數據print(msg)??if __name__ == '__main__':out_pipe, in_pipe = Pipe()Process(target=func, args=(out_pipe, in_pipe)).start() # 啟動子進程out_pipe.close() # 關閉主進程的輸出管道端口for i in range(20):in_pipe.send('hello world!') # 通過管道的端口向子進程寫入in_pipe.close()

　　

基于管道實現生產者消費者模型

from multiprocessing import Process,Pipe?import time,random??def consumer(p,name):in_pipe,out_pipe=pout_pipe.close()while True:try:# time.sleep(random.uniform(0,1))baozi=in_pipe.recv()print('%s 收到包子:%s' %(name,baozi))except EOFError:in_pipe.close()breakdef producer(p,name):in_pipe,out_pipe=pin_pipe.close()for i in range(10):# print(i)str ='%s生產的包子%s'%(name,i)out_pipe.send(str)# time.sleep(1)else:out_pipe.close()if __name__ == '__main__':in_pipe,out_pipe=Pipe()p = Process(target=producer,args=((in_pipe,out_pipe),'jack'))?c1=Process(target=consumer,args=((in_pipe,out_pipe),'c1'))c2=Process(target=consumer,args=((in_pipe,out_pipe),'c2'))c1.start()c2.start()p.start()?in_pipe.close()out_pipe.close()?c1.join()c2.join()print('主進程')?# 基于管道實現進程間通信（與隊列的方式是類似的，隊列就是管道加鎖實現的）## 加鎖來控制操作管道的行為，來避免進程之間爭搶數據造成的數據不安全現象

　　

這里需要加鎖來解決數據不安全的情況

from multiprocessing import Process,Pipe,Lock?def consumer(produce, consume,name,lock):produce.close()while True:lock.acquire()baozi=consume.recv()lock.release()if baozi:print('%s 收到包子:%s' %(name,baozi))else:consume.close()break?def producer(produce, consume,n):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,30))c1.start()c2.start()p1.start()produce.close()consume.close()

　　

二、進程間的數據共享manager

使用Manager可以方便的進行多進程數據共享，事實上Manager的功能遠不止于此 。Manager支持的類型有list,dict,Namespace,Lock,RLock,Semaphore,BoundedSemaphore,Condition,Event,Queue,Value和Array。

但與管道類似，這里的數據也是不安全的。需要用鎖來解決。

from multiprocessing import Manager,Process?def main(dic):dic['count'] -= 1# print(dic)?if __name__ == '__main__':m = Manager()#為這個manager類注冊存儲容器，也就是通過這個manager類實現的共享的變量dic=m.dict({'count':100})p_lst = []for i in range(50):p = Process(target=main, args=(dic,))p_lst.append(p)p.start()for p in p_lst:p.join()print("主進程",dic['count'])

　　

分析：多運行幾次可以看到，每次輸出的結果都基本是不同的，因此這里還是需要用鎖來解決。

from multiprocessing import Manager,Process,Lock??def main(dic,lock):?# with lock:可以這樣寫，也可以寫成下面的樣子lock.acquire()dic['count'] -= 1lock.release()?if __name__ == '__main__':m = Manager()l = Lock()dic=m.dict({'count':100})p_lst = []for i in range(50):p = Process(target=main,args=(dic,l))p.start()p_lst.append(p)for i in p_lst: i.join()print('主進程',dic)

　　

?

?

參考資料

[1]https://segmentfault.com/a/1190000008122273

[2]http://www.th7.cn/system/lin/201605/165994.shtml

[3]https://blog.csdn.net/weixin_39859512/article/details/80898340

?

轉載于:https://www.cnblogs.com/Nicholas0707/p/10787945.html

總結

以上是生活随笔為你收集整理的Python之路(第三十九篇)管道、进程间数据共享Manager的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。