溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
小編給大家分享一下python如何實現多進程通信，希望大家閱讀完這篇文章之后都有所收獲，下面讓我們一起去探討吧！
1. 管道
先來看一下最簡單、古老的一種IPC：管道。通常指的是無名管道，本質上可以看做一種文件，只存在于內存當中，不會存盤。不同進程通過系統提供的接口來向管道中讀取或者寫入數據。
也就是說我們通過這樣一個中間介質為進程提供交流的方式。無名管道的局限在于一般只用于有直接關聯關系的父子進程。下面通過一個簡單的例子來看一下其用法。
from multiprocessing import Process, Pipe
def pstart(pname, conn):
conn.send("Data@subprocess")
print(conn.recv()) # Data@parentprocess
if __name__ == '__main__':
conn1, conn2 = Pipe(True)
sub_proc = Process(target=pstart, args=('subprocess', conn2,))
sub_proc.start()
print (conn1.recv()) # Data@subprocess
conn1.send("Data@parentprocess")
sub_proc.join()管道通信三步曲：
創建Pipe，得到兩個connection對象conn1和conn2；
父進程持有conn1，將conn2傳遞給子進程；
父子進程通過對持有的connection對象進行send和recv操作以進行數據傳遞和接受；
上面我們創建的是全雙工管道，也可以創建半雙工管道，具體使用可以參照官網描述：
Returns a pair
(conn1, conn2)ofConnectionobjects representing the ends of a pipe.If duplex is
True(the default) then the pipe is bidirectional. If duplex isFalsethen the pipe is unidirectional:conn1can only be used for receiving messages andconn2can only be used for sending messages.
2. 具名管道（FIFO）
上面介紹的管道主要用于有直接關系的進程，局限性比較大。下面來看一下可以在任意進程間進行通信的具名管道。
由于window平臺上os模塊沒有mkfifo屬性，因此這個例子只能在linux上運行（測試環境 CentOS 7， Python 2.7.5）：
#!/usr/bin/python
import os, time
from multiprocessing import Process
input_pipe = "./pipe.in"
output_pipe = "./pipe.out"
def consumer():
if os.path.exists(input_pipe):
os.remove(input_pipe)
if os.path.exists(output_pipe):
os.remove(output_pipe)
os.mkfifo(output_pipe)
os.mkfifo(input_pipe)
in1 = os.open(input_pipe, os.O_RDONLY) # read from pipe.in
out1 = os.open(output_pipe, os.O_SYNC | os.O_CREAT | os.O_RDWR)
while True:
read_data = os.read(in1, 1024)
print("received data from pipe.in: %s @consumer" % read_data)
if len(read_data) == 0:
time.sleep(1)
continue
if "exit" in read_data:
break
os.write(out1, read_data)
os.close(in1)
os.close(out1)
def producer():
in2 = None
out2 = os.open(input_pipe, os.O_SYNC | os.O_CREAT | os.O_RDWR)
for i in range(1, 4):
msg = "msg " + str(i)
len_send = os.write(out2, msg)
print("------product msg: %s by producer------" % msg)
if in2 is None:
in2 = os.open(output_pipe, os.O_RDONLY) # read from pipe.out
data = os.read(in2, 1024)
if len(data) == 0:
break
print("received data from pipe.out: %s @producer" % data)
time.sleep(1)
os.write(out2, 'exit')
os.close(in2)
os.close(out2)
if __name__ == '__main__':
pconsumer = Process(target=consumer, args=())
pproducer = Process(target=producer, args=())
pconsumer.start()
time.sleep(0.5)
pproducer.start()
pconsumer.join()
pproducer.join()運行流程如下：
每一輪的過程如下：
producer進程往pipe.in文件中寫入消息數據；
consumer進程從pipe.in文件中讀入消息數據；
consumer進程往pipe.out文件中寫入回執消息數據；
producer進程從pipe.out文件中讀出回執消息數據；
結果如下：
[shijun@localhost python]$ python main.py ------product msg: msg 1 by producer------ received data from pipe.in: msg 1 @consumer received data from pipe.out: msg 1 @producer ------product msg: msg 2 by producer------ received data from pipe.in: msg 2 @consumer received data from pipe.out: msg 2 @producer ------product msg: msg 3 by producer------ received data from pipe.in: msg 3 @consumer received data from pipe.out: msg 3 @producer received data from pipe.in: exit @consumer
兩個進程沒有直接的關系，每個進程有一個讀文件和寫文件，如果兩個進程的讀寫文件是關聯的，就可以進行通信。
3. 消息隊列（Queue）
進程之間通過向隊列中添加數據或者從隊列中獲取數據來進行消息數據的傳遞。下面是一個簡單的例子。
from multiprocessing import Process, Queue
import time
def producer(que):
for product in ('Orange', 'Apple', ''):
print('put product: %s to queue' % product)
que.put(product)
time.sleep(0.5)
res = que.get()
print('consumer result: %s' % res)
def consumer(que):
while True:
product = que.get()
print('get product:%s from queue' % product)
que.put('suc!')
time.sleep(0.5)
if not product:
break
if __name__ == '__main__':
que = Queue(1)
p = Process(target=producer, args=(que,))
c = Process(target=consumer, args=(que,))
p.start()
c.start()
p.join()
c.join()這個例子比較簡單，queue的具體用法可以參考一下官網。
結果：
put product: Orange to queue consumer result: suc! put product: Apple to queue consumer result: suc! put product: to queue consumer result: suc! get product:Orange from queue get product:Apple from queue get product: from queue
這里有幾點需要注意下：
可以指定隊列的容量，如果超出容量會有異常：raise Full；
默認put和get均會阻塞當前進程；
如果put沒有設置成阻塞，那么可能自己從隊列中取出自己放入的數據；
4. 共享內存
共享內存是一種常用的，高效的進程之間的通信方式，為了保證共享內存的有序訪問，需要對進程采取額外的同步措施。
下面的這個例子僅僅簡單的演示了Python中如何在不同進程間使用共享內存進行通信的。
from multiprocessing import Process
import mmap
import contextlib
import time
def writer():
with contextlib.closing(mmap.mmap(-1, 1024, tagname='cnblogs', access=mmap.ACCESS_WRITE)) as mem:
for share_data in ("Hello", "Alpha_Panda"):
mem.seek(0)
print('Write data:== %s == to share memory!' % share_data)
mem.write(str.encode(share_data))
mem.flush()
time.sleep(0.5)
def reader():
while True:
invalid_byte, empty_byte = str.encode('\x00'), str.encode('')
with contextlib.closing(mmap.mmap(-1, 1024, tagname='cnblogs', access=mmap.ACCESS_READ)) as mem:
share_data = mem.read(1024).replace(invalid_byte, empty_byte)
if not share_data:
""" 當共享內存沒有有效數據時結束reader """
break
print("Get data:== %s == from share memory!" % share_data.decode())
time.sleep(0.5)
if __name__ == '__main__':
p_reader = Process(target=reader, args=())
p_writer = Process(target=writer, args=())
p_writer.start()
p_reader.start()
p_writer.join()
p_reader.join()執行結果：
Write data:== Hello == to share memory! Write data:== Alpha_Panda == to share memory! Get data:== Hello == from share memory! Get data:== Alpha_Panda == from share memory!
下面簡單的來說明一下共享內存的原理；
進程虛擬地址到物理地址的一個映射關如下：
上面這個圖已經很明白的展示了共享內存的原理。
左邊是正常情況下，不同進程的線性地址空間被映射到不同的物理內存頁，這樣不管其他進程怎么修改物理內存，都不會影響到其他進程；
右邊表示的是進程共享內存的情況下，不同進程的部分線性地址會被映射到同一物理頁，一個進程對這個物理頁的修改，會對另一個進程立即可見；
當然潛在的問題就是要采取進程同步措施，也就是對共享內存的訪問必須是互斥的。這個可以借助信號量來實現。
5. socket通信
最后再來介紹一種可以跨主機的進程間通信：socket。
懂網絡編程的人，對這個應該都比較熟悉。socket不僅可以跨主機進行通信，甚至有時候可以使用socket在同一主機的不同進程間進行通信。
這部分代碼比較簡單常見，這里僅僅使用流程圖來表示一下socket通信的流程及相關接口。
上圖表示客戶端上某進程使用socket和服務器上監聽程序進行socket通信的一個流程。
看完了這篇文章，相信你對“python如何實現多進程通信”有了一定的了解，如果想了解更多相關知識，歡迎關注億速云行業資訊頻道，感謝各位的閱讀！
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
