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

URL: https://skillaria.top/ja/guides/python/asyncio-python

asyncioはPythonの標準ライブラリで、async def で定義した関数（コルーチン）は await で一時停止でき、その間にイベントループが別の処理を実行します。そのため、1つのスレッドで多数のネットワーク要求を同時に進められます。ネットワークやディスクの待ち時間にはasyncioかスレッドを、重い計算にはプロセスを使います。通常のCPythonでは、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の選び方

| 課題 | 道具 | 理由 |
| --- | --- | --- |

| 200ページのダウンロード、多数のAPIリクエスト | asyncio（httpx/aiohttp）またはThreadPoolExecutor | ネットワーク待ち、CPUは空いている |

| 非同期プログラムの中で複数のリクエスト | TaskGroup | コードはすでにasyncで書かれている |

| 非同期版がない同期ライブラリ | スレッド、`asyncio.to_thread` | コードを書き直さなくてよい |

| 純粋なPythonによる重い計算 | ProcessPoolExecutor | GILを回避し、すべてのコアを使える |

## 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秒で並列に取得します。

## FAQ

**通常のコードから async 関数を呼び出すにはどうすればよいですか？**

asyncio.run(coro()) を使います。イベントループを作り、コルーチンを実行して、ループを閉じます。すでに動いているループの中で asyncio.run を呼ぶことはできません。その場合は await を使います。

**asyncio は複数のスレッドで動きますか？**

いいえ。イベントループはすべてのコルーチンを1つのスレッドで実行し、await の地点でだけ切り替えます。そのためスレッドよりデータ競合は少なくなりますが、1つのブロッキング処理が全体を止めてしまいます。

**asyncio を pip でインストールする必要がありますか？**

いいえ。asyncio はPython 3の標準ライブラリに含まれています。import asyncio と書くだけで使えます。PyPIにある asyncio パッケージは古いもので、インストールする必要はありません。

**asyncio の中から重い関数を呼び出すには、イベントループを止めずにどうすればよいですか？**

計算には loop.run_in_executor と ProcessPoolExecutor を、ブロッキングな入出力には asyncio.to_thread を使います。イベントループは、ほかのタスクの処理を続けられます。

**コルーチンとタスク（Task）の違いは何ですか？**

コルーチンは、await されるまで何もしないオブジェクトです。タスクはコルーチンをラップし、すぐにイベントループのスケジュールに入れるため、ほかのコードと並行して実行されます。


---
Source: Skillaria.Top (https://skillaria.top)