Python非同期処理でエラーが頻発する理由｜asyncioとaiohttpの実践的解決策

非同期処理は性能向上の切り札のはずなのに、本番環境で予期しないエラーが多発する。そんな悩みを抱えるエンジニアは多いのではないでしょうか。

特にPythonのasyncioとaiohttpを使った実装では、同期処理とは異なる落とし穴が数多く存在します。タイムアウト設定の誤り、接続プールの管理ミス、例外処理の不完全さ…こうした問題は運用段階で初めて顕在化することがほとんどです。

本記事では、実務レベルでよく遭遇するPython非同期処理のエラーを8つのパターンに分類し、各々の原因と実践的な解決策をコード例とともに解説します。

asyncioとaiohttpで非同期処理が複雑な理由

非同期処理が難しいのは、実行順序が予測不可能だからです。同期処理なら1行目→2行目→3行目と明確ですが、asyncioではイベントループが複数のタスクを交互に実行するため、デバッグが格段に難しくなります。

さらにaiohttpはHTTP通信という不確定要素を含むため、ネットワークエラーやサーバー応答遅延に対応する必要があります。これらの要因が組み合わさると、予想外のエラーが連鎖的に発生するわけです。

パターン1：RuntimeErrorが発生する理由と対処法

最も多く報告される問題が「RuntimeError: Event loop is closed」です。これはイベントループの管理ミスが原因となります。

具体的には、すでにイベントループが閉じられた状態で新たに非同期関数を実行しようとするときに発生します。以下は誤った実装例です。

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

# 間違った実装
for i in range(3):
    asyncio.run(fetch(session, f"https://api.example.com/data/{i}"))

問題はループのたびにイベントループが作成・閉鎖されることです。正しくは1つのイベントループで複数のタスクを並行実行すべきです。

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, f"https://api.example.com/data/{i}") for i in range(3)]
        results = await asyncio.gather(*tasks)
        return results

asyncio.run(main())

このパターンなら、イベントループはmain関数の実行期間中だけ存在し、完了後に自動的に閉鎖されます。

パターン2：タイムアウト設定の誤りによるハング問題

aiohttpで通信が無限にハングする症状は、タイムアウト値を明示的に設定していないことがほとんどです。デフォルトではタイムアウトが無制限に近い値となっています。

# 誤った実装：タイムアウトなし
async def fetch_without_timeout(session, url):
    async with session.get(url) as response:
        return await response.text()

# 正しい実装：明示的なタイムアウト設定
async def fetch_with_timeout(session, url):
    timeout = aiohttp.ClientTimeout(total=10, connect=5, sock_read=5)
    async with session.get(url, timeout=timeout) as response:
        return await response.text()

timeout値の各要素は以下の意味を持ちます。totalは全体の処理時間、connectは接続確立時間、sock_readはデータ受信時間です。

実装のベストプラクティスとしては、ClientSessionの作成時にグローバルなタイムアウトを設定し、必要に応じて個別のリクエストで上書きする手法をお勧めします。

async def main():
    timeout = aiohttp.ClientTimeout(total=30)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        # セッション全体でタイムアウト適用
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results

asyncio.run(main())

パターン3：接続プール枯渇によるエラー

複数の並行リクエストを実行する際、接続プール上限に達するとエラーが発生します。aiohttpのClientSessionはデフォルトで最大100個の同時接続しか許容しません。

1000個のエンドポイントに同時アクセスしようとすると、ほぼ確実にエラーが発生するわけです。これを解決するには、コネクタの設定でプール上限を引き上げる必要があります。

import asyncio
import aiohttp

async def main():
    # 接続プール上限を500に設定
    connector = aiohttp.TCPConnector(limit=500, limit_per_host=50)
    timeout = aiohttp.ClientTimeout(total=30)
    
    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results

asyncio.run(main())

ただしlimit_per_hostの設定にも注意が必要です。同一ホストへのリクエスト数を制限しないと、サーバーがDoS攻撃と判定してIPアドレスをブロックする可能性があります。

パターン4：例外処理の不完全さがもたらす連鎖エラー

非同期処理では、1つのタスク内で例外が発生すると、他の全タスクに波及する危険性があります。これを防ぐにはreturn_exceptions=Trueを活用すべきです。

# 危険な実装：1つのエラーが全体に波及
results = await asyncio.gather(*tasks)  # 1つが失敗すると全て中断

# 安全な実装：エラーを結果として返す
results = await asyncio.gather(*tasks, return_exceptions=True)
# results = [成功1, エラーオブジェクト, 成功2, ...]

# さらに安全：エラー処理を明示的に記述
for i, result in enumerate(results):
    if isinstance(result, Exception):
        print(f"Task {i} failed: {result}")
    else:
        print(f"Task {i} succeeded: {result}")

加えて、各タスク内でも個別に例外をキャッチすることを推奨します。これにより、特定のエラーに対する復旧処理を個別に実装できます。

パターン5：セッション共有の落とし穴

aiohttpのClientSessionはスレッドセーフではない点に注意が必要です。複数のスレッドから同じセッションにアクセスするとレースコンディションが発生します。

# 誤った実装：グローバルセッションの共有
session = None

async def init():
    global session
    session = aiohttp.ClientSession()

async def fetch(url):
    # 複数のイベントループから呼ばれると危険
    async with session.get(url) as response:
        return await response.text()

# 正しい実装：セッションをコンテキストマネージャで管理
async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_internal(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results

async def fetch_internal(session, url):
    async with session.get(url) as response:
        return await response.text()

asyncio.run(main())

パターン6：メモリリークと接続リークの予防

適切にセッションやコネクションをクローズしないと、メモリリークが発生します。大規模なデータ処理では、これが深刻な問題となります。

# 危険な実装：セッションを開きっぱなし
async def process_large_dataset(urls):
    session = aiohttp.ClientSession()
    for url in urls:
        async with session.get(url) as response:
            data = await response.text()
            # メモリに蓄積される可能性がある

# 安全な実装：バッチ処理とクローズを明示的に記述
async def process_large_dataset(urls, batch_size=100):
    async with aiohttp.ClientSession() as session:
        for i in range(0, len(urls), batch_size):
            batch = urls[i:i+batch_size]
            tasks = [fetch(session, url) for url in batch]
            results = await asyncio.gather(*tasks, return_exceptions=True)
            
            # 処理結果をDBに保存するなど、メモリから解放
            await save_results(results)
            # バッチ終了後、メモリをクリア
            del results

パターン7：デコーダエラーと文字コード問題

レスポンスの文字コード判定が自動で行われますが、サーバー側の設定が不正だとデコーダエラーが発生します。

# 誤った実装：文字コード指定なし
async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()  # 自動判定に依存

# 正しい実装：明示的に文字コード指定
async def fetch(session, url):
    async with session.get(url) as response:
        # サーバー指定の文字コード優先、なければUTF-8を使用
        return await response.text(errors='replace')  # エラーを無視する方法
        
        # または
        # charset = response.charset or 'utf-8'
        # text = await response.text(errors='ignore')

パターン8：デバッグの困難さを乗り越える方法

非同期処理のデバッグは同期処理より難しいため、ログ出力の徹底が重要です。asyncioデバッグモードを有効にすることで、問題の原因を特定しやすくなります。

import asyncio
import logging

# ログレベルを設定
logging.basicConfig(level=logging.DEBUG)

async def main():
    # asyncioのデバッグモード有効化
    asyncio.get_event_loop().set_debug(True)
    
    tasks = [task1(), task2(), task3()]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

asyncio.run(main())

加えて、タスク進捗をタイムスタンプ付きで記録することで、実行順序の予測が容易になります。

おすすめ書籍・ガジェット

• Fluent Python – Pythonの非同期処理・非同期イテレータを深く理解するなら必読の名著です。
• Real Python asyncioチュートリアル – Real Pythonの実践的なオンライン教材で、asyncioの落とし穴を段階的に学べます。
• HHKB Professional – デバッグ作業が増えるエンジニア向けに、長時間の開発作業に最適なメカニカルキーボードです。

実装時の検査リスト

本記事の内容を踏まえ、非同期処理実装時に確認すべきポイントをリスト化しました。

• イベントループは1つだけか、asyncio.runで管理されているか。
• すべてのリクエストにタイムアウト値が明示的に設定されているか。
• 接続プール上限値はアクセス数に適切か、limit_per_hostも設定されているか。
• 例外処理にreturn_exceptions=Trueが使用されているか。
• セッションはasync withで適切にクローズされているか。
• メモリリーク対策として、バッチ処理やメモリ解放処理が実装されているか。
• 文字コード問題に対応するため、errorsパラメータが設定されているか。
• デバッグモードとログ出力が本番環境では無効化されているか。

他の非同期処理技術との比較

Python以外の非同期処理技術も存在します。以下は代表的な選択肢の比較です。

Pythonの非同期処理は学習難易度とパフォーマンスのバランスが取れており、データ分析や機械学習との親和性が高い点が強みです。特にAIエンジニアを目指す場合、30代未経験からAIエンジニアへの転職は現実的かという記事でも指摘される通り、Pythonスキルは必須要件となっています。

開発効率を高めるAIツール活用法

asyncioのコード生成やデバッグ支援には、AIツールが有効です。Claude codeを使った自動化の極意の記事で詳述されている通り、複雑な非同期ロジックの自動生成はAIの得意分野です。

また、コードの規約統一にはCursor Rules書き方・チームで共有する方法が参考になります。非同期処理のベストプラクティスをルール化し、チーム全体で共有することで、エラーの事前防止が可能になります。

まとめ

Python非同期処理でのエラーは、その原因が複雑で予測が難しいものがほとんどです。しかし、イベントループ管理、タイムアウト設定、接続プール管理、例外処理、セッション管理といった8つのパターンを理解すれば、大半の問題を事前に防ぐことができます。

本記事で紹介したコード例を参考に、自分のプロジェクトに適用してみてください。特に本番環境へのデプロイ前に、実装検査リストをすべて確認することを強く推奨します。

非同期処理は強力ですが、正しく使わなければかえってバグの温床となります。この記事が皆さんの開発効率向上の一助となれば幸いです。