Pythonのthreadingモジュールでマルチスレッドを実装する方法

1. 並列化する処理と順番に動かす処理を分ける

1. 並列化する処理と順番に動かす処理を分ける

threadingモジュールでマルチスレッドは、ただライブラリ名を覚えるだけでは実務で使いにくいテーマです。まずは I/O待ちが多い処理を複数スレッドで進める作業 という具体例に置き換えて、入力、処理、出力、エラー時の動きを確認します。

from threading import Thread, Lock

counter = 0
lock = Lock()

def increment():
    global counter
    for _ in range(1000):
        with lock:
            counter += 1

threads = [Thread(target=increment) for _ in range(5)]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()

print(counter)

2. threadingの基本構造を小さなコードで確認する

2. threadingの基本構造を小さなコードで確認する

threadingを使う前に、インストール方法、実行環境、扱うデータの範囲を整理しておきます。特にネットワーク、メール、IoT、セキュリティ系は、自分が管理する環境で試すことが大切です。

from time import perf_counter, sleep

start = perf_counter()
sleep(0.2)
elapsed = perf_counter() - start

print(f"{elapsed:.3f} seconds")

3. I/O待ちとCPU負荷で選ぶ並列処理の違い

3. I/O待ちとCPU負荷で選ぶ並列処理の違い

Pythonコードでは、正常に動くケースだけでなく、タイムアウト、ファイルなし、認証失敗、入力ミスなども想定します。例外処理を入れておくと、学習用コードでも原因を追いやすくなります。

from concurrent.futures import ThreadPoolExecutor, as_completed

def task(value: int) -> int:
    return value * 2

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(task, i) for i in range(5)]
    for future in as_completed(futures):
        print(future.result())

4. 共有データ・ロック・キューで競合を防ぐ

4. 共有データ・ロック・キューで競合を防ぐ

処理結果は、画面に表示するだけでなく、ログやファイルに残すと確認しやすくなります。あとから失敗原因を追えるように、日時、対象、結果を記録する習慣をつけましょう。

import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)

logger.info("処理を開始しました")

5. 例外・キャンセル・タイムアウトを安全に扱う

5. 例外・キャンセル・タイムアウトを安全に扱う

同じ処理を何度も書く場合は、関数や小さなクラスへ分けます。関数名に目的を入れると、あとで見返したときに「何をするコードか」がすぐ分かります。

from pathlib import Path

output = Path("result.txt")
output.write_text("処理結果を保存しました\n", encoding="utf-8")

print(output.read_text(encoding="utf-8"))

6. 処理時間を測って効果を確認する方法

6. 処理時間を測って効果を確認する方法

最後に、I/O待ちが多い処理を複数スレッドで進める作業 として動く形へまとめます。コードを写すだけでなく、対象データや引数を変えて試すと、検索されるだけでなく読者の学習にも役立つ記事になります。

def validate_required(value: str, label: str) -> list[str]:
    errors = []
    if not value.strip():
        errors.append(f"{label}は必須です")
    return errors

print(validate_required("Python", "キーワード"))

7. 並列処理として動かす完成コード

7. 並列処理として動かす完成コード

I/O待ちが多い処理を複数スレッドで進める作業として使うなら、サンプルコードを1回動かして終わりにせず、対象データ、ファイル名、URL、認証情報、実行環境を変えて確認しましょう。threadingの役割が見えると、応用もしやすくなります。

実務では、正常系よりも失敗時の扱いで差が出ます。ログ、例外、タイムアウト、入力チェックを入れておくことで、読者がそのまま練習しやすいコードになります。

8. 本番処理へ入れる前のデバッグポイント

8. 本番処理へ入れる前のデバッグポイント

threadingモジュールでマルチスレッドを身につけるには、まず小さなコードで動きを確認し、次にログ出力、設定ファイル、テスト、自動実行へ広げるのがおすすめです。記事を読んだ人が自分の環境で試せるように、値を変えやすい形で残しておきましょう。

次の練習では、この記事のコードに設定ファイル読み込み、CSV保存、pytestでの確認、GitHub Actionsでの自動実行のどれか1つを追加してみてください。