Pythonのthreadingモジュールで並列処理入門！基本から実践まで

こんにちは、とまだです。

Pythonでプログラムを書いていて、「複数の処理を同時に動かせたらいいのに」と思ったことはありませんか？

たとえばファイルのダウンロード中に別の作業を進めたり、データの処理をしながら画面の更新も続けたり。 そんなときに役立つのが、今回紹介するthreadingモジュールです。

この記事では、並列処理の基本的な考え方から、実際のコード例まで分かりやすく解説していきます。

スレッドとは？並列処理の基本概念

まず「スレッド」という言葉を聞いて、難しそうだなと感じる方も多いでしょう。 でも実は、日常生活でも似たような考え方をしています。

日常で考えるスレッドのイメージ

レストランの厨房を想像してみてください。

料理長一人だけで全部の作業をしていたら、お客さんを待たせてしまいますよね。 だから実際の厨房では、こんな風に分担しています。

• シェフA：前菜を作る
• シェフB：メイン料理を作る
• シェフC：デザートを準備する

これがまさにスレッドの考え方です。 一つのプログラム（レストラン）の中で、複数の処理（シェフ）が同時に動いているんです。

Pythonでは、この仕組みをthreadingモジュールで実現できます。

なぜスレッドが必要なのか

プログラムを書いていると、こんな場面に出会います。

# ファイルをダウンロード（5秒かかる）
download_file()
# データを処理（3秒かかる）
process_data()
# 結果を表示
show_result()

このコードだと、合計8秒かかってしまいます。 でもよく考えると、ダウンロード中は待っているだけですよね。

その待ち時間に、別の処理を進められたら効率的です。 これがスレッドを使う大きな理由の一つです。

threadingモジュールの基本的な使い方

では、実際にスレッドを使ってみましょう。 まずは一番シンプルな例から始めます。

最初のスレッドプログラム

以下のコードで、2つの処理を同時に動かしてみます。

import threading
import time

def cook_pasta():
    print("パスタを茹で始めます")
    time.sleep(3)  # 3秒かかる
    print("パスタが茹で上がりました")

def prepare_sauce():
    print("ソースを作り始めます")
    time.sleep(2)  # 2秒かかる
    print("ソースが完成しました")

# スレッドを作成
thread1 = threading.Thread(target=cook_pasta)
thread2 = threading.Thread(target=prepare_sauce)

# スレッドを開始
thread1.start()
thread2.start()

# 両方の処理が終わるまで待つ
thread1.join()
thread2.join()

print("料理が完成しました！")

このコードのポイントは、Threadオブジェクトを作ってstart()で実行することです。 そしてjoin()で、処理が終わるまで待機します。

実行すると、パスタとソースの準備が同時に始まります。 普通に順番に実行したら5秒かかるところが、3秒で終わるんです。

クラスを使ったスレッドの書き方

もう少し複雑な処理では、クラスを使うと整理しやすくなります。

import threading
import time

class CookingThread(threading.Thread):
    def __init__(self, dish_name, cooking_time):
        super().__init__()
        self.dish_name = dish_name
        self.cooking_time = cooking_time

    def run(self):
        print(f"{self.dish_name}の調理を開始")
        time.sleep(self.cooking_time)
        print(f"{self.dish_name}が完成！")

# 複数の料理を同時に調理
dishes = [
    CookingThread("前菜", 2),
    CookingThread("スープ", 3),
    CookingThread("メイン", 4)
]

for dish in dishes:
    dish.start()

for dish in dishes:
    dish.join()

print("全ての料理が完成しました")

run()メソッドに処理を書くのがポイントです。 このように書くと、各スレッドが独自の情報を持てるので便利です。

スレッド間でのデータ共有と注意点

複数のスレッドで同じデータを扱うときは要注意です。 ちょっとした不注意で、予想外のバグが生まれることがあります。

データ競合の問題

こんな例を考えてみましょう。

import threading

# 共有する変数
total_customers = 0

def count_customers():
    global total_customers
    for _ in range(100000):
        total_customers += 1

# 2つのスレッドで同時にカウント
thread1 = threading.Thread(target=count_customers)
thread2 = threading.Thread(target=count_customers)

thread1.start()
thread2.start()
thread1.join()
thread2.join()

print(f"合計来客数: {total_customers}")

期待する結果は200,000人ですよね。 でも実際に実行すると、それより少ない数になることがあります。

なぜでしょうか？

実は、2つのスレッドが同時に同じ変数を書き換えようとして混乱が起きているんです。 まるで2人の店員が同じ伝票に同時に書き込もうとしているようなものです。

Lockで安全にデータを共有する

この問題を解決するには、Lockを使います。

import threading

total_customers = 0
lock = threading.Lock()  # ロックを作成

def count_customers():
    global total_customers
    for _ in range(100000):
        lock.acquire()  # ロックを取得
        total_customers += 1
        lock.release()  # ロックを解放

thread1 = threading.Thread(target=count_customers)
thread2 = threading.Thread(target=count_customers)

thread1.start()
thread2.start()
thread1.join()
thread2.join()

print(f"合計来客数: {total_customers}")

Lockは「使用中」の札のようなものです。 一度に一つのスレッドだけが変数を変更できるようになります。

これで正確に200,000という結果が得られるようになります。

スレッドを安全に停止する方法

スレッドを開始するのは簡単ですが、停止するのは少し工夫が必要です。 強制的に止めることはできないので、スレッド自身に「終了してください」と伝える必要があります。

Eventを使った停止制御

以下の例では、Eventという仕組みを使っています。

import threading
import time

stop_event = threading.Event()

def background_task():
    count = 0
    while not stop_event.is_set():
        count += 1
        print(f"作業中... {count}回目")
        time.sleep(1)

    print("作業を終了します")

# バックグラウンドタスクを開始
worker = threading.Thread(target=background_task)
worker.start()

# 5秒後に停止を指示
time.sleep(5)
print("停止を指示します")
stop_event.set()

worker.join()
print("プログラム終了")

Eventは信号機のようなものです。 赤信号（set）になったら、スレッドは自分で処理を終了します。

この方法なら、スレッドが必要な後処理をしてから安全に終了できます。

実際の活用シーン

threadingモジュールは、どんな場面で役立つのでしょうか。 具体的な例をいくつか紹介します。

Webアプリケーションでの活用

Webサイトから複数のデータを取得する場合を考えてみましょう。

import threading
import time

def fetch_weather(city):
    print(f"{city}の天気を取得中...")
    time.sleep(2)  # API呼び出しをシミュレート
    print(f"{city}の天気: 晴れ")

cities = ["東京", "大阪", "名古屋", "福岡", "札幌"]
threads = []

# 各都市の天気を並行取得
for city in cities:
    t = threading.Thread(target=fetch_weather, args=(city,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print("全都市の天気を取得完了")

順番に取得すると10秒かかるところが、2秒で済みます。 ユーザーを待たせる時間が大幅に短縮できますね。

GUIアプリケーションでの活用

ボタンを押したときに重い処理をする場合、そのままだと画面が固まってしまいます。 スレッドを使えば、画面は動いたまま処理を進められます。

import threading
import time

def heavy_calculation():
    print("重い計算を開始...")
    time.sleep(5)
    print("計算完了！")

def on_button_click():
    # 別スレッドで重い処理を実行
    calc_thread = threading.Thread(target=heavy_calculation)
    calc_thread.start()
    print("ボタンが押されました（画面は固まりません）")

# ボタンクリックをシミュレート
on_button_click()

このように、ユーザー体験を損なわずに処理を実行できます。

スレッドのデバッグとトラブルシューティング

スレッドを使っていると、思わぬ問題に出会うことがあります。 よくある問題と対処法を見ていきましょう。

ログを使った動作確認

どのスレッドがいつ動いたか分からなくなることがあります。 そんなときは、loggingモジュールが便利です。

import threading
import logging
import time

# ログの設定
logging.basicConfig(
    level=logging.DEBUG,
    format='[%(threadName)s] %(message)s'
)

def worker(task_name):
    logging.debug(f"{task_name}を開始")
    time.sleep(2)
    logging.debug(f"{task_name}を完了")

# 複数のタスクを実行
tasks = ["タスクA", "タスクB", "タスクC"]

for task in tasks:
    t = threading.Thread(target=worker, args=(task,))
    t.start()

このように設定すると、どのスレッドがどの順番で動いたか一目で分かります。

スレッド数の管理

スレッドを作りすぎると、かえって遅くなることがあります。 適切な数に制限することが大切です。

import threading
import time
from concurrent.futures import ThreadPoolExecutor

def process_item(item):
    print(f"処理中: {item}")
    time.sleep(1)
    return f"{item}完了"

# スレッド数を5個に制限
with ThreadPoolExecutor(max_workers=5) as executor:
    items = [f"アイテム{i}" for i in range(20)]
    results = executor.map(process_item, items)

    for result in results:
        print(result)

ThreadPoolExecutorを使えば、スレッド数を簡単に管理できます。

よくある質問

Q: スレッドとプロセスの違いは？

スレッドは同じメモリ空間を共有しますが、プロセスは完全に独立しています。 アパートの部屋（スレッド）と一軒家（プロセス）の違いのようなものです。

軽い処理や素早い応答が必要ならスレッド。 重い計算処理ならプロセスが向いています。

Q: GILって何？

PythonにはGIL（Global Interpreter Lock）という仕組みがあります。 これにより、同時に実行できるPythonコードは実質1つだけです。

でも、ファイル読み込みやネットワーク通信の待ち時間では、GILが解放されます。 だからI/O処理ではスレッドが有効なんです。

Q: スレッドセーフとは？

複数のスレッドから同時にアクセスしても安全なことを「スレッドセーフ」と言います。 Pythonのlist.append()などは基本的にスレッドセーフです。

ただし、複雑な操作では自分でLockを使う必要があります。

まとめ

threadingモジュールを使えば、Pythonでも並列処理が実現できます。

重要なポイントをおさらいしましょう。

• スレッドはThreadクラスで作成し、start()で開始する
• 共有データを扱うときはLockで保護する
• スレッドの停止にはEventなどを使う
• I/O待機が多い処理で特に効果的

最初は小さなプログラムから始めて、徐々に複雑な処理に挑戦してみてください。

並列処理の考え方を身につければ、より効率的なプログラムが書けるようになります。 ぜひ実際にコードを書いて、スレッドの動きを体感してみてください。