2026年8月4日

Pythonのマルチスレッドとマルチプロセスの使い分け｜concurrent.futuresで並列処理を書く

Linux基礎・コマンド

Python で処理を並列化するときは、待ち時間の多い処理（通信・ファイル・DB）ならマルチスレッド、計算の重い処理ならマルチプロセスを選ぶのが基本です。理由は、標準の CPython では GIL（グローバルインタプリタロック）によって、Python のコードを同時に実行できるスレッドが1つに限られるためです。この記事では、この使い分けの考え方と、標準ライブラリの concurrent.futures を使った書き方、共有データの扱い、そして Windows や macOS で起きやすい「プロセスが無限に増える」「pickle できない」といったエラーの対処を説明します。

まず処理の種類を見分ける

並列化の前に、遅い原因が「待ち」なのか「計算」なのかを確かめます。

I/O 待ちが中心（I/O バウンド）：Web API の呼び出し、ファイルのダウンロード、DB への問い合わせなど。CPU はほとんど使わず、応答を待っている時間が長い
計算が中心（CPU バウンド）：画像の変換、圧縮、ハッシュ計算、大量の数値計算など。CPU 使用率が1コア分張り付く

見分けるには、処理中に top やタスクマネージャーで CPU 使用率を見るのが手軽です。1コア分をほぼ使い切った状態が続くなら CPU バウンド、ほとんど使っていないなら I/O バウンドです。

GIL があっても、スレッドが I/O を待っている間は GIL が解放され、ほかのスレッドが動けます。そのため I/O バウンドの処理はスレッドで速くなります。一方、CPU バウンドの処理をスレッドに分けても、同時に計算できるのは1スレッドだけなので速くならず、切り替えの分だけ遅くなることもあります。この場合はプロセスを分けて、それぞれのプロセスに CPU コアを使わせます。

なお、GIL を持たない「フリースレッド版」の CPython も提供されるようになっています。3.13 で実験的に導入され、3.14 では正式にサポートされる段階に進みましたが、通常の配布版は引き続き GIL ありです（2026年10月現在）。使っている拡張モジュールの対応状況にも左右されるため、この記事では通常版を前提にします。

スレッドで並列化する：ThreadPoolExecutor

複数の URL から並行してデータを取得する例です。threading.Thread を直接作ることもできますが、concurrent.futures.ThreadPoolExecutor を使うと、同時実行数の制限、結果の受け取り、例外の伝わり方をまとめて扱えます。

import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed

URLS = [
    "https://example.com/",
    "https://example.org/",
    "https://example.net/",
]


def fetch(url: str) -> tuple[str, int]:
    with urllib.request.urlopen(url, timeout=10) as res:
        return url, len(res.read())


with ThreadPoolExecutor(max_workers=5) as executor:
    futures = {executor.submit(fetch, url): url for url in URLS}
    for future in as_completed(futures):
        url = futures[future]
        try:
            _, size = future.result()
            print(f"{url}: {size} bytes")
        except Exception as e:
            print(f"{url}: 失敗 {e}")

ポイントは次のとおりです。

例外は result() で受け取る：スレッドの中で起きた例外は、future.result() を呼んだときに呼び出し元で発生します。result() を呼ばずにいると、エラーに気づけません。
タイムアウトを付ける：通信処理にタイムアウトが無いと、応答しない相手で1スレッドが止まり続けます。
同時実行数を控えめにする：外部の API やサイトに対して大量のスレッドで同時にアクセスすると、相手に負荷をかけ、アクセス制限を受けることもあります。相手の利用規約やレート制限を確認し、max_workers を小さめにします。

結果を入力と同じ順番で受け取りたい場合は、executor.map(fetch, URLS) を使います。ただし map は途中で例外が起きると、その位置で結果の取り出しが止まります。1件の失敗で全体を止めたくないなら、上の submit と as_completed の形が扱いやすいです。

スレッド間でデータを共有するとき

複数のスレッドから同じ変数を書き換えると、更新が失われることがあります。count += 1 は「読む・足す・書く」の複数の手順に分かれており、その途中で別のスレッドが割り込む可能性があるためです。共有する値の更新は threading.Lock で保護します。

import threading
from concurrent.futures import ThreadPoolExecutor

lock = threading.Lock()
total = 0


def work(n: int) -> None:
    global total
    with lock:
        total += n


with ThreadPoolExecutor(max_workers=8) as ex:
    ex.map(work, range(1000))

print(total)  # 499500

もっと安全なのは、共有をやめて、各スレッドの戻り値を呼び出し元で集計する書き方です。上の取得例のように result() で受け取って合計すれば、ロックは要りません。スレッド間でデータを受け渡し続ける処理（生産者と消費者）なら、スレッドセーフな queue.Queue を使います。

プロセスで並列化する：ProcessPoolExecutor

CPU を使う処理は ProcessPoolExecutor で複数のプロセスに分けます。書き方はスレッド版とほぼ同じです。

import hashlib
import os
from concurrent.futures import ProcessPoolExecutor
from pathlib import Path


def sha256_of(path: Path) -> tuple[str, str]:
    h = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            h.update(chunk)
    return path.name, h.hexdigest()


def main() -> None:
    files = [p for p in Path("data").iterdir() if p.is_file()]
    workers = max(1, (os.cpu_count() or 2) - 1)   # 1コアはほかの処理のために残す
    with ProcessPoolExecutor(max_workers=workers) as ex:
        for name, digest in ex.map(sha256_of, files, chunksize=4):
            print(digest, name)


if __name__ == "__main__":
    main()

プロセス版では、次の点がスレッド版と違います。

**if __name__ == "__main__": が要る**：Windows と macOS では、子プロセスがスクリプトを読み込み直して起動します（spawn 方式）。このガードが無いと、子プロセスがさらにプロセスを作ろうとして、エラーやプロセスの増殖が起きます。Linux でも、Python 3.14 から既定の起動方式が fork から forkserver に変わったため、ガードを書いておくのが安全です。
関数と引数は pickle できる必要がある：子プロセスにはデータをコピーして渡します。モジュールの最上位で定義した関数は渡せますが、ラムダ式や関数の中で定義した関数は渡せず、PicklingError などになります。DB 接続やファイルハンドルも渡せないため、子プロセスの中で開きます。
メモリは共有されない：子プロセスでグローバル変数を書き換えても、親には反映されません。結果は戻り値で返します。
受け渡しにコストがかかる：大きなデータを何度も渡すと、コピーの時間で並列化の効果が消えます。ファイルのパスだけを渡して子プロセス側で読む、chunksize で1回に渡す件数をまとめる、といった工夫をします。

max_workers を CPU のコア数より多くしても、計算は速くなりません。メモリの使用量はプロセスの数だけ増えるため、サーバーで動かすときはほかのサービスの分を残して決めます。

asyncio との違い

I/O バウンドの処理には、asyncio という選択肢もあります。1つのスレッドの中で多数の待ちを切り替えるため、数千件の同時接続のように、スレッドでは数が多すぎる場面に向いています。ただし、使うライブラリが非同期に対応している必要があります。既存のコードが同期的なライブラリ（requests など）で書かれていて、同時実行数が数十程度なら、スレッドのほうが書き換えが少なく済みます。

うまく速くならないときの確認

計測してから判断する：time.perf_counter() で並列化の前後を比べます。件数が少ない、1件の処理が短いといった場合は、スレッドやプロセスを作るコストのほうが大きくなります。
ボトルネックが外にないか：DB や API の側が遅い場合、こちらの並列数を増やしても速くならず、相手の負荷が上がるだけです。
ライブラリが内部で並列化していないか：NumPy などは内部で複数のスレッドを使うことがあります。プロセスをさらに分けると、コア数を超える並列になり、かえって遅くなる場合があります。
終了処理：with ブロックを使うと、抜けるときにすべての処理の完了を待ってから後片付けをします。途中で止めたい場合の扱いは、executor.shutdown(cancel_futures=True) などの挙動を公式ドキュメントで確認してください。

まとめ

待ち時間の多い処理はスレッド、計算の重い処理はプロセスで並列化する
concurrent.futures の ThreadPoolExecutor / ProcessPoolExecutor を使うと、書き方をそろえられる
例外は future.result() で受け取り、通信にはタイムアウトを付け、外部への同時アクセス数は控えめにする
共有データは Lock で守るか、戻り値で集計して共有自体をやめる
プロセス版は if __name__ == "__main__": のガード、pickle できる関数と引数、受け渡しのコストに注意する
速くなったかは計測で確かめる