Pythonでのマルチプロセス間のデータ共有

Pythonでのマルチプロセス間のデータ共有
2016-07-24に発表
|Pythonに分類
| 0
|読解回数387
マルチプロセスでは、各プロセスは独立しており、それぞれがデータを持っており、共有できません.この記事では、プロセスデータ共有のための3つの方法について説明します.

queues

Array

Manager.dict

pipe

Queue

1
2
3
4
5
6
7
8
9
10
11
12
13

from multiprocessing import queues
import multiprocessing

def func(i, q):
    q.put(i)
    print("--->", i, q.qsize())


q = queues.Queue(9, ctx=multiprocessing)
for i in range(5):
    p = multiprocessing.Process(target=func, args=(i, q,))
    p.start()
p.join()

Queueはマルチプロセスの安全なキューであり,Queueを用いてマルチプロセス間のデータ転送を実現できる.putメソッドはキューにデータを挿入するために使用され、putメソッドにはblockedとtimeoutの2つのオプションパラメータがあります.blockedがTrue(デフォルト)であり、timeoutが正の値である場合、このメソッドはキューに空き領域があるまでtimeoutが指定した時間をブロックします.タイムアウトした場合、Queue.Full異常が放出されます.blockedがFalseの場合、Queueがいっぱいになると、Queue.Full異常が直ちに放出されます.
getメソッドは、キューから要素を読み出して削除できます.同様にgetメソッドには、blockedとtimeoutの2つのオプションパラメータがあります.blockedがTrue(デフォルト)、timeoutが正の場合、待機時間内に要素が取り込まれない場合、Queue.Empty例外が放出されます.blockedがFalseの場合、2つのケースがあり、Queueの値が1つ使用可能な場合は、すぐに値を返します.そうでない場合、キューが空の場合、Queue.Empty例外がすぐに放出されます

1
2
3
4
5
6
7
8
9
10
11
12

import multiprocessing

def func(i, q):
    q.put(i)
    print("--->", i, q.qsize())


q = multiprocessing.Queue()
for i in range(5):
    p = multiprocessing.Process(target=func, args=(i, q,))
    p.start()
p.join()

Array

1
2
3
4
5
6
7
8
9
10
11
12
13
14

from multiprocessing import Process
from multiprocessing import Array

def func(i, ar):
    ar[i] = i
    for item in ar:
        print(item)
    print("------")

ar = Array('i', 5)
for i in range(5):
    p = Process(target=func, args=(i, ar,))
    p.start()
p.join()

Arrayの限界は,配列の特性に制約されていること,すなわち,データ型を指定し,長さを固定する必要があることである.

1
2
3
4
5
6
7

#        
'c': ctypes.c_char,  'u': ctypes.c_wchar,
'b': ctypes.c_byte,  'B': ctypes.c_ubyte,
'h': ctypes.c_short, 'H': ctypes.c_ushort,
'i': ctypes.c_int,   'I': ctypes.c_uint,
'l': ctypes.c_long,  'L': ctypes.c_ulong,
'f': ctypes.c_float, 'd': ctypes.c_double

Manager.dict

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

from multiprocessing import Process, Manager

#           
#    ，                     
def func(i, d):
    d[i] = i + 100
    print(d.values())

#            
m = Manager()
d = m.dict()

for i in range(5):
    #                
    p = Process(target=func, args=(i, d))
    p.start()
p.join()

------------
[100]
[100, 101]
[100, 101, 102, 103]
[100, 101, 102, 103]
[100, 101, 102, 103, 104]

Manager.dictはマルチプロセスデータ共有でよく使われる方法です
pipe
Pipeメソッドは、1つのパイプの2つの端を表す(conn 1,conn 2)を返します.Pipeメソッドにはduplexパラメータがあり、duplexパラメータがTrue(デフォルト)の場合、このパイプはフルデュプレクスモード、すなわちconn 1とconn 2が送受信可能である.duplexはFalse,conn 1はメッセージの受信のみ,conn 2はメッセージの送信のみを担当する.
sendメソッドとrecvメソッドは、それぞれメッセージを送信および受信する方法である.例えば、conn 1は、フルデュプレクスモードで呼び出すことができる.send送信メッセージconn 1.recvメッセージを受信します.メッセージが受信できない場合、recvメソッドはブロックされます.パイプが閉じられている場合、recvメソッドはEOFErrorを放出します.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

import multiprocessing

def func1(arg, pp):
    pp.send(arg)

def func2(pp):
    recv = pp.recv()
    print(recv)

pp = multiprocessing.Pipe()
p1 = multiprocessing.Process(target=func1, args=("PolarSnow", pp[0],))
p2 = multiprocessing.Process(target=func2, args=(pp[1],))
p1.start()
p2.start()
p1.join()
p2.join()

------------
PolarSnow

#マルチプロセス#multiprocessing
Pythonのリストと配列の違い
プロセスロック