Pythonで並行処理が書けるようになるには

概要

Pythonをこれまで扱ってきたなかでよく並行処理を耳にすることが多かったのですが、
概念は分かりつつも実際には何をどのようにすることで並行処理を実装できるかが深いところまで理解をしてこなかったので、今回本腰を入れて学習してみようという記録です

本題

といっても、まずは並行処理とはなんぞやという感じの方もいるかも知れません

ざっくりいうと「複数のタスクが「同時に進行しているように見える」状態」だそうです
実際に同時に実行している状態は並列処理というのですがPythonではGILという1度に1つのスレッドしかPythonを実行できないという制約があるので並列処理は少し難し目となっています(それでもmultiprocessingというものが存在するのですが)

さてそんな平行処理ですが、標準モジュールであるthreadingモジュールを使うことで、このスレッドを増やして複数の処理を平行に動かすことができます

サンプルコード

sanmple.py

import threading
import time

def task(name, delay):
    """
    スレッドで実行するタスク
    """
    print(f"スレッド {name}: 開始")
    for i in range(3):
        print(f"スレッド {name}: 処理中 ({i+1}/3)")
        time.sleep(delay)
    print(f"スレッド {name}: 終了")

if __name__ == "__main__":
    print("メインスレッド: 開始")

    # スレッド1の作成と開始
    thread1 = threading.Thread(target=task, args=("A", 1), name="Thread-A")
    thread1.start()

    # スレッド2の作成と開始
    thread2 = threading.Thread(target=task, args=("B", 0.5), name="Thread-B")
    thread2.start()

    # スレッドが終了するのを待つ
    thread1.join()
    thread2.join()

    print("メインスレッド: 全てのスレッドが終了しました")
    print("メインスレッド: 終了")

threadingモジュールを使う上で重要なクラスや関数は一旦以下になりそうです

threading.Thread クラス:
新しいスレッドを作成するための基盤となるクラス
スレッドで実行したい処理は、このクラスのインスタンスを作成する際にtarget引数に関数を渡すか、Threadクラスを継承してrun()メソッドをオーバーライドすることで定義が可能

コンストラクタの主な引数:

target: スレッドで実行する関数を指定
args: target関数に渡す引数をタプルで指定
kwargs: target関数に渡すキーワード引数を辞書で指定
name: スレッドに名前を付けます。デバッグなどに使用することが可能
daemon: スレッドをデーモンスレッドにするかどうかを指定可能
通常はメインのプログラムが完了するまでスレッドの処理を待機しますが、デーモンスレッドだとメインプログラムが終了すると自動的に終了しするようになります

主なメソッド:

start(): スレッドの実行を開始
これにより、targetで指定した関数（またはオーバーライドしたrun()メソッド）が新しいスレッドで呼び出される
join(timeout=None): スレッドの終了を待機
このメソッドを呼び出したスレッド（通常はメインスレッド）は、join()を呼び出されたスレッドが終了するまでブロックされる
timeoutを指定すると、指定した時間だけ待機し、スレッドが終了しなくても処理を続行する
is_alive(): スレッドが現在も実行中（生きている）かどうかを返却する
getName() / name: スレッドの名前を取得・設定します
threading.Lock (ロック):
複数のスレッドが共有リソース（変数、ファイルなど）に同時にアクセスすると、予期せぬ結果（競合状態、データ破損など）が生じる可能性があるのでそれを防ぐために、ロックを使用する
ロックは、一度に1つのスレッドしかアクセスできない排他制御を行うための仕組み
(SQLのロックなど知っていれば話が早いですね)

acquire(): ロックを取得、ロックが既に取得されている場合は解放されるまでブロックされる
release(): ロックを解放します
threading.RLock (再入可能ロック):
Lockと同様だが、同じスレッドが複数回ロックを取得できる点が異なる
ただし、取得した回数だけ解放する必要が発生する

threading.Condition (条件変数):
特定条件が満たされるまでスレッドを待機させたり、条件が満たされたときにスレッドを再開させたりするために使用

threading.Semaphore (セマフォ):
一度にリソースにアクセスできるスレッドの数を制限する場合に使用

threading.Event (イベント):
スレッド間のシグナル伝達に使用します
あるスレッドがイベントを設定し、他のスレッドがそのイベントを待つことで処理の同期を取ることができる

いろいろと書きましたが、基本は以下の形を覚えておくことができればベースはバッチリなような気がします
基本概念は平行に処理したいものを関数に書き出し、それをthreading.Threadクラスで呼び出すということです

import threading
import time

def task(name, delay):
    """
    スレッドで実行したいタスク
    """
    pass

if __name__ == "__main__":
    print("メインスレッド: 開始")

    thread1 = threading.Thread(target=task, args=("A", 1), name="Thread-A")
    thread1.start()
    thread1.join()

おわりに

ということで平行処理の基礎の基礎みたいなところを書いてみました
割と腑に落ちながらまとめられた気がします
今後必要なときにベースさえわかっていれば複雑なものでなければ書けそうな気がしています
requestsモジュールを何かしらのエンドポイントへリクエストをたくさん送って負荷試験するとか簡単にできそうな気がしています

いつかmultiprocessingとかにも手を出してみたいです