Python concurrent.futuresの使い方｜ThreadPoolExecutorでfor文を並列処理にする方法とProcessPoolExecutorとの違い
Pythonで並列処理を手早く書くなら、標準ライブラリのconcurrent.futuresを使います。forループの処理をwith ThreadPoolExecutor() as ex: ex.map(関数, データ)に置き換えるだけで同時に実行でき、通信やファイル待ちが多い処理はThreadPoolExecutor、計算が重い処理はProcessPoolExecutorを選びます。
この記事では、forループを並列化する最小の書き方から、submit()とas_completed()で終わった順に結果を受け取る方法、例外とタイムアウトの扱い、スレッドとプロセスの選び方、max_workersの既定値、ProcessPoolExecutorで出るRuntimeErrorやBrokenProcessPoolの直し方までを、Python 3.12で実行した結果つきで説明します。Python 3.13〜3.15での変更点も公式ドキュメントで確認して載せています。
forループを並列処理にする最小の書き方
0.5秒かかる処理を5回呼ぶ例です。forで順番に呼ぶ場合と、ThreadPoolExecutorのmap()で同時に呼ぶ場合を比べます。
import time
from concurrent.futures import ThreadPoolExecutor
def task(n):
time.sleep( 0.5) # 通信やファイル待ちの代わり
return n * 10
start = time.perf_counter()
results = [task(n) for n in range(5)]
print("forで順番に:", results, f"{time.perf_counter() - start:.2f}秒")
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=5) as ex:
results = list(ex.map(task, range(5)))
print("map で並行:", results, f"{time.perf_counter() - start:.2f}秒")実行結果：
forで順番に: [0, 10, 20, 30, 40] 2.52秒
map で並行: [0, 10, 20, 30, 40] 0.51秒5つの待ち時間が重なったので、約2.5秒が約0.5秒になりました。map()の結果は渡した順番どおりに並びます。withを抜けるときに全タスクの終了を待つので、shutdown()を自分で呼ぶ必要もありません。
書き方の手順は次の3つです。
- 1件分の処理を関数にする（上の
task） with ThreadPoolExecutor() as ex:で実行役（Executor）を作るex.map(関数, データのリスト)で一斉に投げ、list()で結果を受け取る
ThreadPoolExecutorとProcessPoolExecutorの選び方
concurrent.futuresには実行役が主に2つあり、クラス名を入れ替えるだけで切り替えられます。
| クラス | 同時に動かす単位 | 向いている処理 | 例 |
|---|---|---|---|
ThreadPoolExecutor |
スレッド | 待ち時間が長い処理（I/Oバウンド） | HTTPリクエスト、ファイルの読み書き、DB問い合わせ |
ProcessPoolExecutor |
プロセス | 計算が重い処理（CPUバウンド） | 数値計算、画像の変換、ハッシュ計算 |
計算が重い処理で、両方とforを比べました（CPU 8コアのMac、Python 3.12）。
import time, os
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
def heavy(n):
return sum(i * i for i in range(n))
if __name__ == "__main__":
N = 5_000_000
for name, cls in [("Thread", ThreadPoolExecutor), ("Process", ProcessPoolExecutor)]:
start = time.perf_counter()
with cls(max_workers=4) as ex:
list(ex.map(heavy, [N] * 4))
print(f"{name:7}: {time.perf_counter() - start:.2f}秒")
start = time.perf_counter()
[heavy(N) for _ in range(4)]
print(f"for : {time.perf_counter() - start:.2f}秒")
print("CPU数:", os.cpu_count())実行結果：
Thread : 0.77秒
Process: 0.26秒
for : 0.83秒
CPU数: 8計算処理では、スレッドにしてもforとほとんど変わりませんでした。通常のPython（CPython）にはGIL（グローバルインタプリタロック）があり、Pythonのコードを同時に実行できるスレッドは1つだけだからです。プロセスはそれぞれ別のPythonが動くので、CPUのコアを同時に使えます。
公式ドキュメントでは、Python 3.13からGILを無効にできるフリースレッド版のビルドがあるものの、既定では使えないと説明されています（threading、2026年10月18日確認）。GILの仕組みと現状は「PythonのGILとは？」で詳しく扱っています。
逆に、待ち時間が中心の処理をProcessPoolExecutorにすると、プロセスの起動とデータの受け渡しの分だけ遅くなりがちです。まずはThreadPoolExecutorで書き、計算が重いと分かった処理だけをプロセスに切り替えるのが無難です。
終わった順に結果を受け取る：submit()とas_completed()
map()は順番どおりに結果を返すため、最初のタスクが遅いと後ろの結果も待たされます。終わったものから処理したいときは、submit()で1件ずつ投げてFuture（将来の結果の引換券）を受け取り、as_completed()で回します。
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
def task(n):
time.sleep( 0.1 * (5 - n)) # 後ろのタスクほど早く終わる
if n == 2:
raise ValueError(f"n={n}で失敗")
return n * 10
with ThreadPoolExecutor(max_workers=5) as ex:
futures = {ex.submit(task, n): n for n in range(5)}
for fut in as_completed(futures):
n = futures[fut]
try:
print(n, "→", fut.result())
except ValueError as e:
print(n, "→ 例外:", e)実行結果：
4 → 40
3 → 30
2 → 例外: n=2で失敗
1 → 10
0 → 0早く終わった順に表示され、1件が失敗してもほかの結果は受け取れています。Futureと元の引数を辞書で対応づけておくと、どの入力が失敗したかが分かります。submit()の引数の渡し方は「ThreadPoolExecutor submit()の使い方」、as_completed()は「as_completed()とは？」で詳しく説明しています。
| 方法 | 結果の順番 | 例外が起きたとき |
|---|---|---|
ex.map(f, data) |
渡した順 | その結果を取り出した時点で送出され、以降は受け取れない |
ex.submit(f, x)＋as_completed() |
終わった順 | fut.result()ごとに捕まえられる |
ex.submit(f, x)＋wait() |
まとめて（完了・未完了の集合） | 各result()で捕まえる |
map()で例外が起きると途中で止まる
from concurrent.futures import ThreadPoolExecutor
def task(n):
if n == 2:
raise ValueError("n=2で失敗")
return n
with ThreadPoolExecutor() as ex:
try:
for r in ex.map(task, range(5)):
print(r)
except ValueError as e:
print("例外:", e)実行結果：
0
1
例外: n=2で失敗n=3とn=4は実行自体は終わっていますが、map()の結果としては受け取れません。1件の失敗で全体を止めたくないなら、関数の中で例外を捕まえて戻り値で返すか、submit()とas_completed()を使います。
タイムアウトと待ち方：wait()とresult(timeout=)
import time
from concurrent.futures import ThreadPoolExecutor, wait, TimeoutError
def task(n):
time.sleep( n)
return n
with ThreadPoolExecutor() as ex:
futs = [ex.submit(task, n) for n in (0.2, 1.0, 3.0)]
done, not_done = wait(futs, timeout=1.5)
print("1.5秒で終わった:", sorted(f.result() for f in done), "未完了:", len(not_done))
try:
futs[2].result(timeout=0.5)
except TimeoutError:
print("result(timeout=0.5) は TimeoutError")実行結果：
1.5秒で終わった: [0.2, 1.0] 未完了: 1
result(timeout=0.5) は TimeoutErrorwait()は「終わった集合」と「まだの集合」を返します。return_when=FIRST_COMPLETEDを付けると、どれか1つが終わった時点で戻ります。注意したいのは、タイムアウトしても実行中のタスクが止まるわけではない点です。上の例でも3秒のタスクは動き続け、withを抜けるときにその終了を待ちます。まだ始まっていないタスクはex.shutdown(cancel_futures=True)（Python 3.9で追加）で取り消せます。
Python 3.11からはconcurrent.futures.TimeoutErrorは組み込みのTimeoutErrorの別名になっています（公式ドキュメントで確認）。
max_workersの既定値と決め方
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
if __name__ == "__main__":
print("Thread既定:", ThreadPoolExecutor()._max_workers)
print("Process既定:", ProcessPoolExecutor()._max_workers)実行結果（CPU 8コア）：
Thread既定: 12
Process既定: 8公式ドキュメントによると、ThreadPoolExecutorの既定値はmin(32, CPU数 + 4)、ProcessPoolExecutorはCPU数です。Python 3.13からは、どちらもos.cpu_count()ではなく、そのプロセスが使えるCPU数（os.process_cpu_count()）を基準にするよう変わりました（concurrent.futures、2026年10月18日確認）。
- ProcessPoolExecutor：CPU数より増やしても速くなりにくい。既定のままでよいことが多い
- ThreadPoolExecutor：待ちが中心ならCPU数より多くてよい。ただし外部のWebサイトやAPIに投げるときは、相手の負荷と利用規約を考えて少なめにする
ProcessPoolExecutorで起きるエラーと3.14の変更
RuntimeError: An attempt has been made to start a new process…
次のように、if __name__ == "__main__":を書かずにProcessPoolExecutorを使うとエラーになります（macOSで実行）。
from concurrent.futures import ProcessPoolExecutor
def f(n):
return n * 2
with ProcessPoolExecutor(max_workers=2) as ex:
print(list(ex.map(f, [1, 2, 3])))実行結果（抜粋）：
RuntimeError:
An attempt has been made to start a new process before the
current process has finished its bootstrapping phase.子プロセスがスクリプトを読み込み直したときに、またプロセスを作ろうとするのが原因です。実行する部分をif __name__ == "__main__":の中に入れれば直ります。この書き方の意味は「if __name__ == “__main__”: とは？」で説明しています。Jupyter Notebookでは、関数を別の.pyファイルに書いてimportする必要がある場合があります。
公式ドキュメントによると、Python 3.14からLinuxなどPOSIX環境でもプロセスの既定の起動方法がforkから変わりました。これまでLinuxでは__main__ガードなしで動いていたコードも、3.14以降は同じ注意が必要になると考えておくのが安全です。forkが必要な場合はmp_context=multiprocessing.get_context("fork")を明示します。
BrokenProcessPool
import os
from concurrent.futures import ProcessPoolExecutor
from concurrent.futures.process import BrokenProcessPool
def crash(n):
os._exit(1) # ワーカーが異常終了したことにする
if __name__ == "__main__":
with ProcessPoolExecutor(max_workers=2) as ex:
try:
print(ex.submit(crash, 1).result())
except BrokenProcessPool as e:
print("BrokenProcessPool:", e)実行結果：
BrokenProcessPool: A process in the process pool was terminated abruptly while the future was running or pending.ワーカープロセスが途中で落ちると、そのプール全体が使えなくなります。メモリ不足でOSに強制終了された、C拡張がクラッシュした、といったときに出ます。1件あたりのデータ量を減らす、max_workersを減らす、などで原因を切り分けます。chunksizeやinitializerなどProcessPoolExecutor固有の設定は「ProcessPoolExecutorの使い方」で扱っています。
Python 3.14〜3.15で増えたもの
InterpreterPoolExecutor（3.14）：スレッドごとに別のインタプリタで実行する第3の実行役map()のbuffersize引数（3.14）：まだ受け取っていない結果の数を制限し、巨大なデータを少しずつ流せるProcessPoolExecutorのterminate_workers()・kill_workers()（3.14）：ワーカーをすぐに止める- 3.15で、
max_tasks_per_child使用時に止まってしまう不具合（gh-115634）を修正
（いずれも公式ドキュメントで2026年10月18日に確認。この記事のコードは3.12で実行しています）
threading・multiprocessing・asyncioとの使い分け
| やりたいこと | おすすめ |
|---|---|
| 決まった数のタスクを並列で実行して結果を集める | concurrent.futures |
| 常駐するスレッド、細かい同期や停止の制御 | threading |
| プロセス間でキューや共有メモリを使う | multiprocessing |
| 数千件規模の通信を1スレッドでさばく | asyncio |
| 機械学習・データ処理で手軽に並列化 | joblib.Parallel |
よくある質問
Q. Pythonでfor文を並列処理にするには？
ループの中身を関数にして、with ThreadPoolExecutor() as ex: results = list(ex.map(関数, データ))と書き換えます。計算が重い処理ならThreadPoolExecutorをProcessPoolExecutorに替え、if __name__ == “__main__”:の中で実行します。
Q. ThreadPoolExecutorとProcessPoolExecutorはどちらを使えばよいですか？
通信やファイルの読み書きなど待ち時間が長い処理はThreadPoolExecutor、数値計算や画像処理などCPUを使い続ける処理はProcessPoolExecutorです。GILのため、計算処理はスレッドにしても速くなりません。
Q. map()とsubmit()の違いは何ですか？
map()はデータをまとめて投げ、渡した順に結果を返します。submit()は1件ずつ投げてFutureを受け取る方法で、as_completed()と組み合わせると終わった順に結果を受け取り、例外も1件ずつ捕まえられます。
Q. ProcessPoolExecutorでRuntimeErrorになります
実行部分をif __name__ == “__main__”:の中に書いていないのが原因です。子プロセスがスクリプトを読み込み直したときに、またプロセスを作ろうとしてエラーになります。
まとめ
- forループの並列化は、関数にしてex.map()に渡すだけ
- 待ちが中心ならThreadPoolExecutor、計算が重いならProcessPoolExecutor
- 終わった順・例外を個別に扱うならsubmit()＋as_completed()
- タイムアウトしても実行中のタスクは止まらない
- ProcessPoolExecutorはif __name__ == “__main__”:の中で。3.14から既定の起動方法が変わった
