PythonのAsyncio：async/await、TaskGroup、スレッド、マルチプロセス、GIL

asyncioはPythonの標準ライブラリで、非同期処理のためのものです。async def で定義した関数は await で一時停止でき、その間にイベントループが別の処理を実行します。その結果、1つのスレッドで数百のネットワーク要求を同時に進められます。ただし、asyncioは3つの道具の1つにすぎません。ほかに、スレッド（threading）とプロセス（multiprocessing）があります。ここでは、asyncioの仕組み、タスクに応じた選び方、そしてGILの役割を説明します。

待つことと計算すること：別の問題

この区別を理解していないと、以降の内容は分かりにくくなります。プログラムが遅くなる原因は、大きく2つに分かれます。

• I/O-bound（入出力待ち） ― APIの応答、データベース、ディスクの読み込みを待っている状態です。このときプロセッサは遊んでいます。
• CPU-bound（計算待ち） ― 画像処理や、数百万件のレコードに対するループなど、計算をしている状態です。プロセッサは忙しく働いています。

待ち時間には、asyncioやスレッドが向いています。ある処理が待っている間に、別の処理を進められるからです。計算には、複数のコアで本当に並列に計算するプロセスが必要です。ここを取り違えるのが最大のミスです。純粋なPythonの計算をスレッドで速くすることはできません。一方、3件のHTTPリクエストのためにプロセスを立ち上げるのは、やりすぎです。

asyncioとは：コルーチンとイベントループ

非同期コードは2つの言葉で成り立っています。async def はコルーチンを定義します。コルーチンは、途中で一時停止できる関数です。呼び出しても本体は実行されず、コルーチンのオブジェクトが作られます。内部の await は「ここで待つので、その間は他の処理を進めてください」という意味です。全体を管理するのは、イベントループです。asyncio.run() で起動します。

import asyncio

async def fetch_rate(currency):
    await asyncio.sleep(1)          # APIリクエストの模擬
    return {"USD": 150.2, "EUR": 163.5, "CNY": 21.0}[currency]

async def main():
    rate = await fetch_rate("USD")
    print(rate)

asyncio.run(main())  # 150.2

asyncio.sleep() は非同期の待機です。イベントループに制御を返します。一方、コルーチンの中で time.sleep() を使うと、ループ全体が止まってしまい、他のタスクも待たされます。

順番に実行するか、同時に実行するか

ループの中で await を書くと、リクエストは1件ずつ順番に処理されます。3つの通貨で、合計3秒かかります。同時に進めるには、それぞれをタスクとして起動します。

import asyncio, time

async def main():
    start = time.perf_counter()
    for cur in ("USD", "EUR", "CNY"):
        await fetch_rate(cur)                       # 1件ずつ
    print(f"{time.perf_counter() - start:.1f} 秒")   # 3.0 秒

asyncio.run(main())

TaskGroup：まとめて起動し、全部の完了を待つ

現在の推奨方法は、タスクグループです（Python 3.11以降）。async with ブロックの中で、tg.create_task() によってそれぞれのタスクがすぐに開始されます。ブロックを抜けるのは、すべてのタスクが終わったときです。

async def main():
    start = time.perf_counter()
    async with asyncio.TaskGroup() as tg:
        tasks = {cur: tg.create_task(fetch_rate(cur)) for cur in ("USD", "EUR", "CNY")}
    rates = {cur: t.result() for cur, t in tasks.items()}
    print(rates, f"{time.perf_counter() - start:.1f} 秒")   # ... 1.0 秒

asyncio.run(main())

TaskGroupは厳格です。1つのタスクが失敗すると、残りのタスクはキャンセルされ、すべてのエラーが ExceptionGroup にまとめられます。それは except* で捕まえます。

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fetch_rate("USD"))
            tg.create_task(fetch_rate("XXX"))   # KeyError
    except* KeyError as eg:
        print("未知の通貨:", eg.exceptions)

asyncio.gatherとcreate_task

TaskGroupより前は、asyncio.gather() が標準でした。コルーチンを同時に起動し、結果を元の順番で返します。既存のコードには、今でもたくさん使われています。

async def main():
    usd, eur, cny = await asyncio.gather(
        fetch_rate("USD"), fetch_rate("EUR"), fetch_rate("CNY")
    )
    # return_exceptions=True を指定すると、エラーは全体を止めずにリストで返る

asyncio.create_task() は、グループを使わずにコルーチンをバックグラウンドで起動します。タスクへの参照を保持しておかないと、ガベージコレクタに消されたり、エラーが誰にも見られなかったりします。そのため、新しいプロジェクトではTaskGroupの方が安全です。

タスクのキャンセルとタイムアウト

async def main():
    task = asyncio.create_task(fetch_rate("USD"))
    task.cancel()                     # タスク内で CancelledError が発生する
    try:
        await task
    except asyncio.CancelledError:
        print("キャンセルされました")

    try:
        async with asyncio.timeout(0.5):   # Python 3.11以降
            await fetch_rate("EUR")
    except TimeoutError:
        print("時間切れ")

古いバージョンでは、asyncio.timeout() の代わりに asyncio.wait_for(coro, timeout=0.5) を使います。また、asyncio.wait() は、完了したタスクと未完了のタスクの集合を返します。return_when=asyncio.FIRST_COMPLETED を指定すると、最初に終わったものを待てます。

講座で紹介したとおり、Python 3.15ではTaskGroupに tg.cancel() メソッドが追加されました。グループのすべてのタスクを、例外を使わずにキャンセルできます。これにより、「最初に答えたものを採用する」という処理が書きやすくなります。以前は、独自の例外を投げてシグナルにするか、タスクを手作業でキャンセルする必要がありました。

同期：Semaphore、Lock、Event、Queue

asyncioには、await と組み合わせて使う独自の同期プリミティブがあります。最も役立つのは Semaphore です。同時に実行される処理の数を制限します。千件のリクエストで相手のAPIをダウンさせるのを防げます。

sem = asyncio.Semaphore(10)

async def limited(url):
    async with sem:              # 同時に実行されるのは最大10件
        await asyncio.sleep(0.1) # ここにリクエストが入る
        return url

• asyncio.Lock ― 臨界区間に入れるコルーチンは1つだけです。
• asyncio.Event ― あるコルーチンは await event.wait() で待ち、別のコルーチンは event.set() で合図します。
• asyncio.Queue ― 生産者と消費者のためのキューです。await q.put()、await q.get()、q.task_done() を使います。

落とし穴：awaitの書き忘れ

最も多いミスは、await を付けずにコルーチンを呼ぶことです。

async def main():
    rate = fetch_rate("USD")    # await を忘れた
    print(rate * 2)             # TypeError
# RuntimeWarning: coroutine 'fetch_rate' was never awaited

Pythonは関数を実行したのではなく、コルーチンのオブジェクトを作っただけです。ログに「was never awaited」と出ていたら、await が抜けていないか探してください。2つ目のよくあるミスは、コルーチンの中でブロッキングな呼び出し（time.sleep、requests.get）を使うことです。HTTP通信には非同期対応のライブラリ（httpx、aiohttp）を使い、どうしても同期的な処理が必要な場合は、await asyncio.to_thread(func, arg) でスレッドに逃がしましょう。

Pythonのマルチスレッド：threadingモジュール

スレッドは、同じプロセスの中で独立した実行の流れです。メモリは共有されます。ネットワークの待ち時間には非常に有効で、コードを非同期に書き換える必要もありません。

import threading, time

def download(name):
    time.sleep(1)               # ネットワーク待ちの模擬
    print(name, "完了")

threads = [threading.Thread(target=download, args=(n,)) for n in ("a", "b", "c")]
for t in threads:
    t.start()
for t in threads:
    t.join()                    # 合計でおよそ1秒

メモリを共有することは、利点でもあり、バグの元にもなります。2つのスレッドが同じ変数を変更するなら、threading.Lock が必要です。通常は、スレッドを直接操作するよりも、プールを使う方が簡単です。

ThreadPoolExecutor

concurrent.futures.ThreadPoolExecutor は、スレッドのプールを作り、タスクを割り振ります。map は、元の順番どおりに結果を返します。

from concurrent.futures import ThreadPoolExecutor
import time

def fetch(url):
    time.sleep(1)
    return f"{url}: 200"

urls = [f"https://example.com/{i}" for i in range(5)]
with ThreadPoolExecutor(max_workers=5) as pool:
    for result in pool.map(fetch, urls):
        print(result)           # 合計でおよそ1秒

GILとは何か、そしてスレッドが計算を速くしない理由

GIL（Global Interpreter Lock、グローバルインタープリターロック）は、CPythonのグローバルなロックです。ある時点で、Pythonのバイトコードを実行できるスレッドは1つだけです。スレッドが入出力を待っているときは、GILを手放すので、ネットワーク処理にはスレッドが役立ちます。しかし純粋な計算では、4つのスレッドを使っても1つのときより速くなりません。スレッドが順番にロックを受け渡すからです。

from concurrent.futures import ThreadPoolExecutor

def heavy(n):
    return sum(i * i for i in range(n))

with ThreadPoolExecutor(4) as pool:
    list(pool.map(heavy, [5_000_000] * 4))   # ほぼ逐次実行と同じ時間

NumPyのようなC言語で書かれたライブラリは、重い処理の間にGILを手放すことが多いため、スレッドで速くなる場合があります。

multiprocessing：本当の並列処理

プロセスごとに独立したインタープリターとGILがあるため、複数のコアで同時に計算できます。代わりに、メモリは別々です。プロセス同士でデータを渡すには、シリアライズ（pickle）が必要になり、プロセスの起動はスレッドより重くなります。

from concurrent.futures import ProcessPoolExecutor

def heavy(n):
    return sum(i * i for i in range(n))

if __name__ == "__main__":      # WindowsとmacOSでは必須
    with ProcessPoolExecutor() as pool:
        print(list(pool.map(heavy, [5_000_000] * 4)))

if __name__ == "__main__" を書かないと、子プロセスがモジュールを読み込む際に、さらに子プロセスを起動しようとしてしまいます。低レベルの multiprocessing モジュールには Process、Pool、Queue もありますが、ProcessPoolExecutor の方が簡単で、スレッドプールと同じインターフェースです。

asyncio、threading、multiprocessingの選び方

GILのないPythonとサブインタープリター：新しい動き

Python 3.13からは、GILのないビルド（free-threading、PEP 703）が用意されています。名前に t が付きます（例：python3.15t）。3.13では実験的でしたが、3.14（PEP 779）からは正式にサポートされています。このビルドでは、スレッドが複数のコアで並列に計算します。ただし、標準ではGILを持つ通常のPythonが使われます。GILなしのビルドでは、シングルスレッドのコードがやや遅くなり、すべてのライブラリが対応しているわけではありません。

import sys
print(sys._is_gil_enabled())   # free-threadingビルドでは False

並列計算へのもう1つの道は、Python 3.14（PEP 734）で追加された、サブインタープリターです。concurrent.interpreters モジュールで、1つのプロセスの中に独立したインタープリターを複数作れます。それぞれが独自のGILを持ちます。InterpreterPoolExecutor を使うと、スレッドプールとまったく同じように書けます。講座の計測では、4つの重いタスクが、1スレッドでは1.03秒、サブインタープリターでは0.39秒でした。

まとめ

Pythonのasyncioは、待ち時間のための道具です。async def でコルーチンを書き、await で一時停止し、TaskGroupや gather で同時に起動します。スレッドは、非同期版がないライブラリでも、コードを書き直さずに待ち時間を減らせます。重い計算には、GILを回避できるプロセスが必要です。free-threadingとサブインタープリターによって状況は変わりつつありますが、「待つならasyncioかスレッド、計算するならプロセス」という基本のルールは今も有効です。講座では、この内容を実践で扱います。家計簿ツールは、為替レートを3秒ではなく1秒で並列に取得します。