Pythonのasyncioで同時実行数を制限する(Semaphore、TaskGroup、gather)

Python
スポンサーリンク

Pythonのasyncioで、APIの呼び出しなどを並べて実行するときに、同時に動く数を制限する方法です。asyncio.Semaphore で同時に5つまでにし、asyncio.gather と asyncio.TaskGroup の両方と組み合わせました。経過時間だけでなく、「実行中の数」とその最大値を記録して、制限が効いているかを確かめています。後半では、1つのタスクが失敗したときに、gatherとTaskGroupで残りのタスクがどうなるかを比べました。

結論

Semaphore は1つだけ作ってすべてのタスクで共有し、各タスクの中の、制限したい処理を async with sem: で囲みます。gather にも TaskGroup にも、同時に動く数を制限する機能はありません。

async def job(i, sem):
    async with sem:  # 空きがなければ、ここで待つ
        await asyncio.sleep(0.2)  # APIの呼び出しなど
        return i


async def main():
    sem = asyncio.Semaphore(5)  # 同時に5つまで
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(job(i, sem)) for i in range(20)]

上は、記録用の処理を除いて形を示したものです。実際に動かしたコードと出力は、後の章に載せています。0.2秒かかる処理を20個実行した結果です。

書き方実行中の数の最大経過時間
gather だけ200.20秒
gather + Semaphore(5)50.80秒
TaskGroup だけ200.20秒
TaskGroup + Semaphore(5)50.80秒
create_task を呼ぶ側で async with sem20(制限されない)0.20秒
タスクごとに Semaphore(5) を作る20(制限されない)0.20秒

20個のうち1つが0.1秒後に失敗したときは、次のようになりました(Semaphore(5) で制限した状態)。

書き方呼び出し側に届くもの残りの19個
gather(既定)0.10秒後に ValueError取り消されずに動き続け、待てば19個とも完了した
gather(return_exceptions=True)1.20秒後に結果のリスト(例外が1つ入る)19個とも完了した
TaskGroup0.10秒後に ExceptionGroup19個とも取り消された(実行中の5個と、空きを待っていた14個)

gather(既定)で例外を受け取ったあと、残りのタスクを待たずに main() を抜けると、残りのタスクは取り消されました。ただし、取り消したのは gather ではなく、asyncio.run() の終わりの処理です。

動作確認した環境

  • Python 3.15.2(Dockerイメージ python:3.15.2-slim、digest sha256:65a65fc7b30b06cb6df4930fd4650697cb23f4c2698ab81fc87e4b031bc36928)。標準ライブラリだけを使っています
  • Docker 29.7.2(Windows 11 の Docker Desktop、Linuxコンテナの設定、x86_64)
  • 確認日: 2026年11月9日

この記事は、MacとWindowsの両方の読者を対象にしています。Pythonのコードは、Dockerのコンテナの中で実行しました。この記事のコマンドは、Claude(AI)が実行して確認しました。本文の出力は、Windows 11 のGit Bashから上から順に実行した結果です。PowerShellで実行した結果は、最後の章にまとめました。Macでは確認していません。

経過時間は、実行する環境や負荷で少し変わります。TaskGroup と except* は Python 3.11 で追加されたものなので、それより前のPythonでは後半のコードは動きません(公式ドキュメント。3.11〜3.14 では確認していません)。

準備

空のフォルダに、この記事の .py のファイルを作り、そのフォルダで次のように実行します。フォルダをコンテナの /work にマウントして、python で実行するだけです。WindowsのGit Bashを使う場合は、先に export MSYS_NO_PATHCONV=1 を実行してください(理由はこの章の後半に書きました)。Macのターミナルと、WindowsのGit Bashでは次のとおりです。

docker run --rm -v "$PWD:/work" -w /work python:3.15.2-slim python no_limit.py

WindowsのPowerShellでは、$PWD を ${PWD} にします。

docker run --rm -v "${PWD}:/work" -w /work python:3.15.2-slim python no_limit.py

Git Bashで MSYS_NO_PATHCONV=1 を設定しないと、Git Bashが /work をWindowsのパスに変換してしまい、次のエラーになりました。

$ docker run --rm -v "$PWD:/work" -w /work python:3.15.2-slim python semaphore_gather.py
docker: Error response from daemon: the working directory 'C:/Program Files/Git/work' is invalid, it needs to be an absolute path

Run 'docker run --help' for more information

Macで動かす場合は、Docker Desktop for Mac を入れて、ターミナルで同じコマンドを実行する想定です(確認していません)。Pythonを直接入れている場合は、python3 no_limit.py のように実行しても同じコードが動くはずですが、こちらも確認していません。

同時に動いている数を数えるために、記録用の小さなクラスを meter.py に作ります。with meter: で囲んだ区間に入ると「実行中の数」を1増やして最大値を更新し、出ると1減らします。開始した時刻も0.1秒単位で数えます。asyncioのタスクは1つのスレッドで動き、ほかのタスクに切り替わるのは await のところだけなので、ロックなしで数えられます。

import time
from collections import Counter


class Meter:
    """同時に実行中の数と、その最大値を記録する"""

    def __init__(self):
        self.t0 = time.perf_counter()
        self.running = 0
        self.max_running = 0
        self.starts = Counter()  # 開始した時刻(0.1秒単位)ごとの数

    def __enter__(self):
        self.running += 1
        self.max_running = max(self.max_running, self.running)
        self.starts[round(time.perf_counter() - self.t0, 1)] += 1

    def __exit__(self, *exc):
        self.running -= 1

    def elapsed(self):
        return f"{time.perf_counter() - self.t0:.2f}s"

    def report(self):
        print(f"elapsed: {self.elapsed()}")
        print(f"max running: {self.max_running}")
        print(f"start times: {dict(sorted(self.starts.items()))}")

制限しないと、20個が同時に動く

まず、制限なしで gather に20個のコルーチンを渡します。asyncio.sleep(0.2) を、APIの呼び出しなどの待ち時間の代わりにしています。no_limit.py です。

import asyncio

from meter import Meter


async def job(i, meter):
    with meter:
        await asyncio.sleep(0.2)  # APIの呼び出しなどの代わり
        return i


async def main():
    meter = Meter()
    results = await asyncio.gather(*(job(i, meter) for i in range(20)))
    print(f"results: {len(results)}")
    meter.report()


asyncio.run(main())
$ docker run --rm -v "$PWD:/work" -w /work python:3.15.2-slim python no_limit.py
results: 20
elapsed: 0.20s
max running: 20
start times: {0.0: 20}

20個とも0秒の時点で始まり、実行中の数の最大は20でした。相手がAPIなら、20個のリクエストが同時に届きます。

Semaphoreで同時に5つまでにする

asyncio.Semaphore(5) を1つ作り、各タスクの中で async with sem: で囲みます。semaphore_gather.py です。

import asyncio

from meter import Meter


async def job(i, sem, meter):
    async with sem:  # 空きがなければ、ここで待つ
        with meter:
            await asyncio.sleep(0.2)
            return i


async def main():
    sem = asyncio.Semaphore(5)  # 同時に5つまで
    meter = Meter()
    results = await asyncio.gather(*(job(i, sem, meter) for i in range(20)))
    print(f"results: {len(results)}")
    meter.report()


asyncio.run(main())
$ docker run --rm -v "$PWD:/work" -w /work python:3.15.2-slim python semaphore_gather.py
results: 20
elapsed: 0.80s
max running: 5
start times: {0.0: 5, 0.2: 5, 0.4: 5, 0.6: 5}

実行中の数の最大は5で、0秒、0.2秒、0.4秒、0.6秒に5個ずつ始まりました。全体では、5個ずつ4回に分かれるので0.80秒です。

公式ドキュメントによると、Semaphoreは内部のカウンターを持ち、acquire() で1減らし、release() で1増やします。カウンターが0のときの acquire() は、ほかのタスクが release() するまで待ちます。async with sem: は、await sem.acquire() と、finally の中の sem.release() と同じ意味です。例外で抜けたときや取り消されたときも、枠は返されます。

TaskGroupと組み合わせる

TaskGroup は、create_task() で作ったタスクを、async with を抜けるときにすべて待つ仕組みです。同時に動く数を制限する役割はありません。TaskGroup だけの場合と、Semaphoreと組み合わせた場合を比べました。taskgroup.py です。

import asyncio

from meter import Meter


async def job(i, meter):
    with meter:
        await asyncio.sleep(0.2)
        return i


async def limited_job(i, sem, meter):
    async with sem:
        return await job(i, meter)


async def main():
    print("-- TaskGroup だけ")
    meter = Meter()
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(job(i, meter)) for i in range(20)]
    print(f"results: {len([t.result() for t in tasks])}")
    meter.report()

    print("-- TaskGroup + Semaphore(5)")
    sem = asyncio.Semaphore(5)
    meter = Meter()
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(limited_job(i, sem, meter)) for i in range(20)]
    print(f"results: {len([t.result() for t in tasks])}")
    meter.report()


asyncio.run(main())
$ docker run --rm -v "$PWD:/work" -w /work python:3.15.2-slim python taskgroup.py
-- TaskGroup だけ
results: 20
elapsed: 0.20s
max running: 20
start times: {0.0: 20}
-- TaskGroup + Semaphore(5)
results: 20
elapsed: 0.80s
max running: 5
start times: {0.0: 5, 0.2: 5, 0.4: 5, 0.6: 5}

TaskGroup だけでは、gather だけのときと同じく20個が同時に動きました。Semaphoreを組み合わせると、最大は5になりました。Semaphoreを使う場所は、gather でも TaskGroup でも同じで、タスクの中です。

制限が効かない書き方

Semaphoreを使っていても、置く場所を間違えると制限されません。2つの書き方を試しました。wrong_place.py です。

import asyncio

from meter import Meter


async def job(i, meter):
    with meter:
        await asyncio.sleep(0.2)
        return i


async def job_with_own_semaphore(i, meter):
    sem = asyncio.Semaphore(5)  # タスクごとに別のSemaphoreになる
    async with sem:
        return await job(i, meter)


async def main():
    print("-- create_task を呼ぶ側で async with sem")
    sem = asyncio.Semaphore(5)
    meter = Meter()
    async with asyncio.TaskGroup() as tg:
        for i in range(20):
            async with sem:
                tg.create_task(job(i, meter))
    meter.report()

    print("-- タスクの中で Semaphore を作る")
    meter = Meter()
    async with asyncio.TaskGroup() as tg:
        for i in range(20):
            tg.create_task(job_with_own_semaphore(i, meter))
    meter.report()


asyncio.run(main())
$ docker run --rm -v "$PWD:/work" -w /work python:3.15.2-slim python wrong_place.py
-- create_task を呼ぶ側で async with sem
elapsed: 0.20s
max running: 20
start times: {0.0: 20}
-- タスクの中で Semaphore を作る
elapsed: 0.20s
max running: 20
start times: {0.0: 20}

どちらも、実行中の数の最大は20で、制限されませんでした。

  • create_task() を呼ぶ側で async with sem: を使う: create_task() はタスクを登録するだけで、タスクの終わりを待たずに戻ります。枠はすぐに返されるので、タスクの実行は制限されません。
  • タスクの中で Semaphore を作る: タスクごとに別のカウンターになるので、どのタスクもすぐに枠を取れます。Semaphoreは1つだけ作り、引数などで全部のタスクに渡します。

1つが失敗したときの、gatherとTaskGroupの違い

20個のうち、番号2のタスクだけが0.1秒後に ValueError を出すようにします。ほかのタスクは0.3秒かかります。各タスクの状態(waiting は空きを待っている、running は実行中、done は完了、failed は失敗)を辞書に記録し、取り消されたときは、その時点の状態を cancelled while ... として残します。この処理を failing.py にまとめ、3つの確認で使います。

import asyncio
from collections import Counter


async def job(i, sem, meter, status):
    status[i] = "waiting"  # Semaphore の空きを待っている
    try:
        async with sem:
            with meter:
                status[i] = "running"
                if i == 2:
                    await asyncio.sleep(0.1)
                    status[i] = "failed"
                    raise ValueError(f"job {i} failed")
                await asyncio.sleep(0.3)
                status[i] = "done"
                return i
    except asyncio.CancelledError:
        status[i] = f"cancelled while {status[i]}"
        raise


def summary(status):
    return dict(sorted(Counter(status.values()).items()))

gatherは、例外を返しても残りのタスクを取り消さない

既定の gather と、return_exceptions=True をつけた gather を試します。既定のほうは、例外を受け取ったあとに残りのタスクの終わりを待つため、タスクを create_task() で作って、リストに持っておきます。failure_gather.py です。

import asyncio

from failing import job, summary
from meter import Meter


async def gather_default():
    print("-- gather(既定)")
    sem, meter, status = asyncio.Semaphore(5), Meter(), {}
    tasks = [asyncio.create_task(job(i, sem, meter, status)) for i in range(20)]
    try:
        await asyncio.gather(*tasks)
    except ValueError as e:
        print(f"{meter.elapsed()} gather raised {e!r}")
        print(f"  status: {summary(status)}")
    await asyncio.wait(tasks)  # 残りのタスクが終わるのを待つ
    print(f"{meter.elapsed()} after asyncio.wait(tasks)")
    print(f"  status: {summary(status)}")


async def gather_return_exceptions():
    print("-- gather(return_exceptions=True)")
    sem, meter, status = asyncio.Semaphore(5), Meter(), {}
    results = await asyncio.gather(
        *(job(i, sem, meter, status) for i in range(20)),
        return_exceptions=True,
    )
    print(f"{meter.elapsed()} gather returned")
    print(f"  errors: {[r for r in results if isinstance(r, Exception)]}")
    print(f"  status: {summary(status)}")


async def main():
    await gather_default()
    await gather_return_exceptions()


asyncio.run(main())
$ docker run --rm -v "$PWD:/work" -w /work python:3.15.2-slim python failure_gather.py
-- gather(既定)
0.10s gather raised ValueError('job 2 failed')
  status: {'failed': 1, 'running': 5, 'waiting': 14}
1.20s after asyncio.wait(tasks)
  status: {'done': 19, 'failed': 1}
-- gather(return_exceptions=True)
1.20s gather returned
  errors: [ValueError('job 2 failed')]
  status: {'done': 19, 'failed': 1}

既定の gather は、0.10秒後に失敗した時点で、ValueError を呼び出し側に送出しました。そのとき、ほかのタスクは実行中が5個、空きを待っているのが14個で、取り消されたものはありません(番号2が終わって空いた枠で、次のタスクがもう動き始めていました)。そのあと asyncio.wait(tasks) で待つと、1.20秒の時点で19個とも完了していました。公式ドキュメントにも、return_exceptions が False(既定)のとき、最初の例外がすぐに伝わり、ほかのタスクは取り消されずに動き続けると書かれています。

return_exceptions=True をつけると、例外は送出されず、結果のリストに入って返りました。全部が終わるまで待つので、返ったのは1.20秒の時点です。

残りを待たずにmain()を抜けると、asyncio.run()が取り消す

既定の gather で例外を受け取ったあと、残りのタスクを待たずに main() を抜けるとどうなるかを試しました。asyncio.run() から戻ったあとに、各タスクの状態を表示します。failure_run_end.py です。

import asyncio

from failing import job, summary
from meter import Meter

status = {}


async def main():
    sem, meter = asyncio.Semaphore(5), Meter()
    tasks = [asyncio.create_task(job(i, sem, meter, status)) for i in range(20)]
    try:
        await asyncio.gather(*tasks)
    except ValueError as e:
        print(f"{meter.elapsed()} gather raised {e!r}")
    print(f"  status at the end of main(): {summary(status)}")
    # 残りのタスクを待たずに main() を抜ける


asyncio.run(main())
print(f"status after asyncio.run(): {summary(status)}")
$ docker run --rm -v "$PWD:/work" -w /work python:3.15.2-slim python failure_run_end.py
0.10s gather raised ValueError('job 2 failed')
  status at the end of main(): {'failed': 1, 'running': 5, 'waiting': 14}
status after asyncio.run(): {'cancelled while running': 5, 'cancelled while waiting': 14, 'failed': 1}

main() の最後の時点では、取り消されたタスクはありませんでした。asyncio.run() から戻ったあとには、残りの19個が取り消されていました。asyncio.run() は、main() が終わったあとに、残っているタスクをすべて取り消してから終わるためです。この動きは公式ドキュメントの asyncio.run() の説明には書かれていませんが、Python 3.15.2 の asyncio/runners.py では、終わりの処理(Runner.close())で _cancel_all_tasks() を呼び、残りのタスクに cancel() しています。

このように、「gatherで例外が出たら、残りのタスクも止まった」ように見えても、止めたのは gather ではありません。main() がまだ続く場合や、Webサーバーのようにイベントループが動き続ける場合は、上の asyncio.wait(tasks) の例のとおり、残りのタスクは動き続けます。

TaskGroupは、1つが失敗すると残りを取り消す

同じ処理を TaskGroup で実行します。TaskGroup は例外を ExceptionGroup にまとめて送出するので、except* で受け取ります。failure_taskgroup.py です。

import asyncio

from failing import job, summary
from meter import Meter


async def main():
    sem, meter, status = asyncio.Semaphore(5), Meter(), {}
    try:
        async with asyncio.TaskGroup() as tg:
            for i in range(20):
                tg.create_task(job(i, sem, meter, status))
    except* ValueError as eg:
        print(f"{meter.elapsed()} TaskGroup raised {eg!r}")
    print(f"  status: {summary(status)}")


asyncio.run(main())
$ docker run --rm -v "$PWD:/work" -w /work python:3.15.2-slim python failure_taskgroup.py
0.10s TaskGroup raised ExceptionGroup('unhandled errors in a TaskGroup', [ValueError('job 2 failed')])
  status: {'cancelled while running': 5, 'cancelled while waiting': 14, 'failed': 1}

0.10秒後に、残りの19個が取り消されました。実行中だった5個に加えて、Semaphoreの空きを待っていた14個も、始まる前に取り消されています。公式ドキュメントでも、グループのタスクが CancelledError 以外の例外で失敗すると、残りのタスクは取り消され、すべてのタスクが終わったあとに、例外がまとめて送出されると説明されています。今回の ValueError のような例外は ExceptionGroup にまとめられます。ただし、KeyboardInterrupt と SystemExit は特別な扱いで、残りのタスクを取り消して待ったあと、まとめずにそのまま送出されます(この2つは確かめていません)。

1つでも失敗したら残りをやめたいときは TaskGroup、失敗したものがあっても全部の結果がほしいときは gather(return_exceptions=True) が合います。既定の gather を使う場合は、例外を受け取ったあとに残りのタスクを待つか取り消すかを、自分で書く必要があります。

Python 3.15での変更

Python 3.15 の「What’s New」の「Improved modules」にある asyncio の項目は、TaskGroup.cancel() の追加の1件でした。目的を達したときなどに、グループを途中で終わらせるためのメソッドです。この記事のコードでは使っておらず、動作も確かめていません。この記事で使った Semaphore、gather、TaskGroup の失敗時の動きについて、3.15 での変更は書かれていませんでした。非推奨の一覧にある asyncio の項目(イベントループのポリシーの仕組みなど)も、この記事のコードには関係しません(2026年11月9日に確認)。

WindowsのPowerShellで実行する場合

Windows 11 のPowerShell 7.6 から、semaphore_gather.py と failure_taskgroup.py を実行しました。ファイルの改行はCRLF(Windowsのエディタで保存したとき)にして試しています。出力は本文と同じでした。

PS> docker run --rm -v "${PWD}:/work" -w /work python:3.15.2-slim python semaphore_gather.py
results: 20
elapsed: 0.80s
max running: 5
start times: {0.0: 5, 0.2: 5, 0.4: 5, 0.6: 5}
PS> docker run --rm -v "${PWD}:/work" -w /work python:3.15.2-slim python failure_taskgroup.py
0.10s TaskGroup raised ExceptionGroup('unhandled errors in a TaskGroup', [ValueError('job 2 failed')])
  status: {'cancelled while running': 5, 'cancelled while waiting': 14, 'failed': 1}

Macのターミナルでも、準備の章の $PWD のコマンドで動く想定ですが、確認はしていません。

参考

スポンサーリンク
Python
スポンサーリンク
シェアする
hayoをフォローする
暇なのでいろいろと…
スポンサーリンク

コメント

タイトルとURLをコピーしました