Pythonの並行処理を整理する — ThreadPoolExecutor・ProcessPoolExecutor・asyncioの使い分け

Pythonで並行処理・並列処理を調べ始めると、必ずと言っていいほど ThreadPoolExecutor ・ProcessPoolExecutor ・asyncio の3つに出会います。名前だけを見ると「結局どれを使えばいいの？」となりがちですが、実は判断基準はシンプルです。その処理が「何かを待っている」のか、それとも「CPUが計算し続けている」のか ―― これさえ見極められれば、使い分けは一気に整理できます。

この記事では、それぞれの違いを実例付きで整理し、実務でどう組み合わせて使うかまで解説します。

先に結論だけ書くと、ざっくり次のように使い分けます。

処理の種類	向いているもの
HTTP通信、DBアクセス、ファイルI/Oなど	asyncio
同期ライブラリによるI/O処理	ThreadPoolExecutor
CPU負荷の高い計算	ProcessPoolExecutor

ただし実際の開発では、asyncio と ThreadPoolExecutor を組み合わせて使うことも多くあります。この組み合わせ方が今回の記事の後半のテーマです。

---

「並行」と「並列」を分けて考える

最初にこの2つの言葉を区別しておくと、以降の説明が理解しやすくなります。

並行処理は、複数の処理を少しずつ切り替えながら進めることです。1人の店員が複数のお客さんを掛け持ちで担当するイメージに近く、料理を待っている間に別の注文を聞く、というように待ち時間を有効活用します。asyncio はこの考え方に近い仕組みです。

並列処理は、複数の処理を文字通り同時に実行することです。CPUのコアごとに別の処理を割り当てるイメージで、Pythonでは主にCPUバウンドな処理を並列化する際に ProcessPoolExecutor が使われます。

---

1. ThreadPoolExecutor — I/O待ちが多い処理に向く

ThreadPoolExecutor は複数のスレッドを使って処理を実行する仕組みです。特に向いているのは、HTTP通信・ファイル読み込み・DBアクセス・外部API呼び出し・time.sleep() など、I/O待ちが多い処理です。

from concurrent.futures import ThreadPoolExecutor
import time


def task(name):
    print(f"{name}: start")
    time.sleep(2)
    print(f"{name}: end")
    return name


with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [
        executor.submit(task, "A"),
        executor.submit(task, "B"),
        executor.submit(task, "C"),
    ]
    results = [future.result() for future in futures]

print(results)

3つの処理はそれぞれ2秒かかりますが、順番に実行すれば合計約6秒かかるところ、3スレッドで並行実行すればおよそ2秒程度で完了します。

単純に同じ関数へ複数の値を渡したいだけなら、map() を使うともう少し簡潔に書けます。

from concurrent.futures import ThreadPoolExecutor
import time


def task(name):
    time.sleep(2)
    return f"{name}: done"


with ThreadPoolExecutor(max_workers=3) as executor:
    results = executor.map(task, ["A", "B", "C"])

print(list(results))

なぜCPU計算には向かないのか

ここでPython特有の話として、GIL（Global Interpreter Lock）が関係してきます。CPythonでは、複数スレッドが存在していても、Pythonのバイトコードを同時に実行できるスレッドは基本的に1つだけです。そのため、次のようなCPUを使い続ける処理をスレッドに分割しても、期待したほど高速化されません。

def calculate():
    total = 0
    for i in range(100_000_000):
        total += i
    return total

CPUバウンドな処理には、次に紹介する ProcessPoolExecutor が向いています。

---

2. ProcessPoolExecutor — CPU負荷の高い処理に向く

ProcessPoolExecutor はスレッドではなく複数のプロセスを使う仕組みです。プロセスが分かれるため複数のCPUコアを活用でき、数値計算・画像処理・データ変換・圧縮・シミュレーションといったCPU負荷の高い処理に向いています。

from concurrent.futures import ProcessPoolExecutor


def heavy_task(n):
    total = 0
    for i in range(n):
        total += i * i
    return total


if __name__ == "__main__":
    values = [20_000_000, 20_000_000, 20_000_000, 20_000_000]

    with ProcessPoolExecutor(max_workers=4) as executor:
        results = executor.map(heavy_task, values)

    print(list(results))

ここでのポイントは if __name__ == "__main__": のガードです。特にWindowsやmacOSでは、ProcessPoolExecutor を使う際に子プロセス生成時にモジュールが再読み込みされるため、このガードがないと意図しないプロセス生成につながる可能性があります。

ThreadPoolExecutorとの違い

	ThreadPoolExecutor	ProcessPoolExecutor
実行単位	スレッド	プロセス
メモリ	基本的に共有	原則別
GIL	影響を受ける	プロセスごとに独立
起動コスト	比較的小さい	大きい
I/O処理	得意	通常は不要
CPU計算	苦手なことが多い	得意
データ受け渡し	比較的簡単	シリアライズが必要

一言でまとめると、Thread は軽いがGILの影響を受け、Process は重いがCPU並列処理ができるという違いです。

---

3. asyncio — 待ち時間が大量に発生する処理に向く

asyncio は ThreadPoolExecutor や ProcessPoolExecutor とは考え方が異なり、基本的には1つのイベントループ上で処理を切り替えながら実行します。得意なのは、待ち時間が大量に発生する処理です。

import asyncio


async def task(name):
    print(f"{name}: start")
    await asyncio.sleep(2)
    print(f"{name}: end")
    return name


async def main():
    results = await asyncio.gather(
        task("A"),
        task("B"),
        task("C"),
    )
    print(results)


asyncio.run(main())

3つのタスクはそれぞれ2秒待ちますが、並行して実行されるため全体では約2秒で完了します。

awaitが意味すること

await asyncio.sleep(2) は、単に「2秒間何もしない」という意味ではありません。より正確には「自分は2秒待つ必要があるので、その間ほかのタスクを動かしていいですよ」という合図です。この合図によって、イベントループはタスクAが通信待ちになった隙にタスクBを動かし、タスクBも待ちに入ればタスクCを動かす、というように処理を切り替えていきます。

time.sleepとasyncio.sleepはまったく違う

これは非常に重要な注意点です。次のコードは良くない例です。

import time


async def task():
    time.sleep(5)

time.sleep() はスレッドそのものを止めてしまうため、asyncio のイベントループごと止まり、その間は他のタスクも一切実行できなくなります。非同期処理の中では、必ず await asyncio.sleep(5) を使う必要があります。

---

asyncioとThreadPoolExecutorを組み合わせる

ここからは実務でかなり重要になる話です。asyncio の中で、非同期対応していない同期関数を使いたい場面がよくあります。例えば requests のような同期HTTPクライアントを、そのまま async 関数の中で呼んでしまうと次のようになります。

import requests


async def get_data():
    response = requests.get("https://example.com")
    return response.text

これでは requests.get() が完了するまでイベントループそのものが止まってしまい、せっかくの asyncio の利点が失われます。

asyncio.to_threadで同期関数をスレッドに逃がす

この場合、asyncio.to_thread() を使うと同期関数を別スレッドに逃がすことができます。

import asyncio
import requests


async def get_data():
    response = await asyncio.to_thread(
        requests.get,
        "https://example.com",
    )
    return response.text

asyncio → ThreadPool → requests.get() という構造になり、同期処理は別スレッドで実行されるため、その間イベントループは別の非同期タスクを進めることができます。

複数の同期処理をまとめて実行したい場合は、asyncio.gather() と組み合わせます。

import asyncio
import time


def blocking_task(name):
    print(f"{name}: start")
    time.sleep(3)
    print(f"{name}: end")
    return name


async def main():
    results = await asyncio.gather(
        asyncio.to_thread(blocking_task, "A"),
        asyncio.to_thread(blocking_task, "B"),
        asyncio.to_thread(blocking_task, "C"),
    )
    print(results)


asyncio.run(main())

より低レベルなrun_in_executor

asyncio.to_thread() は内部的に run_in_executor() を使っており、こちらを直接使うことでスレッドプールを自分で管理することもできます。

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor


def blocking_task(name):
    time.sleep(2)
    return name


async def main():
    loop = asyncio.get_running_loop()

    with ThreadPoolExecutor(max_workers=3) as executor:
        futures = [
            loop.run_in_executor(executor, blocking_task, name)
            for name in ["A", "B", "C"]
        ]
        results = await asyncio.gather(*futures)

    print(results)


asyncio.run(main())

asyncioからProcessPoolExecutorを呼ぶ

CPUバウンドな処理であれば、同じ仕組みで ProcessPoolExecutor を呼び出すこともできます。

import asyncio
from concurrent.futures import ProcessPoolExecutor


def cpu_task(n):
    total = 0
    for i in range(n):
        total += i * i
    return total


async def main():
    loop = asyncio.get_running_loop()

    with ProcessPoolExecutor() as executor:
        result = await loop.run_in_executor(executor, cpu_task, 20_000_000)

    print(result)


if __name__ == "__main__":
    asyncio.run(main())

Webサーバーのような実運用のシステムでは、次のように複数の仕組みを組み合わせて使うことがあります。

flowchart TD
    A["asyncio イベントループ"] --> B["HTTP通信"]
    A --> C["DBアクセス"]
    A --> D["WebSocket"]
    A --> E["同期SDK呼び出し"]
    E --> F["ThreadPoolExecutor"]
    A --> G["CPU負荷の高い処理"]
    G --> H["ProcessPoolExecutor"]

---

お店の例えで直感的に理解する

個人的には、店員に例えるのが一番理解しやすいと感じています。

asyncio は店員が1人だけのお店です。料理ができるのを待っている間に、別のお客さんの注文を聞きに行きます。待ち時間が多い仕事に強い一方、店員自身が調理（CPU計算）をし続けると他のお客さんの対応ができなくなります。

ThreadPoolExecutor は店員を複数人用意するイメージです。それぞれが並行してお客さんを担当できますが、全員が同じ厨房（GIL）を共有しているため、Pythonコードそのものをガリガリ計算するような仕事では思ったほど速くなりません。

ProcessPoolExecutor は店舗そのものを複数用意するイメージです。それぞれが完全に独立して調理できるため複数のCPUコアを活かせますが、店舗を新しく構える分、起動のコストはスレッドより大きくなります。

---

よくある勘違い

「async defを付ければ非同期になる」は間違い

async def heavy():
    total = 0
    for i in range(100_000_000):
        total += i

async def を付けていても、処理の途中に await がなければイベントループに制御が返らず、この計算中は他のタスクを一切実行できません。async def を付けることが自動的に並行処理になるわけではなく、await できるポイントがあって初めて asyncio は効率よく動きます。

awaitを順番に書くだけでは同時実行されない

async def main():
    await task("A")
    await task("B")
    await task("C")

このコードは「Aが終わってからBを開始し、Bが終わってからCを開始する」という逐次実行になります。複数を並行して動かしたい場合は、必ず asyncio.gather() を使う必要があります。

async def main():
    await asyncio.gather(
        task("A"),
        task("B"),
        task("C"),
    )

---

具体的な使い分けの判断

• HTTP APIを大量に呼びたい：非同期対応のHTTPクライアントがあるなら asyncio.gather() で並行実行する
• 同期APIクライアントしか存在しない：既存SDKが client.get_data() のような同期APIしか持っていない場合は、await asyncio.to_thread(client.get_data) として asyncio と ThreadPoolExecutor を組み合わせる
• 大量の画像をCPUで変換する：処理そのものがCPUバウンドなら ProcessPoolExecutor が候補になる
• Webサーバー全体の設計：リクエスト受付を asyncio にまとめ、DB・API・Redis通信はasync対応ライブラリで直接扱い、同期SDKは ThreadPoolExecutor に逃がし、CPU負荷の高い処理だけ ProcessPoolExecutor に切り出す、という構成にすることが多い

判断に迷ったときは、次のフローチャートで考えると整理しやすくなります。

flowchart TD
    A["処理に時間がかかる"] --> B{"主に何を待っている？"}
    B -->|I/O待ち| C{"async対応ライブラリがある？"}
    B -->|CPU計算| D["ProcessPoolExecutor"]
    C -->|Yes| E["asyncio"]
    C -->|No| F["ThreadPoolExecutor"]

---

まとめ

選択肢	一言で言うと	向いている処理
ThreadPoolExecutor	ブロッキングする同期処理を複数スレッドで実行する	同期HTTP通信、同期DBアクセス、ファイルI/O、古い同期SDK
ProcessPoolExecutor	CPU負荷の高い処理を複数プロセスで並列実行する	数値計算、画像処理、重いデータ加工、シミュレーション
asyncio	I/O待ちの間に別の処理を進める	HTTP通信、DB通信、WebSocket、大量のネットワークI/O

実務では、どれか1つだけを選ぶというより、asyncio をベースに、async未対応の同期I/Oは ThreadPoolExecutor に、CPUバウンド処理は ProcessPoolExecutor に逃がす、という組み合わせ方をすることも多くあります。

一番重要なのは、その処理が「何かを待っている」のか、それとも「CPUが計算し続けている」のかを見極めることです。ここさえ判断できれば、ThreadPoolExecutor ・ProcessPoolExecutor ・asyncio の使い分けはシンプルになります。

なお、実際にRedisキューとPython Workerを使ってCSV処理を非同期化した実装パターンは、以下の記事で詳しく解説しています。

Laravel + Python Workerでファイル処理を非同期化する方法

参考リンク

• concurrent.futures — 並行タスクの非同期実行 — Python公式ドキュメント
• asyncio — 非同期I/O — Python公式ドキュメント
• Global Interpreter Lock（GIL）とは — Python公式用語集