Python asyncio.gatherとTaskGroup (3.11+) の例外ハンドリング比較

# Python 3.11+ 公式仕様：TaskGroup と ExceptionGroup ハンドリング
import asyncio

async function fetch_service(name: str, delay: int):
    await asyncio.sleep(delay)
    if name == "bad":
        raise ValueError("サービス通信エラー")
    return f"{name} データ"

async function main():
    try:
        # ⭕ TaskGroup による構造化並行処理
        async with asyncio.TaskGroup() as tg:
            task1 = tg.create_task(fetch_service("good", 1))
            task2 = tg.create_task(fetch_service("bad", 2))
            task3 = tg.create_task(fetch_service("slow", 5)) # task2 失敗時に自動キャンセルされる
            
        print(task1.result())
    except* ValueError as eg: # Python 3.11+ except* 構文
        print(f"補獲された例外グループ: {eg.exceptions}")

asyncio.run(main())

Python 3.11+ 公式仕様：TaskGroup と gather の比較マトリクス

Python公式ドキュメント（docs.python.org/3/library/asyncio-task.html）における並行タスク管理の仕様比較です。

評価軸	asyncio.gather() (従来手法)	asyncio.TaskGroup() (Python 3.11+)
構文スタイル	関数呼び出し (await gather(...))	コンテキストマネージャー (async with)
失敗時の他タスク処理	❌ 他タスクは停止せず裏で動き続ける	◯ 他の全タスクを即座に自動キャンセル
例外ハンドリング	最初の例外1つのみ送出（他は失われる）	ExceptionGroup で発生した全例外を保持
タスクリーク防止	手動でキャンセルコードの記述が必要	構造化（Scope）により全タスクの終了を保証

---

実際に起こる障害：asyncio.gather() による「ゾンビタスク放置」事故

従来の asyncio.gather() では、並列実行しているタスクAが例外を出して失敗した場合、gather 自体は即座に例外をスローしますが、併行して動いているタスクBやタスクCの実行はキャンセルされず、バックグラウンドで永久に動き続ける「ゾンビタスク障害」 が発生します。

これにより、失敗したにもかかわらず外部APIやDBへのムダなリクエストが送信され続け、エラーの根本原因が埋もれてログ解析が困難になります。

---

移行手順

1. Python 3.11 以上の環境で async with asyncio.TaskGroup() as tg: 構文に書き換える
2. tg.create_task() で各非同期関数を登録する
3. 例外ハンドリング側では try ... except* ValueError:（except* 構文）を使い、複数の例外を漏れなくキャッチする