Python の asyncio は、通信や待機のように「待ち時間が長い処理」を1つのスレッドの中で並行して進めるための標準ライブラリです。async def で定義した関数（コルーチン）を asyncio.run() で動かし、複数の処理を同時に待ちたいところで TaskGroup や gather() を使う、というのが基本の形です。この記事では、その書き方に加えて、同時実行数の制限、タイムアウト、例外の扱い、同期処理との混在、そして「coroutine was never awaited」などのよくあるエラーへの対処を説明します。
asyncio が向いている処理・向いていない処理
asyncio で速くなるのは、待ち時間の多い処理です。たとえば多数の API 呼び出し、Web ページの取得、DB やソケットとの通信などがこれにあたります。1つのリクエストが応答を待っている間に、別のリクエストを送ったり受け取ったりできるため、全体の所要時間が短くなります。
一方、画像変換や大量の数値計算のように CPU を使い続ける処理は、asyncio にしても速くなりません。asyncio は1つのスレッドの上で処理を切り替えているだけなので、計算中は他の処理に順番が回らず、むしろ全体が止まって見えます。こうした処理はマルチプロセスに任せるのが適切です。
スレッド（ThreadPoolExecutor）でも I/O 待ちの並行化はできます。数十件程度の並列なら、スレッドのほうが既存のコードをそのまま使えて手軽です。数百〜数千の接続を同時に扱う場合や、使いたいライブラリが非同期版を提供している場合に asyncio の利点が大きくなります。
最小の例：async def と await
まず、待ち時間を asyncio.sleep() で模した例です。
import asyncio
import time
async def job(name: str, seconds: float) -> str:
print(f"{name} 開始")
await asyncio.sleep(seconds) # ここで他のタスクに順番を譲る
print(f"{name} 終了")
return f"{name}: {seconds}秒"
async def main() -> None:
start = time.perf_counter()
results = await asyncio.gather(
job("A", 2),
job("B", 1),
job("C", 3),
)
print(results)
print(f"所要時間: {time.perf_counter() - start:.1f}秒")
if __name__ == "__main__":
asyncio.run(main())3つの処理を順番に実行すると合計6秒かかりますが、この例では最も長い3秒ほどで終わります。押さえておきたい点は3つです。
- **
async defの関数は呼んだだけでは実行されない**：job("A", 2)はコルーチンオブジェクトを返すだけで、awaitするか、タスクとして登録して初めて動きます。 - **
awaitは「ここで待つ間、他の処理に順番を譲る」という印**：awaitのない箇所で時間のかかる処理をすると、その間はほかのタスクも止まります。 - **入口は
asyncio.run()**：イベントループの作成と後片付けをまとめて行います。古い記事にあるget_event_loop()とrun_until_complete()の組み合わせは、新しいバージョンでは警告やエラーの原因になるため、新しく書くコードでは使いません。
gather() の結果は、渡した順番どおりのリストで返ります。完了した順ではない点に注意してください。
TaskGroup でまとめて実行する（Python 3.11以降）
Python 3.11 以降では asyncio.TaskGroup が使えます。async with のブロックを抜けるときに、登録したタスクがすべて終わるまで待ってくれます。
import asyncio
async def fetch(i: int) -> int:
await asyncio.sleep(0.5)
if i == 3:
raise ValueError(f"id={i} の取得に失敗")
return i * 10
async def main() -> None:
try:
async with asyncio.TaskGroup() as tg:
tasks = [tg.create_task(fetch(i)) for i in range(5)]
except* ValueError as eg:
# except* の中では return / break / continue は書けない
for e in eg.exceptions:
print("失敗:", e)
else:
print([t.result() for t in tasks])
asyncio.run(main())TaskGroup では、どれか1つのタスクが例外を出すと、残りのタスクはキャンセルされ、例外は ExceptionGroup にまとめて送出されます。そのため受け取り側は except* を使います。「1件でも失敗したら全体を止めたい」処理に向いています。
逆に「失敗したものは記録して、成功したものは使いたい」場合は、gather(..., return_exceptions=True) が扱いやすいです。結果のリストに例外オブジェクトがそのまま入るので、isinstance(r, Exception) で振り分けます。
asyncio.create_task() を単独で使う場合は、戻り値のタスクを変数やリストで保持してください。イベントループはタスクを弱い参照でしか持たないため、参照がどこにも無いと、実行途中のタスクが消えてしまうことがあると公式ドキュメントでも注意されています。
同時実行数を制限する：Semaphore
1,000件の URL を一度にすべて投げると、相手のサーバーに負荷をかけ、アクセス制限やエラーの原因になります。自分の側でもファイルディスクリプタの上限に達することがあります。asyncio.Semaphore で、同時に実行する数に上限を設けます。
import asyncio
import httpx # pip install httpx
MAX_CONCURRENCY = 10
async def fetch(client: httpx.AsyncClient, sem: asyncio.Semaphore, url: str) -> tuple[str, int]:
async with sem: # 同時に入れるのは MAX_CONCURRENCY 件まで
res = await client.get(url, timeout=10.0)
return url, res.status_code
async def main(urls: list[str]) -> None:
sem = asyncio.Semaphore(MAX_CONCURRENCY)
async with httpx.AsyncClient() as client:
results = await asyncio.gather(
*(fetch(client, sem, u) for u in urls),
return_exceptions=True,
)
for r in results:
if isinstance(r, Exception):
print("エラー:", repr(r))
else:
print(*r)
asyncio.run(main(["https://example.com/", "https://example.org/"]))ここで使っている httpx は、同期版と非同期版の両方を提供している HTTP クライアントです（aiohttp も広く使われています）。ポイントは、クライアントを1つ作って使い回すことです。リクエストのたびにクライアントを作ると、接続の再利用が効かず遅くなります。また、外部サイトへアクセスする場合は、相手の利用規約や robots.txt、API のレート制限を確認したうえで同時実行数を決めてください。
タイムアウトを付ける
応答しない相手を待ち続けると、処理全体が終わらなくなります。Python 3.11 以降では asyncio.timeout() で、ブロック全体に制限時間を設けられます。
import asyncio
async def slow() -> str:
await asyncio.sleep(10)
return "done"
async def main() -> None:
try:
async with asyncio.timeout(3):
print(await slow())
except TimeoutError:
print("3秒以内に終わらなかったため中断しました")
asyncio.run(main())3.10 以前でも使える方法として asyncio.wait_for(coro, timeout=3) があります。タイムアウトすると中の処理はキャンセルされます。HTTP クライアント側のタイムアウト（上の例の timeout=10.0）と、処理全体のタイムアウトは役割が違うので、必要に応じて両方設定します。
同期処理（ブロッキング処理）と混ぜるとき
asyncio のコードの中で、time.sleep()、requests.get()、同期版の DB ドライバのような「待っている間に順番を譲らない処理」を呼ぶと、その間イベントループ全体が止まります。エラーにはならないので気づきにくく、「非同期にしたのに速くならない」原因の多くはこれです。
対処は次の順で検討します。
- 非同期版のライブラリに置き換える：
requests→ httpx の非同期クライアント、time.sleep→asyncio.sleepなど - 置き換えられない処理はスレッドに逃がす：Python 3.9 以降の
await asyncio.to_thread(func, 引数)を使うと、同期関数を別スレッドで実行し、結果を待てます - CPU を使う重い計算はプロセスに逃がす：
loop.run_in_executor()にProcessPoolExecutorを渡します
どこでループが止まっているかを調べるには、デバッグモードが役立ちます。python -X dev script.py で実行するか、環境変数 PYTHONASYNCIODEBUG=1 を設定すると、一定時間以上ループを占有した処理が警告として表示されます。
よくあるエラーと対処
RuntimeWarning: coroutine '...' was never awaited
async def の関数を、await を付けずに呼んだときに出ます。関数の中身は実行されていません。呼び出し箇所に await を付けるか、create_task() でタスクにします。同期関数の中からコルーチンを呼んでいる場合は、その関数自体を async def にする必要があります。
RuntimeError: asyncio.run() cannot be called from a running event loop
すでにイベントループが動いている場所で asyncio.run() を呼んだときのエラーです。Jupyter Notebook や一部の Web フレームワークの中でよく起こります。Jupyter ではセルのトップレベルで await main() と直接書けるので、asyncio.run() は使いません。フレームワークの中では、そのフレームワークが用意している非同期の書き方（非同期ビュー・非同期ハンドラなど）に合わせます。
処理が終わらない・Ctrl+C で止まらない
await を含まない無限ループや、ブロッキング処理がループを占有している可能性があります。前の節のデバッグモードで占有箇所を探し、タイムアウトを追加します。
動作環境についての注意
この記事のコードは、TaskGroup、asyncio.timeout()、except* を使っているため Python 3.11 以降が前提です（2026年10月現在、3.10 は公式サポートの終了時期に差しかかっているため、新しく環境を作るなら 3.11 以降を選ぶのが無難です）。手元のバージョンは python --version で確認してください。古い環境で動かす必要がある場合は、gather() と wait_for() を使った書き方に置き換えます。asyncio は版ごとに細かな挙動や非推奨の扱いが変わるため、アップグレード時は公式ドキュメントの「What's New」で asyncio の項目を確認すると安心です。
まとめ
- asyncio は I/O 待ちの多い処理を1スレッドで並行させる仕組み。CPU を使う計算は速くならない
- 入口は
asyncio.run()、まとめて実行するならTaskGroup（3.11以降）かgather() - 外部アクセスは
Semaphoreで同時実行数を絞り、asyncio.timeout()で制限時間を付ける - 同期のブロッキング処理は非同期版に置き換えるか
asyncio.to_thread()に逃がす - 「never awaited」は
awaitの付け忘れ、「running event loop」は二重のasyncio.run()が原因