【Python】Threadクラスを継承して独自の処理スレッドを作る

2026年1月14日

概要

Pythonの threading モジュールでマルチスレッド処理を実装する際、Thread クラスを継承（サブクラス化）する方法を解説します。単に関数を target に指定する方法と比較して、スレッド自体に独自のデータ（状態）を持たせたり、複雑なロジックをクラス内に隠蔽してコードの可読性を高めたりする場合に有効な手法です。

仕様（入出力）

入力: スレッド固有の識別名やパラメータ（初期化時）
出力: 各スレッドによる並行処理の実行ログ
前提: threading.Thread を継承し、run() メソッドをオーバーライドする実装パターン。

基本の使い方

threading.Thread を継承したクラスを作成し、run メソッドに実行したい処理を記述します。

import threading

class MyThread(threading.Thread):
    def run(self):
        # start()実行時にこのメソッドが別スレッドで呼ばれます
        print("スレッド処理実行中")

t = MyThread()
t.start() # run()を直接呼ばず、必ずstart()を呼ぶ
t.join()

コード全文

「数字を出力するスレッド」と「文字を出力するスレッド」をそれぞれクラスとして定義し、並行動作させる完全な実装例です。 __init__ で名前を受け取り、インスタンス変数として保持させています。

import threading
import time

class NumberPrinterThread(threading.Thread):
    """数字をカウントアップ出力するスレッドクラス"""
    
    def __init__(self, thread_name, loops=5):
        # 親クラス(Thread)の初期化を必ず最初に行う
        super().__init__()
        self.thread_name = thread_name
        self.loops = loops

    def run(self):
        """スレッドで実行されるメイン処理"""
        print(f"[{self.thread_name}] 開始")
        for i in range(self.loops):
            print(f"{self.thread_name}: {i}")
            time.sleep(1)
        print(f"[{self.thread_name}] 完了")

class LetterPrinterThread(threading.Thread):
    """リストの文字を順に出力するスレッドクラス"""

    def __init__(self, thread_name, chars=None):
        super().__init__()
        self.thread_name = thread_name
        # デフォルト引数の設定
        self.chars = chars if chars else ["A", "B", "C", "D", "E"]

    def run(self):
        print(f"[{self.thread_name}] 開始")
        for letter in self.chars:
            print(f"{self.thread_name}: {letter}")
            time.sleep(1.5)
        print(f"[{self.thread_name}] 完了")

if __name__ == "__main__":
    # 各クラスのインスタンスを生成
    thread1 = NumberPrinterThread("Num-Thread")
    thread2 = LetterPrinterThread("Char-Thread")

    # スレッド開始
    # start() を呼ぶことで、内部的に run() が別スレッドで実行されます
    thread1.start()
    thread2.start()

    # 終了待機
    thread1.join()
    thread2.join()

    print("全ての処理が終了しました")

カスタムポイント

サブクラス化の必須ルール

run() メソッドのオーバーライド
- スレッドが開始されたときに実行される処理をここに記述します。
- ここに書かれたコードだけが別スレッドで動きます。
__init__() での super().__init__()
- コンストラクタ（初期化メソッド）を定義する場合は、必ず最初に親クラスの初期化 super().__init__() を呼んでください。
- これを忘れるとスレッドとしての機能（start や name 属性など）が正しく初期化されず、エラーになります。

メリット

データの保持: self.data のように、スレッド固有のデータをインスタンス変数として管理しやすくなります。
構造化: 複雑な処理をクラス内にメソッドとして分割できるため、大規模なスレッド処理の見通しが良くなります。

注意点

run() を直接呼ばない
- インスタンスの .run() メソッドを直接呼び出すと、新しいスレッドは作成されず、メインスレッド上で同期的に実行されてしまいます。
- 必ず .start() メソッドを呼び出してください。
スレッドセーフ
- クラス化したからといって、共有リソースへのアクセスが自動的に安全になるわけではありません。
- 複数のスレッドから同じオブジェクトの属性を書き換える場合は、threading.Lock などでの排他制御が必要です。

応用

スレッドを外部から安全に停止できるように、停止フラグを持たせた応用例です。クラス化することで、このような「制御用の状態」をきれいに管理できます。

import threading
import time

class StoppableThread(threading.Thread):
    def __init__(self):
        super().__init__()
        self._stop_event = threading.Event()

    def stop(self):
        """外部からスレッドを停止させるためのメソッド"""
        self._stop_event.set()

    def run(self):
        print("処理開始...")
        while not self._stop_event.is_set():
            print("実行中...")
            time.sleep(1)
            # ここにメインの処理を書く
        print("停止指示を受け取りました。終了します。")

# 使用例
if __name__ == "__main__":
    t = StoppableThread()
    t.start()
    time.sleep(3)
    t.stop() # 外部から停止
    t.join()