Python asyncioの使い方｜TaskGroupとtimeoutを実測で解説

Python の asyncio とは、通信やファイルの「待ち時間」の間に別の処理を進めることで、たくさんの待ちを1つのスレッドでまとめてさばく標準ライブラリです。async def で関数を定義し、待つ場所に await を書き、asyncio.run() で動かします。0.5秒かかる通信を10回行う処理なら、1つずつ待つと約5秒、asyncio でまとめて待つと約0.5秒で終わります。

この記事では、Python 3.11 以降の書き方（TaskGroup、asyncio.timeout、except*）を中心に、そのまま貼って動くコードを集めました。すべて Python 3.14.7 で実行した結果つきで、直列・gather・TaskGroup・スレッドの所要時間を実測した表、gather と TaskGroup で「1つ失敗したとき」の挙動の違い、ローカルサーバーを相手にした httpx・aiohttp の並列HTTP、よくあるエラーの実際のメッセージまで確認できます。

asyncio 早見表

所要時間の実測：直列・gather・TaskGroup・スレッド

まず、asyncio で何が速くなるのかを数字で確認します。「0.5秒待つ処理を10個」を、書き方を変えて実行しました（Python 3.14.7、2コアの環境）。

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

N, WAIT = 10, 0.5      # 0.5秒待つ処理を10個

def blocking_io(i):
    time.sleep(WAIT)   # 同期の待ち（requests などを想定）
    return i

async def async_io(i):
    await asyncio.sleep(WAIT)
    return i

async def bad_async_io(i):
    time.sleep(WAIT)   # async 関数の中で同期の待ちをしてしまう
    return i

def measure(label, func):
    start = time.perf_counter()
    func()
    print(f"{label:<34} {time.perf_counter() - start:5.2f}秒")

async def serial():
    return [await async_io(i) for i in range(N)]

async def gather():
    return await asyncio.gather(*(async_io(i) for i in range(N)))

async def taskgroup():
    async with asyncio.TaskGroup() as tg:
        ts = [tg.create_task(async_io(i)) for i in range(N)]
    return [t.result() for t in ts]

async def gather_bad():
    return await asyncio.gather(*(bad_async_io(i) for i in range(N)))

async def to_thread():
    return await asyncio.gather(*(asyncio.to_thread(blocking_io, i) for i in range(N)))

def threads():
    with ThreadPoolExecutor(max_workers=N) as ex:
        return list(ex.map(blocking_io, range(N)))

measure("直列（1つずつ await）", lambda: asyncio.run(serial()))
measure("asyncio.gather", lambda: asyncio.run(gather()))
measure("asyncio.TaskGroup", lambda: asyncio.run(taskgroup()))
measure("ThreadPoolExecutor（10スレッド）", threads)
measure("asyncio.to_thread + gather", lambda: asyncio.run(to_thread()))
measure("gather だが中で time.sleep", lambda: asyncio.run(gather_bad()))

タスクを1000個に増やすと差がはっきりします。同じ0.5秒の待ちを1000個実行すると、ThreadPoolExecutor(max_workers=100) は 5.02秒、TaskGroup は 0.51秒でした。スレッドは1本ごとにメモリを使うため数を増やしにくい一方、asyncio のタスクは軽く、数千の待ちを1スレッドで同時に扱えます。

基本：async def・await・asyncio.run

import asyncio
import time

async def fetch(name, sec):
    print(f"{name}: 開始")
    await asyncio.sleep(sec)          # 待っている間、他の処理に順番を譲る
    print(f"{name}: 完了")
    return f"{name}の結果"

async def main():
    start = time.perf_counter()
    result = await fetch("A", 1)      # 1つを待つ
    print(result, f"{time.perf_counter() - start:.2f}秒")

asyncio.run(main())

A: 開始
A: 完了
Aの結果 1.00秒

プログラムの入口で asyncio.run(main()) を1回だけ呼び、あとは async def の中で await をつないでいくのが基本の形です。

gather と TaskGroup で並行実行する

import asyncio
import time

async def fetch(name, sec):
    await asyncio.sleep(sec)
    return f"{name}({sec}s)"

async def main():
    start = time.perf_counter()
    results = await asyncio.gather(fetch("A", 1), fetch("B", 2), fetch("C", 1))
    print("gather   :", results, f"{time.perf_counter() - start:.2f}秒")

    start = time.perf_counter()
    async with asyncio.TaskGroup() as tg:                  # 3.11+
        tasks = [tg.create_task(fetch(n, s)) for n, s in [("A", 1), ("B", 2), ("C", 1)]]
    print("TaskGroup:", [t.result() for t in tasks], f"{time.perf_counter() - start:.2f}秒")

asyncio.run(main())

gather   : ['A(1s)', 'B(2s)', 'C(1s)'] 2.00秒
TaskGroup: ['A(1s)', 'B(2s)', 'C(1s)'] 2.00秒

どちらも一番遅い処理（2秒）の時間で終わり、結果は渡した順に並びます。違いが出るのは、途中で失敗したときです。

1つ失敗したときの違い（実行で確認）

import asyncio

async def work(name, sec, fail=False):
    try:
        await asyncio.sleep(sec)
        if fail:
            raise ValueError(f"{name} が失敗")
        print(f"  {name}: 完了")
        return name
    except asyncio.CancelledError:
        print(f"  {name}: キャンセルされた")
        raise

async def use_gather():
    print("① gather（既定）")
    try:
        await asyncio.gather(work("A", 0.1, fail=True), work("B", 0.3))
    except ValueError as e:
        print("  except:", e)
    await asyncio.sleep(0.5)          # 少し待つと…B は裏で動き続けて完了する

async def use_gather_return():
    print("② gather(return_exceptions=True)")
    results = await asyncio.gather(work("A", 0.1, fail=True), work("B", 0.3),
                                   return_exceptions=True)
    print("  results:", results)

async def use_taskgroup():
    print("③ TaskGroup")
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(work("A", 0.1, fail=True))
            tg.create_task(work("B", 0.3))
            tg.create_task(work("C", 0.3, fail=True))
    except* ValueError as eg:                       # 3.11+
        print("  except* ValueError:", [str(e) for e in eg.exceptions])

asyncio.run(use_gather())
asyncio.run(use_gather_return())
asyncio.run(use_taskgroup())

① gather（既定）
  except: A が失敗
  B: 完了
② gather(return_exceptions=True)
  B: 完了
  results: [ValueError('A が失敗'), 'B']
③ TaskGroup
  B: キャンセルされた
  C: キャンセルされた
  except* ValueError: ['A が失敗']

3.11以降で新しく書くなら、まず TaskGroup を選ぶと、失敗したときに裏でタスクが動き続けることがなく、後片付けも確実です。「失敗した分だけ飛ばして続けたい」ときは gather(return_exceptions=True) を使います。except* を書かずに TaskGroup の例外が外に出ると、次のように表示されます（ファイルのパスは /work に短縮、途中省略）。

  + Exception Group Traceback (most recent call last):
  ...
  | ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
  +-+---------------- 1 ----------------
    | Traceback (most recent call last):
    |   File "/work/a04b_tg_raw.py", line 5, in boom
    |     raise ValueError("在庫の取得に失敗")
    | ValueError: 在庫の取得に失敗
    +------------------------------------

asyncio.timeout で時間制限を付ける

import asyncio

async def slow():
    await asyncio.sleep(3)
    return "done"

async def main():
    try:
        async with asyncio.timeout(1):            # 3.11+
            await slow()
    except TimeoutError:
        print("asyncio.timeout: 1秒で打ち切り")

    try:
        await asyncio.wait_for(slow(), timeout=1)  # 1つの処理だけなら wait_for も可
    except TimeoutError:
        print("asyncio.wait_for: 1秒で打ち切り")

asyncio.run(main())

asyncio.timeout: 1秒で打ち切り
asyncio.wait_for: 1秒で打ち切り

asyncio.timeout() はブロック全体（複数の await）にまとめて時間制限を付けられます。時間を過ぎると中の処理はキャンセルされ、組み込みの TimeoutError が発生します（3.11以降は asyncio.TimeoutError と同じもの）。

Semaphore で同時実行数を制限する

相手のサーバーやDBに負荷をかけすぎないよう、同時に動く数を決めておきます。

import asyncio
import time

running = 0
peak = 0

async def fetch(i, sem):
    global running, peak
    async with sem:                     # 同時に入れるのは3つまで
        running += 1
        peak = max(peak, running)
        await asyncio.sleep(0.5)
        running -= 1
        return i

async def main():
    sem = asyncio.Semaphore(3)
    start = time.perf_counter()
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch(i, sem)) for i in range(10)]
    print("結果:", [t.result() for t in tasks])
    print(f"最大同時実行数: {peak} / 所要時間: {time.perf_counter() - start:.2f}秒")

asyncio.run(main())

結果: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
最大同時実行数: 3 / 所要時間: 2.00秒

10個を3つずつ処理するので、0.5秒×4回＝2.00秒になりました。タスクは全部作ってよく、実際に動く数だけを Semaphore が絞ります。

as_completed で終わった順に受け取る

import asyncio
import time

async def fetch(name, sec):
    await asyncio.sleep(sec)
    return f"{name}({sec}s)"

async def main():
    start = time.perf_counter()
    tasks = [asyncio.create_task(fetch(n, s)) for n, s in [("A", 0.3), ("B", 0.1), ("C", 0.2)]]
    # 終わった順に受け取る（async for は 3.13+。3.12以前は for f in as_completed(...): await f）
    async for task in asyncio.as_completed(tasks):
        print(f"{time.perf_counter() - start:.1f}秒:", await task)

asyncio.run(main())

0.1秒: B(0.1s)
0.2秒: C(0.2s)
0.3秒: A(0.3s)

進み具合を表示したいときや、早く終わったものから次の処理に回したいときに便利です。同じコードを Python 3.12 で実行すると TypeError: 'async for' requires an object with __aiter__ method, got generator になります。3.12以前はコメントのとおり通常の for で書きます。

to_thread で同期関数を並行に動かす

async に対応していないライブラリの処理は、asyncio.to_thread() でスレッドに逃がすとイベントループを止めずに済みます。

import asyncio
import time

def read_legacy(i):          # 同期でしか書けない処理（同期専用のSDK、ファイル処理など）
    time.sleep(0.5)
    return i * 10

async def main():
    start = time.perf_counter()
    results = await asyncio.gather(*(asyncio.to_thread(read_legacy, i) for i in range(3)))
    print(results, f"{time.perf_counter() - start:.2f}秒")

asyncio.run(main())

[0, 10, 20] 0.50秒

実測の表のとおり、to_thread は既定のスレッド数（min(32, CPU数 + 4)）までしか同時に動きません。大量に投げるときは Semaphore と組み合わせるか、async 対応のライブラリ（HTTPなら httpx や aiohttp）に置き換えます。

Queue で生産者・消費者を作る

import asyncio
import random

async def producer(q):
    for order_id in range(101, 107):
        await q.put(order_id)
        print(f"投入 {order_id}")
    q.shutdown()                        # 3.13+: これ以上入れない合図

async def consumer(name, q):
    while True:
        try:
            order_id = await q.get()
        except asyncio.QueueShutDown:   # 空になり、shutdown 済みなら終わる
            print(f"{name}: 終了")
            return
        await asyncio.sleep(random.uniform(0.05, 0.15))
        print(f"  {name} 処理 {order_id}")
        q.task_done()

async def main():
    random.seed(1)
    q = asyncio.Queue(maxsize=2)        # 満杯なら put が待つ（投入しすぎを防ぐ）
    async with asyncio.TaskGroup() as tg:
        tg.create_task(producer(q))
        tg.create_task(consumer("W1", q))
        tg.create_task(consumer("W2", q))

asyncio.run(main())

投入 101
投入 102
投入 103
投入 104
  W1 処理 101
投入 105
  W2 処理 102
投入 106
  W1 処理 103
  W2 処理 104
  W1 処理 105
W1: 終了
  W2 処理 106
W2: 終了

maxsize=2 なので、消費が追いつくまで投入が待たされているのがわかります。3.12以前では shutdown() がないため、終わりの印として None を人数分 put し、受け取った側が None で終わる書き方をします。

キャンセルと後片付け

import asyncio

async def download():
    try:
        print("ダウンロード開始")
        await asyncio.sleep(10)
    except asyncio.CancelledError:
        print("キャンセルを受け取った → 後片付け")
        raise                            # 握りつぶさずに投げ直す
    finally:
        print("finally: 一時ファイルを削除")

async def main():
    task = asyncio.create_task(download())
    await asyncio.sleep(0.2)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("main: キャンセル完了", task.cancelled())

asyncio.run(main())

ダウンロード開始
キャンセルを受け取った → 後片付け
finally: 一時ファイルを削除
main: キャンセル完了 True

CancelledError を受けたら、後片付けをして必ず raise で投げ直します。投げ直さないと、TaskGroup や timeout が「キャンセルが終わった」ことを正しく判断できなくなります。

httpx・aiohttp で並列にHTTPリクエストする

ローカルに「0.5秒待ってから返す」テスト用サーバーを立て、10件取得する時間を測りました（httpx 0.28.1、aiohttp 3.14.3）。

# server.py（テスト用：/items/{id} は0.5秒待ってJSONを返す）
import asyncio
from aiohttp import web

async def item(request):
    await asyncio.sleep(0.5)
    return web.json_response({"id": int(request.match_info["id"]), "stock": 3})

app = web.Application()
app.router.add_get("/items/{id}", item)
web.run_app(app, host="127.0.0.1", port=7160)

httpx（Semaphore で同時5件まで）

import asyncio
import time
import httpx  # pip install httpx

BASE = "http://127.0.0.1:7160"

async def get_item(client, sem, item_id):
    async with sem:
        r = await client.get(f"/items/{item_id}")
        r.raise_for_status()
        return r.json()

async def main():
    sem = asyncio.Semaphore(5)                         # 同時接続は5まで
    async with httpx.AsyncClient(base_url=BASE, timeout=10) as client:
        start = time.perf_counter()
        for i in range(1, 11):                          # 直列
            await get_item(client, sem, i)
        print(f"直列  : {time.perf_counter() - start:.2f}秒")

        start = time.perf_counter()
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(get_item(client, sem, i)) for i in range(1, 11)]
        print(f"並列  : {time.perf_counter() - start:.2f}秒", [t.result()["id"] for t in tasks])

asyncio.run(main())

直列  : 5.13秒
並列  : 1.01秒 [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

aiohttp（コネクタで同時5件まで）

import asyncio
import time
import aiohttp  # pip install aiohttp

BASE = "http://127.0.0.1:7160"

async def get_item(session, item_id):
    async with session.get(f"{BASE}/items/{item_id}") as r:
        r.raise_for_status()
        return await r.json()

async def main():
    connector = aiohttp.TCPConnector(limit=5)            # 同時接続は5まで
    timeout = aiohttp.ClientTimeout(total=10)
    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        start = time.perf_counter()
        results = await asyncio.gather(*(get_item(session, i) for i in range(1, 11)))
        print(f"aiohttp 並列: {time.perf_counter() - start:.2f}秒", [r["id"] for r in results])

asyncio.run(main())

aiohttp 並列: 1.01秒 [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

同時5件×0.5秒×2回で約1秒です。どちらも クライアント（セッション）は1つ作って使い回すのがポイントで、接続が再利用されて速くなります。

スレッド・マルチプロセス・asyncio の使い分け

「何を待っているか」で選びます。計算だけの重い処理（sum(i * i for i in range(600万)) を4回）でも測りました。

計算はどれも待ち時間がないため、asyncio やスレッドでは速くならず、プロセスを分けたときだけ約半分になりました。

よくあるミスと実際のエラー

async 関数の中で time.sleep や同期ライブラリを使う

実測表の最後の行のとおり、gather で10個並べても time.sleep があると5.00秒かかります。エラーは出ないので気づきにくい点に注意します。async 関数の中では await asyncio.sleep()、async 対応ライブラリ、または asyncio.to_thread() を使います。

await を書き忘れる

import asyncio

async def save(order_id):
    await asyncio.sleep(0.1)
    print("保存しました", order_id)

async def main():
    save(101)                 # await を忘れた
    print("main 終了")

asyncio.run(main())

/work/m1_noawait.py:8: RuntimeWarning: coroutine 'save' was never awaited
  save(101)                 # await を忘れた
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
main 終了

「保存しました」は表示されません。never awaited の警告が出たら、その行に await を付けるか create_task でタスクにします。

実行中のイベントループの中で asyncio.run を呼ぶ

import asyncio

async def inner():
    return 1

async def main():
    return asyncio.run(inner())   # 実行中のループの中で asyncio.run

asyncio.run(main())

RuntimeError: asyncio.run() cannot be called from a running event loop
<sys>:0: RuntimeWarning: coroutine 'inner' was never awaited

async def の中では asyncio.run() ではなく await inner() と書きます。

Jupyter で asyncio.run を使う

Jupyter（IPython 9.17.1 / ipykernel 7.3.0）で実行すると、上と同じ RuntimeError: asyncio.run() cannot be called from a running event loop になりました。Jupyter はすでにイベントループを動かしているためです。セルに直接 await fetch() と書けば動きます（結果 'ok' を確認）。

create_task の投げっぱなし

import asyncio

async def send_mail(n):
    await asyncio.sleep(0.2)
    print("送信しました", n)

async def main():
    asyncio.create_task(send_mail(1))   # 投げっぱなし
    print("main 終了")

asyncio.run(main())

main 終了

asyncio.run() は main が終わると残りのタスクをキャンセルするため、「送信しました」は出ません。作ったタスクは TaskGroup の中で作るか、変数に入れて最後に await します。

以前の書き方 asyncio.get_event_loop()

ループがない状態で asyncio.get_event_loop() を呼ぶと、Python 3.14.7 では RuntimeError: There is no current event loop in thread 'MainThread'.、3.12・3.13 では DeprecationWarning: There is no current event loop が出ました。loop = asyncio.get_event_loop(); loop.run_until_complete(main()) という以前の書き方は、asyncio.run(main()) に置き換えます。

よくある質問

Q. asyncio にすると何でも速くなりますか？ A. いいえ。速くなるのは「待ち時間」が多い処理だけです。計算だけの処理は実測のとおりほとんど変わらないため、マルチプロセスを使います。

Q. gather と TaskGroup はどちらを使えばよいですか？ A. Python 3.11 以降なら、1つでも失敗したら全体を止めたいときは TaskGroup、失敗した分を除いて結果を使いたいときは gather(return_exceptions=True) です。既定の gather は、失敗しても残りのタスクが裏で動き続ける点に気をつけます。

Q. requests を asyncio の中で使ってもよいですか？ A. そのまま呼ぶと、time.sleep と同じくイベントループ全体が止まります。asyncio.to_thread(requests.get, url) で包むか、httpx の AsyncClient や aiohttp に置き換えます。

Q. 実行中のタスクの様子を調べる方法はありますか？ A. Python 3.14 から、python -m asyncio ps <プロセスID> や python -m asyncio pstree <プロセスID> で、動いているプログラムのタスク一覧や待ち合わせの関係を表示できるようになりました（公式ドキュメントで確認。本記事では未実行）。

まとめ

• asyncio は「待ち」をまとめてさばく仕組み。0.5秒×10個の待ちが直列5.01秒 → gather・TaskGroup 0.50秒になった
• 入口で asyncio.run(main()) を1回、中は async def と await でつなぐ
• 3.11以降の並行実行は TaskGroup。1つ失敗すると残りをキャンセルし、except* で受け取れる
• 時間制限は asyncio.timeout()、同時実行数の制限は Semaphore
• 同期の処理は asyncio.to_thread()、HTTPは httpx・aiohttp でクライアントを使い回す
• async 関数の中の time.sleep は全体を止める。never awaited の警告は await の付け忘れ
• 計算が重い処理は asyncio ではなく マルチプロセスを選ぶ

株式会社RJCでは、処理の待ち時間を見極めて設計を選ぶ考え方を、プロジェクトの中で仲間と学び合っています。こうした知識を共有しながら、日々技術を磨いています。

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