Python並列処理｜concurrent.futures使い分けと速度比較

Pythonの並列処理とは、複数の仕事を同時に進めて全体の待ち時間を短くする書き方で、I/O待ちならThreadPoolExecutor（またはasyncio）、CPU計算ならProcessPoolExecutorを選ぶのが基本です。 標準ライブラリの concurrent.futures を使えば、スレッドとプロセスをほぼ同じ書き方で切り替えられます。

この記事では、5つの方式（ThreadPoolExecutor／ProcessPoolExecutor／asyncio／multiprocessing.Pool／Python 3.14で追加されたInterpreterPoolExecutor）を同じ課題で実測した比較表と、どれを選ぶかの判断フローを最初に示します。そのあと、submit・map・as_completed・wait、タイムアウト、例外、キャンセル、max_workersの決め方をコピペで動くコードと実行結果つきで解説し、if __name__ == "__main__" 忘れやpickleのエラーなど、つまずきやすい箇所を実際のエラー文で確認します。

検証環境：Linux（2 vCPU）、Python 3.11.15／Python 3.14.7（uv）／Python 3.14.0rc2 free-threading版（uvの 3.14t）。

Python並列処理の早見表

まず判断フローで方式を選ぶ

迷ったら「待ち時間が長いのか、計算が重いのか」で分けます。 I/O待ち（ネットワーク、ディスク、DB）の間はスレッドがGILを手放すので、スレッドで十分速くなります。CPU計算はGIL（同時に1スレッドしかPythonコードを実行できない仕組み）があるため、通常版Pythonではプロセスかサブインタプリタで分けます。

flowchart TD
    A[重いのはどこ？] -->|ネットワーク・DB・ファイル待ち| B{同時数は？}
    B -->|数十〜数百| C[ThreadPoolExecutor]
    B -->|数千以上・async対応ライブラリあり| D[asyncio]
    A -->|Pythonでの計算| E{Pythonの版は？}
    E -->|3.14t free-threading| F[ThreadPoolExecutor]
    E -->|通常版| G{データの受け渡しは？}
    G -->|少ない・関数で完結| H[ProcessPoolExecutor<br/>または InterpreterPoolExecutor 3.14+]
    G -->|imap_unordered などを使いたい| I[multiprocessing.Pool]
    A -->|NumPyなどC拡張の計算| J[まずThreadPoolで試す<br/>GILを手放す処理が多い]

同じ課題で実測した速度比較表

I/O待ちはスレッドとasyncioが約20倍速く、CPU計算はプロセス系が約2倍（2コア）速くなりました。 課題は次の3つです。ローカルHTTPサーバーは http.server.ThreadingHTTPServer をポート7910で起動し、1リクエストごとに time.sleep(0.2) してから応答するようにしました。

# bench.py：課題の定義（抜粋）
import asyncio, time, urllib.request

def io_sleep(i):                 # I/O待ち1：0.2秒待つ
    time.sleep(0.2)
    return i

def io_http(i):                  # I/O待ち2：ローカルHTTP（0.2秒で応答）
    with urllib.request.urlopen("http://127.0.0.1:7910/", timeout=5) as r:
        return r.read()

def cpu_task(n):                 # CPU計算：純Pythonのループ
    total = 0
    for i in range(n):
        total += i * i
    return total

async def a_sleep(i):            # asyncio版（HTTPは asyncio.open_connection で同様に）
    await asyncio.sleep(0.2)
    return i

計測は、プールを先に起動してから各方式を5回実行し、中央値をとりました（I/Oは20件、CPUは500万回ループ×4件）。

表から読み取れるポイントです。

• I/O待ちは版を問わずスレッドで約20倍：待っている間はGILが解放されるため、通常版でも十分です。
• ProcessPool(2)でI/Oを並べると2倍止まり：同時に動けるのがワーカー数（2）までだからです。I/O待ちのためにプロセスを増やすより、スレッドのほうが軽く済みます。
• CPU計算はスレッドでは速くならない（通常版）：GILがあるためです。ProcessPool・Pool・InterpreterPoolはコア数（2）ぶん約2倍になりました。
• 3.14tのスレッドはCPU計算でも並列に動く：単独で計測した回は逐次1.01秒→ThreadPool(2) 0.51秒でした。ただしこの環境は他の処理とCPUを共有しており、3.14tのCPU行は計測ごとのぶれが大きかったため、参考値として見てください。
• 3.14からLinuxの既定の起動方式がforkからforkserverに変わったため、プールの起動に少し時間がかかります（0.01秒→0.11秒）。プールは使い回すと影響が小さくなります。

concurrent.futuresの基本：submit・map・as_completed・wait

map は入力順、as_completed は終わった順、wait は条件がそろった時点で戻ります。 ThreadPoolExecutorとProcessPoolExecutorは同じメソッドを持つので、クラス名を差し替えるだけで切り替えられます。

# demo_basic.py（Python 3.8+）
import time
from concurrent.futures import ThreadPoolExecutor, as_completed, wait, FIRST_COMPLETED

def work(sec):
    time.sleep(sec)
    return f"{sec}秒の仕事"

secs = [0.3, 0.1, 0.2]

with ThreadPoolExecutor(max_workers=3) as ex:
    # 1) map：結果は「入力の順番」で返る
    print("map         :", list(ex.map(work, secs)))

    # 2) submit + as_completed：終わった順に受け取る
    futures = {ex.submit(work, s): s for s in secs}
    print("as_completed:", [f.result() for f in as_completed(futures)])

    # 3) wait：最初の1つが終わった時点で戻る
    futures = [ex.submit(work, s) for s in secs]
    done, not_done = wait(futures, return_when=FIRST_COMPLETED)
    print("wait        :", len(done), "件完了 /", len(not_done), "件実行中")

実行結果（Python 3.14.7）：

map         : ['0.3秒の仕事', '0.1秒の仕事', '0.2秒の仕事']
as_completed: ['0.1秒の仕事', '0.2秒の仕事', '0.3秒の仕事']
wait        : 1 件完了 / 2 件実行中

with 文を使うと、ブロックを抜けるときに shutdown(wait=True) が呼ばれ、すべての仕事が終わるまで待ってくれます。

例外・タイムアウト・キャンセルの扱い

ワーカー内の例外は future.result() を呼んだときにメイン側で同じ例外として再送出されます。 map は例外が出た位置で反復が止まるので、1件の失敗で全体を止めたくない場合は submit と as_completed を使います。

# demo_error.py（Python 3.8+）
import time
from concurrent.futures import ThreadPoolExecutor, as_completed, TimeoutError

def fetch(item_id):
    if item_id == 3:
        raise ValueError(f"商品ID {item_id} が見つかりません")
    time.sleep(0.1)
    return {"id": item_id, "stock": item_id * 10}

with ThreadPoolExecutor(max_workers=4) as ex:
    futures = {ex.submit(fetch, i): i for i in range(1, 6)}
    for f in as_completed(futures):
        item_id = futures[f]
        try:
            print("OK ", f.result())
        except ValueError as e:
            print("NG ", item_id, type(e).__name__, e)

# map は例外が出た位置で止まる
with ThreadPoolExecutor(max_workers=4) as ex:
    results = ex.map(fetch, range(1, 6))
    try:
        for r in results:
            print("map:", r)
    except ValueError as e:
        print("map が途中で止まった:", e)

# タイムアウト
with ThreadPoolExecutor(max_workers=1) as ex:
    f = ex.submit(time.sleep, 1)
    try:
        f.result(timeout=0.3)
    except TimeoutError:
        print("0.3秒で打ち切り。cancel() の戻り値:", f.cancel(), "（実行中は取り消せない）")

実行結果（Python 3.14.7）：

NG  3 ValueError 商品ID 3 が見つかりません
OK  {'id': 1, 'stock': 10}
OK  {'id': 2, 'stock': 20}
OK  {'id': 4, 'stock': 40}
OK  {'id': 5, 'stock': 50}
map: {'id': 1, 'stock': 10}
map: {'id': 2, 'stock': 20}
map が途中で止まった: 商品ID 3 が見つかりません
0.3秒で打ち切り。cancel() の戻り値: False （実行中は取り消せない）

result(timeout=...) は「待つのをやめる」だけで、実行中の仕事は止まりません。 スレッドは外から止められないため、時間制限が必要な処理は、関数の中で requests.get(url, timeout=5) のように呼び出し先にタイムアウトを渡すのが確実です。ProcessPoolExecutorなら3.14で追加された terminate_workers()／kill_workers() でワーカーごと終了させる手もあります。

未着手の仕事をまとめて取り消す

# demo_cancel.py（cancel_futures は Python 3.9+）
import time
from concurrent.futures import ThreadPoolExecutor

def task(i):
    time.sleep(0.2)
    return i

ex = ThreadPoolExecutor(max_workers=2)
futures = [ex.submit(task, i) for i in range(10)]
time.sleep(0.05)
print("未着手の3件目を cancel():", futures[2].cancel())
ex.shutdown(wait=True, cancel_futures=True)   # 未着手をまとめて取り消す
print("完了:", sum(f.done() and not f.cancelled() for f in futures),
      "/ 取り消し:", sum(f.cancelled() for f in futures))

実行結果（Python 3.14.7）：

未着手の3件目を cancel(): True
完了: 2 / 取り消し: 8

実行中の2件は最後まで動き、キューに残っていた8件が取り消されました。 Ctrl+Cやエラー時に早く抜けたいときは shutdown(cancel_futures=True) が便利です。

max_workersの決め方

既定値は、ThreadPoolExecutorが min(32, CPU数 + 4)、ProcessPoolExecutorが CPU数 です。 3.13からは、実際に使えるCPU数を返す os.process_cpu_count() が基準になりました。

# Python 3.13+
import os
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

print("os.cpu_count()        =", os.cpu_count())
print("os.process_cpu_count()=", os.process_cpu_count())
print("ThreadPool 既定        =", ThreadPoolExecutor()._max_workers)
print("ProcessPool 既定       =", ProcessPoolExecutor()._max_workers)

実行結果（2 vCPU、Python 3.14.7。_max_workers は確認用の内部属性です）：

os.cpu_count()        = 2
os.process_cpu_count()= 2
ThreadPool 既定        = 6
ProcessPool 既定       = 2

I/O待ちは「相手が受けられる数」で決めるのがコツです。2 vCPUの既定値（6）で sleep 0.2秒×20件を流すと 0.80秒（4巡）、max_workers=20 なら 0.20秒 でした。全件を同時に待たせたいときは明示します。

巨大な入力は buffersize で少しずつ流す（3.14+）

3.14から map に buffersize が追加され、入力を全部読み込まずに処理を始められるようになりました。

# buffersize.py（Python 3.14+）
import time
from concurrent.futures import ThreadPoolExecutor

def gen_ids():
    for i in range(1_000_000):
        yield i

def work(i):
    return i * 2

with ThreadPoolExecutor(4) as ex:
    t = time.perf_counter()
    it = ex.map(work, gen_ids(), buffersize=8)
    print("最初の結果:", next(it), f"({time.perf_counter()-t:.4f} 秒)")

3.11で buffersize を渡すと TypeError: Executor.map() got an unexpected keyword argument 'buffersize' になります。3.13以前では、入力を数千件ずつに区切って map を呼ぶと同じ効果が得られます。

つまずきやすいポイントと実際のエラー文

if name == "main" を書き忘れる

ProcessPoolExecutorとmultiprocessingは、メインの処理を if __name__ == "__main__": の中に入れます。 spawn・forkserver方式では子プロセスがメインのファイルを読み直すため、ガードがないと子プロセスがさらにプールを作ろうとします。

# noguard.py（直す前）
from concurrent.futures import ProcessPoolExecutor

def square(x):
    return x * x

with ProcessPoolExecutor(max_workers=2) as ex:   # ガードがない
    print(list(ex.map(square, range(5))))

spawn方式（Windows・macOSの既定）で実行した結果（Python 3.14.7、抜粋）：

RuntimeError:
        An attempt has been made to start a new process before the
        current process has finished its bootstrapping phase.

        This probably means that you are not using fork to start your
        child processes and you have forgotten to use the proper idiom
        in the main module:

            if __name__ == '__main__':

Linuxの3.14（forkserver）では同じ RuntimeError に加えて ConnectionResetError: [Errno 104] Connection reset by peer も表示されました。一方、3.11のLinux（fork）では [0, 1, 4, 9, 16] と動いてしまうため、Linuxで動いていたコードが3.14やWindowsでエラーになる、という形で気づくことが多い箇所です。

# 直した後
from concurrent.futures import ProcessPoolExecutor

def square(x):
    return x * x

if __name__ == "__main__":
    with ProcessPoolExecutor(max_workers=2) as ex:
        print(list(ex.map(square, range(5))))

pickleできないものを渡す

プロセスやサブインタプリタへは、関数・引数・戻り値をpickle（バイト列への変換）で送ります。 lambdaやロックなどはpickleできません。

3.11ではlambdaのエラー文が attribute lookup <lambda> on __main__ failed でした。直し方は、関数をモジュールの先頭（トップレベル）に def で定義することです。ファイルやDB接続は、パスや接続情報（文字列）を渡してワーカーの中で開き直すか、initializer 引数でワーカー起動時に1回だけ開きます。

fork・forkserver・spawnの違い

forkは親プロセスの状態をコピーし、forkserver・spawnは新しく起動し直します。 そのため、メイン側で書き換えたグローバル変数が子プロセスから見えるかどうかが変わります。

# start_method.py（Python 3.14 / Linux）
import multiprocessing as mp
from concurrent.futures import ProcessPoolExecutor

CONFIG = {"mode": "初期値"}

def read_config(_):
    return CONFIG["mode"]

if __name__ == "__main__":
    CONFIG["mode"] = "mainで書き換えた値"
    for method in ["fork", "forkserver", "spawn"]:
        ctx = mp.get_context(method)
        with ProcessPoolExecutor(1, mp_context=ctx) as ex:
            print(f"{method:<10} → 子プロセスから見た値: {ex.submit(read_config, 0).result()}")

実行結果：

fork       → 子プロセスから見た値: mainで書き換えた値
forkserver → 子プロセスから見た値: 初期値
spawn      → 子プロセスから見た値: 初期値

設定値は引数か initializer で明示的に渡すと、どの方式でも同じ結果になります。

ワーカーが異常終了したとき

ワーカープロセスが途中で終了すると、プール全体が使えなくなります。

concurrent.futures.process.BrokenProcessPool : A process in the process pool was terminated abruptly while the future was running or pending.

上は、ワーカー内で os._exit(1) を呼んで再現した結果です。この例外を受けたらプールを作り直し、1件あたりのデータ量やmax_workersを見直します。

GILとfree-threading版

GIL（Global Interpreter Lock）は、同時に1スレッドだけがPythonのコードを実行できるようにする仕組みです。 通常版でスレッドがCPU計算で速くならないのはこのためです。3.14の free-threading版（python3.14t）ではGILを無効にでき、sys._is_gil_enabled() で確認できます。

uv python install 3.14t
python3.14t -c "import sys; print(sys._is_gil_enabled())"
# False

C拡張ライブラリがfree-threadingに未対応の場合はGILが自動で有効に戻ることがあるので、起動後に sys._is_gil_enabled() で確かめると安心です。

InterpreterPoolExecutorはグローバル変数を共有しない

# interp.py（Python 3.14+）
from concurrent.futures import InterpreterPoolExecutor
COUNTER = []
def add(x):
    COUNTER.append(x)
    return len(COUNTER)
if __name__ == "__main__":
    with InterpreterPoolExecutor(2) as ex:
        print("ワーカー側の戻り値:", list(ex.map(add, range(4))))
    print("メイン側の COUNTER:", COUNTER)

実行結果：

ワーカー側の戻り値: [1, 1, 2, 3]
メイン側の COUNTER: []

各ワーカーは独立したインタプリタ（モジュールもグローバル変数も別）なので、結果は戻り値で受け取ります。考え方はProcessPoolExecutorと同じで、プロセスを増やさずに済むのが特徴です。

asyncioとの使い分け（要点のみ）

asyncioは、1つのスレッドの中で「待っている間に別の仕事を進める」仕組みです。 実測のとおりI/O待ちはThreadPoolと同じ速さで、同時数が数千を超えるときに軽く動きます。

TaskGroupやtimeoutなどasyncio本体の書き方は、別記事「Python asyncioの使い方」で詳しく扱っています。

multiprocessing.Poolを使う場面

imap_unordered や starmap を使いたいときは multiprocessing.Pool が便利です。 速度はProcessPoolExecutorとほぼ同じでした（pool.starmap(area, [(2, 3), (4, 5)]) → [6, 20]）。

よくある質問

Q. ThreadPoolExecutorとProcessPoolExecutor、迷ったらどちらですか？ A. まずThreadPoolExecutorで試し、CPU使用率が1コアに張り付いて速くならなければProcessPoolExecutorに切り替えます。 クラス名を差し替えるだけで済むのが concurrent.futures の利点です。

Q. 並列にしたのに遅くなりました。なぜですか？ A. 1件の仕事が軽すぎると、プロセスの起動やpickleの受け渡しの時間が上回ります。map(..., chunksize=100) で複数件をまとめて送る、プールを使い回す、の2点で改善することが多いです。

Q. Jupyter NotebookでProcessPoolExecutorが止まります。 A. spawn・forkserver方式では、ノートブック内で定義した関数を子プロセスが読み込めないことがあります。関数を別の .py ファイルに書いてimportすると動きます。I/O待ちならThreadPoolExecutorを使うと手軽です。

Q. free-threading版の3.14tはすぐ使ったほうがよいですか？ A. 純Pythonの計算をスレッドで並べたい場合に試す価値があります。利用するC拡張ライブラリの対応状況を確認し、sys._is_gil_enabled() がFalseになっているかを見ながら段階的に導入すると安心です。

まとめ

• I/O待ちはThreadPoolExecutor（大量ならasyncio）、CPU計算はProcessPoolExecutorが基本の選び方です。
• 実測では、I/O待ち20件が4.0秒→0.2秒、CPU計算は2コアで約2倍になりました。
• map は入力順・as_completed は終わった順。1件の失敗で止めたくないときは submit＋as_completed。
• result(timeout=) は待つのをやめるだけ。止めたい処理には呼び出し先のタイムアウトを渡します。
• ProcessPoolはmainガード必須・pickleできる値だけ。3.14からLinuxの既定がforkserverになりました。
• 3.14では InterpreterPoolExecutor、map(buffersize=)、free-threading版が選択肢に加わりました。

株式会社RJCでは、Pythonの並列処理の選び方を仲間と学び合っています。こうした知識を共有しながら、日々技術を磨いています。

株式会社RJC ― SI事業・SES事業・AI駆動開発。RJCは一緒に成長を楽しめる会社です。