Python asyncioによる非同期処理入門

はじめに

Pythonの asyncio は、非同期I/O処理を実現する標準ライブラリです。ネットワーク通信やファイルI/Oなど、待ち時間（I/Oバウンド）の多い処理を効率的に並行実行できます。

本記事では、async/await 構文の基本からタスクの並行実行、実践的な非同期HTTPリクエストまでを解説します。

同期 vs 非同期

同期処理の問題

import time

def fetch_data(url, delay):
    print(f"Fetching {url}...")
    time.sleep(delay)  # I/O待ちをシミュレート
    print(f"Done {url}")
    return f"data from {url}"

# 逐次実行: 合計6秒
start = time.time()
fetch_data("api/users", 2)
fetch_data("api/posts", 3)
fetch_data("api/comments", 1)
print(f"Total: {time.time() - start:.1f}s")  # 約6秒

非同期処理による改善

import asyncio

async def fetch_data(url, delay):
    print(f"Fetching {url}...")
    await asyncio.sleep(delay)  # 非同期I/O待ち
    print(f"Done {url}")
    return f"data from {url}"

async def main():
    results = await asyncio.gather(
        fetch_data("api/users", 2),
        fetch_data("api/posts", 3),
        fetch_data("api/comments", 1),
    )
    return results

# 並行実行: 約3秒（最も遅いタスクの時間）
import time
start = time.time()
results = asyncio.run(main())
print(f"Total: {time.time() - start:.1f}s")  # 約3秒

イベントループの仕組み：なぜ1スレッドで大量のI/Oを捌けるのか

asyncio の速さの正体は「魔法のような並列実行」ではなく、協調的（cooperative） な単一スレッド内でのタスク切り替えです。仕組みを正確に理解しておくと、後述する「やってはいけないこと」の理由も自然に見えてきます。

協調的マルチタスクの流れ

イベントループは実行可能なコルーチンを1つ選び、await に到達するまでそのまま実行する。
await some_io() に到達すると、コルーチンは自分から実行権をイベントループに明け渡す（yieldする）。OSのプリエンプション（強制的な割り込み）ではなく、コード側が自発的に制御を返す点がスレッドとの決定的な違い。
イベントループは、control が戻ってきた間に他の実行可能なコルーチンを実行する。
I/O待ちしていたソケットやファイルディスクリプタが「読み書き可能」になったら、イベントループがそれを検知し、対応するコルーチンを再開キューに積む。

この4番目のステップを支えているのが、OSカーネルが提供する I/O多重化（I/O multiplexing） の仕組みです。CPython の asyncio は内部で selectors モジュール（select モジュールの薄いラッパー）を使い、プラットフォームごとに最も効率の良い実装を自動選択します。

import selectors

# DefaultSelectorはOSごとに最適な実装を自動選択する
# Linux: EpollSelector (epoll)
# macOS/BSD: KqueueSelector (kqueue)
# それ以外: SelectSelector (select) など
sel = selectors.DefaultSelector()
print(type(sel).__name__)  # 例: KqueueSelector (macOS), EpollSelector (Linux)

select(2) は監視対象のファイルディスクリプタ数に対して線形（O(n)）にコストが増えるのに対し、epoll（Linux）や kqueue（BSD/macOS）はカーネル側に監視対象を登録しておき、レディになったものだけを通知してもらう方式（O(1)に近い）です。数千〜数万コネクションを同時に扱う「C10K問題」で epoll/kqueue が主流になったのはこのためで、asyncio のイベントループはこの仕組みの上に構築されています。

重要なのは、このループを回しているのはただ1本のOSスレッドだという点です。1万個のコネクションを待つのに1万個のOSスレッドを作る必要はなく、カーネルに「このソケット群のどれかが準備できたら教えて」と一括で問い合わせるだけで済みます。

threading（GIL）・multiprocessing との違い

I/Oバウンドな並行処理を実現する手段は asyncio だけではありません。それぞれのトレードオフを正確に比較します。

方式	並行の単位	メモリ	GILの影響	主なオーバーヘッド	得意な処理
`asyncio`	コルーチン（1スレッド内）	プロセス内で共有	該当なし（そもそも1スレッド）	タスク切り替えはユーザー空間の関数呼び出し程度で軽い	I/Oバウンド（大量の同時接続）
`threading`	OSスレッド	プロセス内で共有	GILにより同時に1スレッドしかPythonバイトコードを実行できない（CPUバウンドは並列化されない）	OSによるコンテキストスイッチ（レジスタ退避・カーネルモード遷移）	ブロッキングI/O呼び出しを隔離する用途（`to_thread`など）
`multiprocessing`	OSプロセス	共有されない（別アドレス空間、IPCが必要）	該当なし（プロセスごとに独立したGIL）	プロセス生成（fork/spawn）のコストとプロセス間通信のシリアライズコスト	CPUバウンド（真の並列計算）

threading: CPython の GIL（Global Interpreter Lock）は「一度に1つのOSスレッドしかPythonバイトコードを実行できない」という制約です。そのためI/O待ち中はGILが解放され他スレッドが動けますが、CPUバウンドな処理は複数スレッドを使っても速くなりません。加えてOSスレッドはコンテキストスイッチのたびにカーネルモードへの遷移とレジスタ退避が発生し、スレッド数が増えるほどこのオーバーヘッドが無視できなくなります。
multiprocessing: プロセスは独立したメモリ空間を持つため、GILの制約を受けずに真の並列実行ができます。しかし fork/spawn によるプロセス生成コストが大きく、プロセス間でデータをやり取りするにはpickle化などのシリアライズが必要になり、共有メモリのように単純にオブジェクトを渡すことはできません。
asyncio: 1スレッドの中でコルーチンを切り替えるだけなので、コンテキストスイッチもプロセス生成も発生しません。数千の同時接続を処理してもオーバーヘッドがほぼ増えないのはこのためです。ただしこれは「I/O待ちの間に他の仕事をする」という発想そのものであり、CPUを使う計算そのものを高速化する仕組みではない、という点が次章の重要な前提になります。

（CPython 3.13以降ではビルドオプションでGILを無効化できる「フリースレッド版」も提供されていますが、これはCPUバウンドなマルチスレッド処理の並列化を目的としたもので、I/Oバウンドな asyncio の設計判断には影響しません。詳細は末尾の参考文献を参照してください。）

コルーチンとasync/await

コルーチンの定義

async def で定義された関数はコルーチン関数であり、呼び出すとコルーチンオブジェクトを返します。

async def greet(name):
    return f"Hello, {name}"

# コルーチンオブジェクトが返る（実行はされない）
coro = greet("Alice")

# 実行するにはawaitまたはasyncio.run()が必要
result = asyncio.run(greet("Alice"))
print(result)  # "Hello, Alice"

awaitの役割

await は非同期処理の完了を待つ式です。await 中にイベントループは他のタスクに制御を移すことができます。

async def process():
    data = await fetch_data("api/users", 1)  # 待機中に他のタスクが実行可能
    return data

タスクの並行実行

asyncio.create_task()

コルーチンをタスクとしてスケジュールし、バックグラウンドで実行を開始します。

async def main():
    task1 = asyncio.create_task(fetch_data("api/users", 2))
    task2 = asyncio.create_task(fetch_data("api/posts", 3))

    # 両タスクはここで並行実行中
    result1 = await task1
    result2 = await task2
    return result1, result2

asyncio.gather()

複数のコルーチンを同時に実行し、すべての結果をまとめて返します。

async def main():
    results = await asyncio.gather(
        fetch_data("api/users", 2),
        fetch_data("api/posts", 3),
        fetch_data("api/comments", 1),
    )
    # resultsは入力順にリストで返される
    return results

asyncio.as_completed()

完了順にイテレートしたい場合に使います。

async def main():
    tasks = [
        fetch_data("api/users", 2),
        fetch_data("api/posts", 3),
        fetch_data("api/comments", 1),
    ]
    for coro in asyncio.as_completed(tasks):
        result = await coro
        print(f"Completed: {result}")
    # 出力順: comments → users → posts（完了が早い順）

実践例：非同期HTTPリクエスト

aiohttp を使った並行HTTPリクエストの例です。

import asyncio
import aiohttp
import time

async def fetch_url(session, url):
    """単一URLの非同期フェッチ"""
    async with session.get(url) as response:
        data = await response.text()
        return {"url": url, "status": response.status, "length": len(data)}

async def fetch_all(urls):
    """複数URLの並行フェッチ"""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        return await asyncio.gather(*tasks)

# --- 実行 ---
urls = [
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/2",
    "https://httpbin.org/delay/1",
    "https://httpbin.org/delay/2",
    "https://httpbin.org/delay/1",
]

start = time.time()
results = asyncio.run(fetch_all(urls))
elapsed = time.time() - start

for r in results:
    print(f"{r['url']}: status={r['status']}, length={r['length']}")
print(f"Total: {elapsed:.1f}s")  # 逐次なら7秒、並行なら約2秒

非同期パターン

Semaphoreによるレート制限

同時接続数を制限して、サーバーに負荷をかけすぎないようにします。

async def fetch_with_limit(session, url, semaphore):
    async with semaphore:  # 同時実行数を制限
        async with session.get(url) as response:
            return await response.text()

async def main():
    semaphore = asyncio.Semaphore(5)  # 最大5並行
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_with_limit(session, url, semaphore) for url in urls]
        return await asyncio.gather(*tasks)

Queueによるプロデューサー・コンシューマー

async def producer(queue):
    for i in range(10):
        await queue.put(f"item-{i}")
        await asyncio.sleep(0.1)
    await queue.put(None)  # 終了シグナル

async def consumer(queue, name):
    while True:
        item = await queue.get()
        if item is None:
            await queue.put(None)  # 他のコンシューマーにも伝搬
            break
        print(f"{name} processed {item}")
        await asyncio.sleep(0.2)

async def main():
    queue = asyncio.Queue(maxsize=5)
    await asyncio.gather(
        producer(queue),
        consumer(queue, "worker-1"),
        consumer(queue, "worker-2"),
    )

エッジケース①：ブロッキング呼び出しがイベントループ全体を止める

初心者が最も陥りやすい間違いは、コルーチンの中で同期的にブロックする関数（time.sleep() や、await の付いていない同期I/O・重いCPU処理）を直接呼び出してしまうことです。よくある誤解は「ブロックされるのはその1つのコルーチンだけ」というものですが、実際にはそのブロッキング呼び出しを実行している間、イベントループそのものが停止し、他の全てのタスクも進行できなくなります。前章で説明した通り、イベントループはただ1本のスレッドで動いているため、そのスレッドが time.sleep() の中でOSに制御を奪われている間は、他のコルーチンに実行権を渡す機会が一切ないのです。

以下のコードで、3つのタスクを同時に gather した際の挙動を比較します。

import asyncio
import time


async def main_blocking():
    """1つのタスクがtime.sleep()（ブロッキング）を使うケース"""
    t0 = time.perf_counter()

    async def worker(name, use_blocking, duration):
        start = time.perf_counter() - t0
        print(f"[{start:6.3f}] {name} START")
        if use_blocking:
            time.sleep(duration)  # NG: イベントループ全体を止める
        else:
            await asyncio.sleep(duration)
        end = time.perf_counter() - t0
        print(f"[{end:6.3f}] {name} END (took {end - start:.3f}s)")

    await asyncio.gather(
        worker("blocking-task", True, 2.0),
        worker("async-task-A", False, 0.5),
        worker("async-task-B", False, 0.5),
    )
    print(f"total: {time.perf_counter() - t0:.3f}s")


async def main_nonblocking():
    """全タスクがasyncio.sleep()（非ブロッキング）を使うケース"""
    t0 = time.perf_counter()

    async def worker(name, duration):
        start = time.perf_counter() - t0
        print(f"[{start:6.3f}] {name} START")
        await asyncio.sleep(duration)
        end = time.perf_counter() - t0
        print(f"[{end:6.3f}] {name} END (took {end - start:.3f}s)")

    await asyncio.gather(
        worker("async-task-C", 2.0),
        worker("async-task-A", 0.5),
        worker("async-task-B", 0.5),
    )
    print(f"total: {time.perf_counter() - t0:.3f}s")


asyncio.run(main_blocking())
print()
asyncio.run(main_nonblocking())

実際に実行した結果（Python 3.14、macOS）は次の通りです。

[ 0.000] blocking-task START
[ 2.005] blocking-task END (took 2.005s)
[ 2.005] async-task-A START
[ 2.005] async-task-B START
[ 2.506] async-task-A END (took 0.501s)
[ 2.506] async-task-B END (took 0.501s)
total: 2.507s

[ 0.000] async-task-C START
[ 0.000] async-task-A START
[ 0.000] async-task-B START
[ 0.501] async-task-A END (took 0.500s)
[ 0.501] async-task-B END (took 0.500s)
[ 2.001] async-task-C END (took 2.001s)
total: 2.001s

time.sleep() を使ったケースでは、async-task-A・async-task-B は本来 0.5 秒で終わるはずなのに、blocking-task の time.sleep(2.0) が終わるまでSTARTすらできていません。イベントループが blocking-task に専有され、他のタスクへ制御を渡す隙がなかったためです。結果として合計時間は 2.5 秒（2.0秒のブロック + 0.5秒）となり、本来の並行処理（0.5秒と2.0秒のうち遅い方＝2.0秒程度で終わるはず）よりも明らかに遅くなっています。一方、全タスクが asyncio.sleep() を使った下段のケースでは、3つのタスクが同時にSTARTし、合計時間は最も遅いタスク（2.0秒）とほぼ同じ 2.001 秒に収まっています。

このタイムラインを図示すると次のようになります。

asyncio: time.sleepによるブロックとasyncio.sleepによる非ブロックのタスク実行タイムライン比較

対策: コルーチンの中でブロッキング処理（同期I/O、requests のような同期HTTPクライアント、重いCPU計算など）を呼ぶ必要がある場合は、必ず asyncio.to_thread()（Python 3.9+）または loop.run_in_executor() で別スレッド・別プロセスに逃がします。

async def worker(name, duration):
    # time.sleep(duration) を直接呼ばず、別スレッドで実行する
    await asyncio.to_thread(time.sleep, duration)

エッジケース②：CPUバウンドな処理にasyncioは効かない

asyncio はI/Oバウンドな処理（ネットワーク待ち、ディスクI/O待ちなど）を効率化する仕組みであり、CPUを使う計算そのものを速くするものではありません。前章で説明した通り、イベントループは単一スレッドで動くため、asyncio.gather() でCPUバウンドなコルーチンを束ねても、それらは並行に実行されるわけではなく、単に1本のスレッドの中で順番に実行されるだけです。

以下は、CPUバウンドな計算（6,000,000回のループ）を4つ、(1) 逐次実行、(2) asyncio.gather、(3) multiprocessing（ProcessPoolExecutor）で比較したベンチマークです。

import asyncio
import time
from concurrent.futures import ProcessPoolExecutor

N_TASKS = 4
N = 6_000_000


def cpu_bound_task(n):
    """計算だけを行うCPUバウンドな関数（I/O待ちなし）"""
    total = 0
    for i in range(n):
        total += i * i
    return total


async def cpu_bound_coro(n):
    # awaitを一切含まない -- 単なる同期関数の皮を被ったコルーチン
    return cpu_bound_task(n)


def run_sequential():
    start = time.perf_counter()
    results = [cpu_bound_task(N) for _ in range(N_TASKS)]
    return time.perf_counter() - start, results


def run_asyncio_gather():
    async def main():
        return await asyncio.gather(*[cpu_bound_coro(N) for _ in range(N_TASKS)])

    start = time.perf_counter()
    results = asyncio.run(main())
    return time.perf_counter() - start, results


def run_multiprocessing():
    start = time.perf_counter()
    with ProcessPoolExecutor(max_workers=N_TASKS) as pool:
        results = list(pool.map(cpu_bound_task, [N] * N_TASKS))
    return time.perf_counter() - start, results


if __name__ == "__main__":
    t_seq, r_seq = run_sequential()
    t_async, r_async = run_asyncio_gather()
    t_mp, r_mp = run_multiprocessing()

    print(f"Sequential (for-loop):      {t_seq:.3f}s")
    print(f"asyncio.gather (1 thread):  {t_async:.3f}s")
    print(f"multiprocessing (4 procs):  {t_mp:.3f}s")
    print(f"asyncio vs sequential speedup: {t_seq / t_async:.2f}x (expect ~1x)")
    print(f"multiprocessing vs sequential speedup: {t_seq / t_mp:.2f}x")

実際に実行した結果（Python 3.14、8コアのmacOS環境）は次の通りです。

Sequential (for-loop):      1.223s
asyncio.gather (1 thread):  1.015s
multiprocessing (4 procs):  0.428s
asyncio vs sequential speedup: 1.20x (expect ~1x)
multiprocessing vs sequential speedup: 2.86x

asyncio.gather は逐次実行とほぼ同じ時間（誤差レベルの差）で終わっており、speedupと呼べるほどの改善はありません。対して multiprocessing は実測で約2.9倍高速化しています（理論上の上限はコア数の4倍ですが、プロセス生成やpickleによるシリアライズのオーバーヘッドがあるため、実測値はそれよりやや低くなります）。ベンチマーク数値はマシン構成やタイミングによって変動するため、絶対値そのものよりも「asyncioは逐次とほぼ同じ・multiprocessingは明確に速い」という定性的な傾向を読み取ってください。

CPUバウンドタスクの実行時間比較：逐次実行・asyncio.gather・multiprocessingの棒グラフ

対策: CPUバウンドな処理を並列化したい場合は、asyncio ではなく multiprocessing や concurrent.futures.ProcessPoolExecutor を使います。非同期コードの中からCPUバウンドな処理を呼び出す必要がある場合は、loop.run_in_executor(ProcessPoolExecutor(), func, arg) のようにイベントループから別プロセスへオフロードします。

エラーハンドリング

gatherでの例外伝播（実行検証）

asyncio.gather() に渡した複数のコルーチンのうち1つが例外を送出すると、デフォルト（return_exceptions=False）では gather() はその例外を即座に呼び出し元へ伝播させます。ここで見落とされがちなのが、他の（まだ完了していない）タスクが自動的にキャンセルされるわけではないという点です。既に create_task() で生成済みのタスクは、gather() が例外を送出した後もバックグラウンドで実行を継続します。

import asyncio


async def risky_task(n, delay):
    await asyncio.sleep(delay)
    if n == 2:
        raise ValueError(f"error in task {n}")
    return f"result-{n}"


async def without_return_exceptions():
    t1 = asyncio.create_task(risky_task(1, 0.3), name="task-1")
    t2 = asyncio.create_task(risky_task(2, 0.1), name="task-2")  # 最初に失敗する
    t3 = asyncio.create_task(risky_task(3, 0.5), name="task-3")  # 最も遅い

    try:
        results = await asyncio.gather(t1, t2, t3)
        print("results:", results)
    except ValueError as e:
        print(f"gather raised: {e!r}")

    print("--- 例外伝播の直後の状態 ---")
    for t in (t1, t2, t3):
        print(f"  {t.get_name()}: done={t.done()} cancelled={t.cancelled()}")

    await asyncio.sleep(0.6)  # バックグラウンドのタスクが終わるのを待つ
    print("--- 待機後の状態 ---")
    for t in (t1, t2, t3):
        print(f"  {t.get_name()}: done={t.done()} cancelled={t.cancelled()}")


asyncio.run(without_return_exceptions())

実際の出力：

gather raised: ValueError('error in task 2')
--- 例外伝播の直後の状態 ---
  task-1: done=False cancelled=False
  task-2: done=True cancelled=False
  task-3: done=False cancelled=False
--- 待機後の状態 ---
  task-1: done=True cancelled=False
  task-2: done=True cancelled=False
  task-3: done=True cancelled=False

task-2 が0.1秒後に例外を送出した瞬間、gather() はその例外をすぐに呼び出し元へ伝播させますが、その時点で task-1（0.3秒後に完了予定）と task-3（0.5秒後に完了予定）は done=False のまま、つまりキャンセルされず、バックグラウンドで実行され続けています。呼び出し元が例外をキャッチして処理を先に進めてしまうと、これらのタスクは誰にも観測されないまま裏で動き続け、結果が握りつぶされたり、さらに例外を送出した場合に “Task exception was never retrieved” という警告が出たりすることがあります。これが gather() のexcept-and-forget的な使い方に潜む落とし穴です。

return_exceptions=True を指定すると、例外は結果リストの中に値として格納され、他のタスクの結果と一緒にすべて待ち合わせてから返されます。

async def with_return_exceptions():
    t1 = asyncio.create_task(risky_task(1, 0.3), name="task-1")
    t2 = asyncio.create_task(risky_task(2, 0.1), name="task-2")
    t3 = asyncio.create_task(risky_task(3, 0.5), name="task-3")

    results = await asyncio.gather(t1, t2, t3, return_exceptions=True)
    for r in results:
        if isinstance(r, Exception):
            print(f"Error: {r}")
        else:
            print(f"OK: {r}")

実際の出力：

OK: result-1
Error: error in task 2
OK: result-3

TaskGroupなら失敗した瞬間に兄弟タスクをキャンセルできる

「1つ失敗したら残りも打ち切りたい」という構造化された並行処理をしたい場合は、gather() ではなく Python 3.11 で追加された asyncio.TaskGroup を使います。TaskGroup は子タスクの1つが例外を送出すると、他の子タスクを明示的にキャンセルしてから ExceptionGroup をまとめて送出します。

async def with_taskgroup():
    tasks = {}
    try:
        async with asyncio.TaskGroup() as tg:
            tasks["task-1"] = tg.create_task(risky_task(1, 0.3), name="task-1")
            tasks["task-2"] = tg.create_task(risky_task(2, 0.1), name="task-2")
            tasks["task-3"] = tg.create_task(risky_task(3, 0.5), name="task-3")
    except* ValueError as eg:
        print(f"ExceptionGroup caught: {[repr(e) for e in eg.exceptions]}")

    for name, t in tasks.items():
        print(f"  {name}: done={t.done()} cancelled={t.cancelled()}")


asyncio.run(with_taskgroup())

実際の出力：

ExceptionGroup caught: ["ValueError('error in task 2')"]
  task-1: done=True cancelled=True
  task-2: done=True cancelled=False
  task-3: done=True cancelled=True

gather() の場合と違い、task-1 と task-3 は cancelled=True になっています。task-2 が失敗した時点で TaskGroup が残りのタスクに cancel() を呼び出したためです。新しいコードでは、単純な集約には gather()、失敗時に全体を打ち切りたい構造化された並行処理には TaskGroup を使い分けるのが安全です。

よくある落とし穴

落とし穴	対策
イベントループのブロック	ブロッキングI/O・CPU負荷の高い処理は `asyncio.to_thread()` / `run_in_executor` に逃がす
CPUバウンド処理にasyncioを使う	速度は逐次実行とほぼ同じ。並列化したいなら `multiprocessing`/`ProcessPoolExecutor` を使う
`gather()` の例外で兄弟タスクが残り続ける	打ち切りたいなら `TaskGroup`、集約したいだけなら `return_exceptions=True`
`await` の付け忘れ	コルーチンが実行されず `RuntimeWarning` が出る
同期コードからの呼び出し	`asyncio.run()` で実行、ネストは `nest_asyncio`
タスクの参照消失	`create_task` の結果を変数に保持する

PythonでCSVファイルを効率的に読み書きする方法：csv・pandas・polars比較 - 多数のCSVファイルを asyncio.gather や ProcessPoolExecutor と組み合わせて並行処理する具体例を解説しています。
Pythonのloggingモジュール実践ガイド：printからの卒業 - 非同期タスクごとの進捗・エラー追跡に欠かせない構造化ロギングの設計を解説しています。
Python型ヒント実践ガイド：基礎からmypy活用まで - Awaitable[T]・Coroutine・AsyncIterator など非同期APIに型を付けるためのベストプラクティスを解説しています。
Pythonデコレータ完全ガイド - asyncデコレータの実装にも役立つデコレータパターンを解説しています。
Python正規表現パターン集 - 非同期テキスト処理と組み合わせて使える正規表現のパターンを紹介しています。
Pythonでプログレスバーを自作する - 非同期処理の進捗表示に応用できるプログレスバーの実装を解説しています。
Matplotlib実践Tips：論文品質のグラフを作る - 非同期処理で取得したデータの可視化に役立つTipsを紹介しています。
pythonでprintの上書きをする方法 - Pythonの実践的なTipsを紹介しています。
PythonのMatplotlibで3Dアニメーション（GIF）を作成する方法 - Pythonの応用的な使い方を紹介しています。
PythonでSlackに実験結果を通知する方法 - 外部APIを叩く実例を紹介しています。非同期処理と組み合わせることで効率的な通知が可能です。

参考文献

Python公式ドキュメント: asyncio
Python公式ドキュメント: selectors — 高水準のI/O多重化 - epoll/kqueue/select をラップし、asyncioのイベントループが内部で利用するI/O多重化の仕組みを解説しています。
aiohttp公式ドキュメント
PEP 492 – Coroutines with async and await syntax
What’s New In Python 3.13 — asyncio - asyncio.as_completed() がTask/Futureをそのまま返すようになった変更や、TaskGroup のキャンセル挙動の改善など、最新の変更点をまとめた公式ドキュメントです。
Python公式ドキュメント: Free-threaded CPython (PEP 703) - Python 3.13から提供されるGILを無効化したビルドについての公式ガイド。CPUバウンドなマルチスレッド処理の並列化が目的で、I/Oバウンドなasyncioの設計とは独立した改善であることが説明されています。
DevToolBox JSON整形ツール - 非同期処理で取得したJSONレスポンスの確認・整形に手軽に使える無料ツールです。

はじめに

同期 vs 非同期

同期処理の問題

非同期処理による改善

イベントループの仕組み：なぜ1スレッドで大量のI/Oを捌けるのか

協調的マルチタスクの流れ

threading（GIL）・multiprocessing との違い

コルーチンとasync/await

コルーチンの定義

awaitの役割

タスクの並行実行

asyncio.create_task()

asyncio.gather()

asyncio.as_completed()

実践例：非同期HTTPリクエスト

非同期パターン

Semaphoreによるレート制限

Queueによるプロデューサー・コンシューマー

エッジケース①：ブロッキング呼び出しがイベントループ全体を止める

エッジケース②：CPUバウンドな処理にasyncioは効かない

エラーハンドリング

gatherでの例外伝播（実行検証）

TaskGroupなら失敗した瞬間に兄弟タスクをキャンセルできる

よくある落とし穴

関連記事

参考文献

この記事に関連する無料ツール

関連記事

Python型ヒント実践ガイド：基礎からmypy活用まで

Pythonのloggingモジュール実践ガイド：printからの卒業

Pythonデコレータの仕組みと実践パターン