Pythonのマルチスレッド処理【プログラミング初心者向け教材】

たくさんの処理を効率よく行うための手法として並列処理があります。

処理の単位をスレッドと呼び、並列で処理することをマルチスレッドと呼びます。

Pythonではマルチスレッド処理を行うための機能が標準で提供されています。

本記事では、Pythonのマルチスレッド処理について解説します。

プログラミング初心者の方の学習や、忘れてしまった方の復習として、参考にしていただければ幸いです。

記載しているプログラムは、Python3.6.8を使って動作確認をしています。

Pythonのマルチスレッド処理
時間のかかる処理は非同期処理で効率よく実行しよう

Pythonのマルチスレッド処理

Pythonのマルチスレッド処理について、以下の内容で解説します。

マルチスレッドを扱うモジュール
スレッドの作成と起動
スレッドプールの作成と起動
スレッドからの戻り値の取得

マルチスレッドを扱うモジュール

Pythonでスレッドを扱うには、threadingモジュールをimportします。

import threading

また、スレッドを複数まとめて効率よく扱うスレッドプールを利用するには、concurrent.futuresモジュールをimportします。

from concurrent.futures import ThreadPoolExecutor

スレッドの作成と起動

スレッドは処理をまとめたグループのようなものですが、Pythonでは関数をスレッドとして指定します。
スレッドを作成するには、スレッド化する関数を指定して以下のように作成します。

def thread1():
    print("thread")

t1 = threading.Thread(target=thread1)

targetにスレッドとする関数を指定します。
作成したスレッドはstartで起動します。

t1.start()

スレッドの処理が終了するのを待たないと、メイン処理が先に終了してしまい、スレッド処理が途中までになってしまいます。
そのようなことにならないように、joinでスレッドが終了するまで待ちます。

t1.join()

このようなスレッドを複数作成して、並列に処理させることで待ち時間を少なくして効率よく処理させることができます。

def thread1():
    for i in range(1, 5):
        time.sleep(random.random())
        print(f"thread1: {i}")

def thread2():
    for i in range(1, 5):
        time.sleep(random.random())
        print(f"thread2: {i}")

print("start.")

t1 = threading.Thread(target=thread1)
t2 = threading.Thread(target=thread2)
t1.start()
t2.start()
t1.join()
t2.join()

print("end.")

上記のプログラムでは、2つのスレッドがランダムな時間をスリープしながら5回処理を繰り返します。
スリープ中の待ち時間の間に、別のスレッドの処理を行うことで、待ち時間が最小となるように処理することが出来ます。
実行すると以下のように、スレッド1とスレッド2が並列で処理されていきます。

start.
thread2: 1
thread2: 2
thread1: 1
thread1: 2
thread2: 3
thread1: 3
thread2: 4
thread1: 4
end.

スレッドプールの作成と起動

スレッドはいくつでも作って並列で処理させることが可能ですが、数が多すぎると逆に効率が悪くなることがあります。
スレッドが処理する数を制限する際に用いるのが、スレッドプールです。

スレッドプールは、スレッドの登録はいくつでも出来ますが、処理する最大数は決められた数までになります。
最大数は基本的に処理するマシンのCPUの数にすることが多いです。

スレッドプールの作成は以下のように行います。

pool = ThreadPoolExecutor(max_workers=3)

上記ではスレッド実行数を3に設定しています。
起動は submit にスレッド化する関数を指定します。

pool.submit(thread1)

スレッドが最後まで実行されるのを待つために、shutdown を呼出します。

pool.shutdown()

スレッドを最大処理数以上作ってみると、最大数までしか同時に実行されないことがわかります。

def thread3(n):
    print(f"thread {n}: start.")
    time.sleep(random.random())
    print(f"thread {n}: end.")

print("start.")
pool = ThreadPoolExecutor(max_workers=3, thread_name_prefix="thread")
for i in range(1,6):
    pool.submit(thread3, i)
pool.shutdown()
print("end.")

上記のプログラムでは、スレッドを5つループで作成して起動しています。
それぞれのスレッドでは作成された番号をもらって表示しています。

実行すると以下のようになります。

start.
thread 1: start.
thread 2: start.
thread 3: start.
thread 1: end.
thread 4: start.
thread 4: end.
thread 5: start.
thread 5: end.
thread 2: end.
thread 3: end.
end.

上記の結果では、スレッド1、2、3が最初に起動し、スレッド4がスレッド1の終了後まで待っていることがわかります。
スレッド4の終了後に、スレッド5が開始されて、最大実行数が3になっています。

スレッドからの戻り値の取得

スレッド内で処理した結果を受け取る場合は、戻り値をFutureとして受け取ります。

future = pool.submit(thread3)

受け取ったFutureから本来の戻り値を取得するには、resultを呼出します。

r = future.result()

本来の戻り値を取得するには、スレッドの処理が最後まで終了しないと受け取れないので、スレッドの処理時間が長い場合は、待ちが発生することになります。

複数の処理結果を一括で受け取って、最後に本来の戻り値を取得するにはFutureをリストなどで一旦保持しておきます。

def thread4(n):
    print(f"thread {n}: start.")
    time.sleep(random.random())
    print(f"thread {n}: end.")
    return 10+n

futures = []
print("start.")

with ThreadPoolExecutor(max_workers=3, thread_name_prefix="thread") as pool:
    for i in range(1,6):
        future = pool.submit(thread4, i)
        futures.append(future)
for future in futures:
    print(future.result())

print("end.")

上記のプログラムでは、5つのスレッドの結果をリストに保持しておき、最後にまとめてresultを呼出しています。
先にスレッドの実行をしているので、戻り値を受け取るのに待ち時間を最小限にすることができます。
実行結果は以下のようになります。

start.
thread 1: start.
thread 2: start.
thread 3: start.
thread 1: end.
thread 4: start.
thread 3: end.
thread 5: start.
thread 2: end.
thread 4: end.
thread 5: end.
11
12
13
14
15
end.