Pythonスレッド間通信のマスター：同期とデータ共有

スレッド間通信とは、Pythonのマルチスレッドプログラム内でスレッド間の通信と同期を可能にするプロセスを指します。

一般的に、Pythonのスレッドはプロセス内で同じメモリ空間を共有するため、共有変数、オブジェクト、threading モジュールが提供する特殊な同期プリミティブを通じてデータを交換し、活動を調整できます。

スレッド間通信を容易にするため、threading モジュールは Lock、Event、Condition、Semaphore などの同期プリミティブを提供します。このチュートリアルでは、EventオブジェクトとConditionオブジェクトを使用したスレッド間通信の方法を学びます。

Eventオブジェクト

Eventオブジェクトは内部フラグの状態を管理し、スレッドが待機（wait）または設定（set）できるようにします。このフラグの状態を制御するメソッドを提供し、共有条件に基づいてスレッドが活動を同期できるようにします。

フラグは初期状態で false であり、set() メソッドで true に、clear() メソッドで false にリセットされます。wait() メソッドはフラグが true になるまでブロックします。

Eventオブジェクトの主要メソッド：

• is_set() — 内部フラグが true の場合のみ True を返す
• set() — 内部フラグを true に設定。true になるのを待っているすべてのスレッドを起床。フラグがすでに true の場合、wait() を呼んでもブロックしない
• clear() — 内部フラグを false にリセット。その後、wait() を呼ぶスレッドは set() が呼ばれるまでブロック
• wait(timeout=None) — 内部フラグが true になるまでブロック。入力時にフラグが true なら即座に返る。そうでなければ、別スレッドが set() を呼ぶか、オプションのタイムアウトが発生するまでブロック。timeout引数がある場合は秒数を浮動小数点数で指定

例：交通信号シミュレーション

以下のコードは、交通信号の状態（青/赤）によって交通流を制御するシミュレーションです。

2つのスレッドで、それぞれ異なる関数を対象とします。signal_state() 関数は定期的にイベントを設定/リセットし、青信号から赤信号への切り替えを示します。traffic_flow() 関数はイベントが設定されるのを待ち、設定されている間ループを実行します。

 from threading import Event, Thread import time terminate = False def signal_state(): global terminate while not terminate: time.sleep(0.5) print("Traffic Police Giving GREEN Signal") event.set() time.sleep(1) print("Traffic Police Giving RED Signal") event.clear() def traffic_flow(): global terminate num = 0 while num < 10 and not terminate: print("Waiting for GREEN Signal") event.wait() print("GREEN Signal ... Traffic can move") while event.is_set() and not terminate: num += 1 print("Vehicle No:", num," Crossing the Signal") time.sleep(1) print("RED Signal ... Traffic has to wait") event = Event() t1 = Thread(target=signal_state) t2 = Thread(target=traffic_flow) t1.start() t2.start() # しばらくしてスレッドを終了 time.sleep(5) terminate = True # すべてのスレッドの完了を待機 t1.join() t2.join() print("Exiting Main Thread") 

出力例

 Waiting for GREEN Signal Traffic Police Giving GREEN Signal GREEN Signal ... Traffic can move Vehicle No: 1 Crossing the Signal Traffic Police Giving RED Signal RED Signal ... Traffic has to wait Waiting for GREEN Signal Traffic Police Giving GREEN Signal GREEN Signal ... Traffic can move Vehicle No: 2 Crossing the Signal Vehicle No: 3 Crossing the Signal Traffic Police Giving RED Signal Traffic Police Giving GREEN Signal Vehicle No: 4 Crossing the Signal Traffic Police Giving RED Signal RED Signal ... Traffic has to wait Traffic Police Giving GREEN Signal Traffic Police Giving RED Signal Exiting Main Thread 

Conditionオブジェクト

Pythonの threading モジュールの Condition オブジェクトは、より高度な同期メカニズムを提供します。スレッドが別スレッドからの通知を待ってから進行できるようにします。Conditionオブジェクトは常にロックと関連付けられ、スレッド間のシグナリングメカニズムを提供します。

構文

 threading.Condition(lock=None) 

Conditionオブジェクトの主要メソッド：

• acquire(*args) — 基盤となるロックを獲得。対応するロックメソッドを呼び出し、その戻り値を返す
• release() — 基盤となるロックを解放。対応するロックメソッドを呼び出し、戻り値なし
• wait(timeout=None) — 基盤ロックを解放し、別スレッドからの notify() または notify_all() 呼び出し、またはオプションのタイムアウト発生までブロック。起床またはタイムアウト後、ロックを再獲得して返る
• wait_for(predicate, timeout=None) — 述語が満たされるまで、またはタイムアウトまで繰り返し wait() を呼び出すユーティリティメソッド。戻り値は述語の最後の戻り値。タイムアウト時は False
• notify(n=1) — 条件変数を待っているスレッドのうち最大 n 個を起床。待機スレッドがなければ no-op
• notify_all() — この条件で待機しているすべてのスレッドを起床。notify() と同様だが、1つではなくすべてを起床。呼び出しスレッドがロックを獲得していない状態で呼ぶと RuntimeError

例1：Conditionオブジェクトによるシンプルな通信

スレッドAとスレッドBがConditionオブジェクトで通信。スレッドAはスレッドBからの通知を待ち、スレッドBは2秒スリープ後にスレッドAに通知して終了。

 from threading import Condition, Thread import time c = Condition() def thread_a(): print("Thread A started") with c: print("Thread A waiting for permission...") c.wait() print("Thread A got permission!") print("Thread A finished") def thread_b(): print("Thread B started") with c: time.sleep(2) print("Notifying Thread A...") c.notify() print("Thread B finished") Thread(target=thread_a).start() Thread(target=thread_b).start() 

出力：

 Thread A started Thread A waiting for permission... Thread B started Notifying Thread A... Thread B finished Thread A got permission! Thread A finished 

例2：生産者・消費者パターン

スレッド t2 が taskB()、スレッド t1 が taskA() を実行。t1 が条件を獲得し通知。その間 t2 は待機状態。条件解放後、待機スレッドが通知関数で生成された乱数を取得・消費。

 from threading import Condition, Thread import time import random numbers = [] def taskA(c): for _ in range(5): with c: num = random.randint(1, 10) print("Generated random number:", num) numbers.append(num) print("Notification issued") c.notify() time.sleep(0.3) def taskB(c): for i in range(5): with c: print("waiting for update") while not numbers: c.wait() print("Obtained random number", numbers.pop()) time.sleep(0.3) c = Condition() t1 = Thread(target=taskB, args=(c,)) t2 = Thread(target=taskA, args=(c,)) t1.start() t2.start() t1.join() t2.join() print("Done") 

実行結果：

 waiting for update Generated random number: 2 Notification issued Obtained random number 2 Generated random number: 5 Notification issued waiting for update Obtained random number 5 Generated random number: 1 Notification issued waiting for update Obtained random number 1 Generated random number: 9 Notification issued waiting for update Obtained random number 9 Generated random number: 2 Notification issued waiting for update Obtained random number 2 Done