目次
概要
Pythonの threading モジュールで作成したスレッドが、現在動作中(生存中)なのか、それとも既に処理を終えているのかを確認する方法を解説します。
is_alive() メソッドを使用することで、スレッドの完了を待ちながら別の処理(進捗表示やログ出力など)を行う制御が可能になります。
仕様(入出力)
- 入力:
Threadオブジェクト - 出力: 真偽値 (
bool)True: スレッドが実行中の場合False: スレッドがまだ開始されていない、または終了した場合
基本の使い方
thread.is_alive() を呼び出すだけで、その時点でのスレッドの状態を取得できます。
import threading
import time
def task():
time.sleep(1)
t = threading.Thread(target=task)
# start() 前は False
print(f"開始前: {t.is_alive()}")
t.start()
# 実行中は True
print(f"実行中: {t.is_alive()}")
t.join()
# 終了後は False
print(f"終了後: {t.is_alive()}")
コード全文
2つのスレッドを起動し、両方が終了するまで「実行中」というログを出し続ける監視処理のサンプルコードです。
join() でただ待つだけでなく、待っている間にメインスレッドで何かを行いたい場合に有効なパターンです。
import threading
import time
def print_numbers(thread_name, loops):
"""数字をカウントアップして出力する関数"""
print(f"[{thread_name}] 処理開始")
for i in range(loops):
print(f" {thread_name}: {i}")
time.sleep(1)
print(f"[{thread_name}] 処理完了")
if __name__ == "__main__":
# 1. スレッドの作成
# thread_name 引数に合わせて kwargs を設定
thread1 = threading.Thread(
target=print_numbers,
kwargs={"thread_name": "Thread-1", "loops": 5}
)
thread2 = threading.Thread(
target=print_numbers,
kwargs={"thread_name": "Thread-2", "loops": 3}
)
# 2. スレッドの開始
thread1.start()
thread2.start()
print("--- 監視ループ開始 ---")
# 3. 状態確認ループ
# thread1 か thread2 のどちらか一方でも動いていればループ継続
while thread1.is_alive() or thread2.is_alive():
print(f"現在、スレッド実行中です... (T1: {thread1.is_alive()}, T2: {thread2.is_alive()})")
# 監視負荷を下げるために少し待機
time.sleep(1)
print("--- 監視ループ終了 ---")
# 念のため join しておく(既に終わっているはずなので一瞬で通過します)
thread1.join()
thread2.join()
print("全ての処理が終了しました。")
カスタムポイント
状態確認メソッド
スレッドの状態を知るためのメソッドです。
| メソッド | 戻り値 | 説明 |
thread.is_alive() | bool | スレッドが開始され、かつ終了していない場合に True を返します。 |
threading.active_count() | int | 現在生存しているスレッドの総数を返します(メインスレッド含む)。 |
threading.enumerate() | list | 現在生存している全ての Thread オブジェクトのリストを返します。 |
注意点
- start() 前の状態
is_alive()はstart()メソッドが呼ばれる前はFalseを返します。「まだ始まっていない」状態と「終わった」状態の区別はつきません。
- join() との併用
thread.join()を呼ぶと、そのスレッドが終わるまでメイン処理がブロック(停止)されます。is_alive()を使った監視ループ(ポーリング)を行う場合は、join()はループを抜けた後、またはループ内でのタイムアウト付きjoin(timeout=0.1)などの形で使用します。
- メインスレッドの終了
- デーモンスレッドではない通常のスレッドが
is_alive() == Trueのままだと、メインプログラム(スクリプト)自体も終了しません。
- デーモンスレッドではない通常のスレッドが
応用
現在実行中のすべてのスレッドの名前をリストアップして表示するデバッグ用のコードです。
import threading
import time
def long_task():
time.sleep(2)
if __name__ == "__main__":
# 複数のスレッドを起動
for i in range(3):
t = threading.Thread(target=long_task, name=f"Worker-{i}")
t.start()
# 現在動いているスレッド一覧を表示
print(f"現在のアクティブスレッド数: {threading.active_count()}")
print("--- スレッド一覧 ---")
for t in threading.enumerate():
print(f"Thread Name: {t.name}, ID: {t.ident}, Alive: {t.is_alive()}")
# 全てのサブスレッドの終了を待機
for t in threading.enumerate():
if t is not threading.current_thread(): # 自分自身(Main)は待てない
t.join()
まとめ
is_alive() は、スレッドの生死確認を行う最も基本的なメソッドです。
「スレッドが動いている間、プログレスバーを更新する」といったUI周りの制御や、ログ監視などで頻繁に使用されます。
