進(jìn)程間通信表示進(jìn)程之間的數(shù)據(jù)交換。 為了開發(fā)并行應(yīng)用程序,需要在進(jìn)程間交換數(shù)據(jù)。 下圖顯示了多個子過程之間同步的各種通信機(jī)制 -
在本節(jié)中,我們將了解各種通信機(jī)制。 機(jī)制如下所述 -
隊列
隊列可以用于多進(jìn)程程序。 多處理模塊的Queue
類與Queue.Queue
類相似。 因此,可以使用相同的API。 Multiprocessing.Queue
提供了進(jìn)程間通信的線程和進(jìn)程安全FIFO(先進(jìn)先出)機(jī)制。
例子
下面是一個簡單的例子,從python官方文檔多處理了解Queue類的多處理概念。
from multiprocessing import Process, Queue
import queue
import random
def f(q):
q.put([42, None, 'hello'])
def main():
q = Queue()
p = Process(target = f, args = (q,))
p.start()
print (q.get())
if __name__ == '__main__':
main()
執(zhí)行上面示例代碼,得到以下結(jié)果 -
[42, None, 'hello']
管道
它是一種數(shù)據(jù)結(jié)構(gòu),用于在多進(jìn)程程序中的進(jìn)程之間進(jìn)行通信。Pipe()
函數(shù)返回一對由管道連接的連接對象,默認(rèn)情況下是雙工(雙向)。 它的工作原理如下 -
它返回一對代表管道兩端的連接對象。
每個對象都有兩個方法 - send()
和recv()
,以在進(jìn)程之間進(jìn)行通信。
例子
下面是一個簡單的例子,摘自python官方文檔多處理,以理解Pipe()
函數(shù)的多進(jìn)程概念。
from multiprocessing import Process, Pipe
def f(conn):
conn.send([42, None, 'hello'])
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()
執(zhí)行上面代碼,得到以下結(jié)果 -
[42, None, 'hello']
管理器
Manager是一類多處理模塊,它提供了一種協(xié)調(diào)所有用戶之間共享信息的方式。管理器對象控制服務(wù)器進(jìn)程,該進(jìn)程管理共享對象并允許其他進(jìn)程操縱它們。 換句話說,管理器提供了一種方法來創(chuàng)建可以在不同進(jìn)程之間共享的數(shù)據(jù)。 以下是Manager
對象的不同屬性 -
例子
以下是使用管理器對象在服務(wù)器進(jìn)程中創(chuàng)建列表記錄,然后在該列表中添加新記錄的示例。
import multiprocessing
def print_records(records):
for record in records:
print("Name: {0}\nScore: {1}\n".format(record[0], record[1]))
def insert_record(record, records):
records.append(record)
print("A New record is added\n")
if __name__ == '__main__':
with multiprocessing.Manager() as manager:
records = manager.list([('Computers', 1), ('Histoty', 5), ('Hindi',9)])
new_record = ('English', 3)
p1 = multiprocessing.Process(target = insert_record, args = (new_record, records))
p2 = multiprocessing.Process(target = print_records, args = (records,))
p1.start()
p1.join()
p2.start()
p2.join()
執(zhí)行上面代碼,得到以下結(jié)果 -
A New record is added
Name: Computers
Score: 1
Name: Histoty
Score: 5
Name: Hindi
Score: 9
Name: English
Score: 3
管理器命名空間的概念
Manager
類帶有名稱空間的概念,這是一種在多個進(jìn)程間共享多個屬性的快速方法。 命名空間不具有任何可以調(diào)用的公共方法,但它們具有可寫的屬性。
例子
以下Python腳本示例如何使用命名空間在主進(jìn)程和子進(jìn)程之間共享數(shù)據(jù) -
import multiprocessing
def Mng_NaSp(using_ns):
using_ns.x +=5
using_ns.y *= 10
if __name__ == '__main__':
manager = multiprocessing.Manager()
using_ns = manager.Namespace()
using_ns.x = 1
using_ns.y = 1
print ('before', using_ns)
p = multiprocessing.Process(target = Mng_NaSp, args = (using_ns,))
p.start()
p.join()
print ('after', using_ns)
執(zhí)行上面示例代碼,得到以下結(jié)果 -
before Namespace(x = 1, y = 1)
after Namespace(x = 6, y = 10)
Multiprocessing
模塊提供了Array
和Value
對象,用于將數(shù)據(jù)存儲在共享內(nèi)存映射中。 Array是從共享內(nèi)存分配的Array
和Value
是從共享內(nèi)存分配的ctypes
對象。
Multiprocessing
模塊導(dǎo)入Process
,Value
,Array
。
例子
下面的Python腳本是一個從python文檔中獲取的例子,它利用Ctypes Array
和Value
在進(jìn)程間共享一些數(shù)據(jù)。
def f(n, a):
n.value = 3.1415927
for i in range(len(a)):
a[i] = -a[i]
if __name__ == '__main__':
num = Value('d', 0.0)
arr = Array('i', range(10))
p = Process(target = f, args = (num, arr))
p.start()
p.join()
print (num.value)
print (arr[:])
執(zhí)行上面示例代碼,得到以下結(jié)果 -
3.1415927
[0, -1, -2, -3, -4, -5, -6, -7, -8, -9]
CSP用于說明系統(tǒng)與具有并行模型的其他系統(tǒng)的交互。 CSP是通過消息傳遞編寫并發(fā)或編程的框架,因此它對于描述并發(fā)是有效的。
要實現(xiàn)在CSP中找到的核心原語,Python有一個名為PyCSP的庫。 它使實現(xiàn)非常簡短和易讀,因此可以非常容易地理解它。 以下是PyCSP的基本流程網(wǎng)絡(luò) -
在上面的PyCSP過程網(wǎng)絡(luò)中,有兩個過程 - 進(jìn)程1和進(jìn)程2。這些過程通過傳遞消息通過兩個通道 - 通道1和通道2進(jìn)行通信。
安裝PyCSP
通過以下命令來安裝Python的PyCSP庫 -
pip install PyCSP
例子
下面的Python腳本是一個簡單的例子,它可以并行運行兩個進(jìn)程。 它是在PyCSP庫的幫助下完成的 -
from pycsp.parallel import *
import time
@process
def P1():
time.sleep(1)
print('P1 exiting')
@process
def P2():
time.sleep(1)
print('P2 exiting')
def main():
Parallel(P1(), P2())
print('Terminating')
if __name__ == '__main__':
main()
在上面的腳本中,已經(jīng)創(chuàng)建了兩個函數(shù),即P1
和P2
,然后用@process
進(jìn)行裝飾,將它們轉(zhuǎn)換為進(jìn)程。執(zhí)行上面代碼后,得到以下輸出結(jié)果 -
P2 exiting
P1 exiting
Terminating