Python | Use of process pool

プロセスプールの使い方

作成するサブプロセスの数が多くない場合は、multiprocessing の Process を直接使用して複数のプロセスを動的に生成できます。しかし、数百から数千ものプロセスが必要になる場合、手動でプロセスを作成する作業量は膨大になります。そのような場合は、multiprocessing モジュールが提供する Pool メソッドを使用できます。

Pool

プロセスを過剰に起動しても効率は向上せず、むしろ低下します。500 のタスクがあり、同時に 500 のプロセスを起動した場合を考えてみましょう。これらの 500 のプロセスを同時に実行できないだけでなく (CPU にそれほど多くのコアがない)、オペレーティングシステムが 500 のプロセスをスケジューリングして 4 個または 8 個の CPU で平均的に実行させるだけでも、多大なリソースを消費します。

大量の子プロセスを起動する場合は、プロセスプールを使用して子プロセスをバッチで作成できます:

def task(n):
print('{}----->start'. format(n))
time. sleep(1)
print('{}------>end'. format(n))


if __name__ == '__main__':
p = Pool(8) # プロセスプールを作成し、スレッドプールの数を指定。デフォルトは CPU コア数
for i in range(1, 11):
# p.apply(task, args=(i,)) # 同期実行。1 つずつ順次実行され、並行処理の効果なし
p.apply_async(task, args=(i,)) # 非同期実行。並行処理を実現
p. close()
p. join()
プロセスプールからタスクの実行結果を取得します:

def task(n):
print('{}----->start'. format(n))
time. sleep(1)
print('{}------>end'. format(n))
return n ** 2


if __name__ == '__main__':
p = Pool(4)
for i in range(1, 11):
res = p.apply_async(task, args=(i,)) # res はタスクの実行結果
print(res.get()) # 直接結果を取得するデメリットは、マルチタスキングが再び同期になること
p. close()
# p.join() は不要。res.get() 自体がブロッキングメソッドのため
スレッドの実行結果を非同期に取得します:

import time
from multiprocessing.pool import Pool


def task(n):
print('{}----->start'. format(n))
time. sleep(1)
print('{}------>end'. format(n))
return n ** 2


if __name__ == '__main__':
p = Pool(4)
res_list = []
for i in range(1, 11):
res = p.apply_async(task, args=(i,))
res_list.append(res) # リストを使用してプロセスの実行結果を保存
for re in res_list:
print(re. get())
p. close()
Pool の初期化時に、プロセスの最大数を指定できます。新しいリクエストが Pool に送信されたとき、プールに空きがあれば、新しいプロセスが作成されてリクエストを実行します。ただし、プール内のプロセス数が指定された最大値に達している場合、プール内のプロセスの 1 つが終了するまでリクエストは待機し、その後、既存のプロセスが新しいタスクを実行します。次の例を参照してください:

from multiprocessing import Pool
import os, time, random


def worker(msg):
t_start = time. time()
print("%s starts to execute, the process number is %d" % (msg, os.getpid()))
# random.random() は 0 から 1 の間の浮動小数点数をランダムに生成
time. sleep(random. random() * 2)
t_stop = time. time()
print(msg, "Completed, time-consuming %0.2f" % (t_stop - t_start))


if __name__ == '__main__':
po = Pool(3) # プロセスプールを定義。最大プロセス数は 3
for i in range(0, 10):
# Pool().apply_async(呼び出し対象, (対象に渡す引数のタプル,))
# 各ループでアイドル状態の子プロセスを使用して対象を呼び出す
po.apply_async(worker, (i,))

print("----start----")
po.close() # プロセスプールを閉じる。close 以降、po は新しいリクエストを受け付けない
po.join() # po 内のすべての子プロセスの完了を待つ。必ず close 文の後に配置
print("-----end-----")
実行結果:

----start----
0 starts execution, the process number is 21466
1 starts execution, the process number is 21468
2 starts to execute, the process number is 21467
0 Execution completed, time-consuming 1.01
3 start execution, the process number is 21466
2 Execution is complete, time-consuming 1.24
4 starts to execute, the process number is 21467
3 Execution is complete, time-consuming 0.56
5 starts to execute, the process number is 21466
1 Execution is complete, time-consuming 1.68
6 starts to execute, the process number is 21468
4 Execution is complete, time-consuming 0.67
7 starts to execute, the process number is 21467
5 Execution is complete, time-consuming 0.83
8 starts to execute, the process number is 21466
6 Execution is complete, time-consuming 0.75
9 starts to execute, the process number is 21468
7 Execution is complete, time-consuming 1.03
8 Execution is complete, time-consuming 1.05
9 Execution is complete, time-consuming 1.69
-----end-----
multiprocessing.Pool の主な関数の分析:

apply_async(func[, args[, kwds]]) : ノンブロッキング方式で func を呼び出します (並列実行。ブロッキングモードでは、前のプロセスが終了するまで次のプロセスの実行を待ちます)。args は func に渡すパラメーターリスト、kwds は func に渡すキーワード引数のリストです;
close(): 新しいタスクを受け付けなくなるように Pool を閉じます;
terminate(): タスクの完了に関係なく即座に終了します;
join(): メインプロセスをブロックし、子プロセスの終了を待ちます。必ず close または terminate の後に使用します;
プロセスプール内の Queue
Pool を使用してプロセスを作成する場合、multiprocessing.Queue() ではなく multiprocessing.Manager() の Queue() を使用する必要があります。そうしないと、次のようなエラーメッセージが発生します:

RuntimeError: Queue objects should only be shared between processes through inheritance.
次の例は、プロセスプール内のプロセス間通信の方法を示しています:

# import の Queue を Manager の Queue に変更
from multiprocessing import Manager, Pool
import os, time, random


def reader(q):
print("reader starts (%s), parent process is (%s)" % (os.getpid(), os.getppid()))
for i in range(q.qsize()):
print("reader got the message from Queue: %s" % q.get(True))


def writer(q):
print("writer starts (%s), parent process is (%s)" % (os.getpid(), os.getppid()))
for i in "helloworld":
q. put(i)


if __name__ == "__main__":
print("(%s) start" % os. getpid())
q = Manager().Queue() # Manager の Queue を使用
po = Pool()
po.apply_async(writer, (q,))

time.sleep(1) # 上記のタスクで先に Queue にデータを保存してから、次のタスクでデータ取得を開始

po.apply_async(reader, (q,))
po. close()
po. join()
print("(%s) End" % os. getpid())
実行結果:

(4171) start
The writer starts (4173), and the parent process is (4171)
The reader starts (4174), and the parent process is (4171)
The reader gets the message from the Queue: h
The reader gets the message from the Queue: e
The reader gets the message from the Queue: l
The reader gets the message from the Queue: l
The reader gets the message from the Queue: o
The reader gets the message from the Queue: w
The reader gets the message from the Queue: o
The reader gets the message from the Queue: r
The reader gets the message from the Queue: l
The reader gets the message from the Queue: d
(4171) End
join メソッドの使い方
# スレッドとプロセスにはいずれも join メソッドがある
import threading
import time

x = 10

def test(a, b):
time. sleep(1)
global x
x = a + b

# test(1, 1)
# print(x) # 2

t = threading.Thread(target=test, args=(1, 1))
t. start()
t.join() # メインスレッドを待機させる

print(x) # 10