Python非同期プログラミング実践ガイド：asyncioでI/O待ちを短縮する実装手法

お疲れ様です！IT業界で働くアライグマです！

「PythonでI/O待ちが多くて処理が遅い」「複数のAPIを並行で呼び出したいが、どう実装すればいいか分からない」

Python非同期プログラミングは、I/O待ち時間を有効活用し、全体の処理時間を短くする技術です。
どれだけ速くなるかは、待ち時間の長さや同時に進める数で変わるので、自分の処理で計測して確かめるのが確実です。

本記事では、Python非同期プログラミングの基礎から、HTTP通信・データベース・最適化の実装例まで、コードとともに解説します。

Python非同期プログラミングの基本概念

非同期プログラミングは、I/O待ち時間を有効活用する並行処理の手法です。
ここでは、非同期プログラミングの基本概念を解説します。

非同期処理が必要な理由

同期処理では、I/O操作の完了を待つ間、CPUが遊休状態になります。
非同期処理により、待ち時間中に他のタスクを実行でき、全体の処理時間を短縮できます。

たとえば、1回に3秒かかるAPIを10回呼ぶ場合、順番に呼ぶと待ち時間だけで約30秒かかります。
10回を同時に投げて待てば、相手のサーバーが同時に応えられる限り、待ち時間は1回分に近づきます。

asyncioの基本構造

asyncioは、Pythonの標準ライブラリで提供される非同期I/Oフレームワークです。
async/await構文を使用して、非同期関数を定義し、イベントループで実行します。

イベントループは1つのスレッドで動き、awaitで待ちに入ったタスクから、準備のできた別のタスクへ処理を切り替えます。
そのため、待ち時間の長い処理（I/Oバウンド）には効きますが、計算し続ける処理（CPUバウンド）は速くなりません。

コルーチンとタスクの理解

コルーチンは、async defで定義される非同期関数です。
呼び出しただけでは動かず、awaitするか、asyncio.create_taskでタスクにしてイベントループに渡すと実行されます。

import asyncio

async def work(name, seconds):
    await asyncio.sleep(seconds)  # I/O待ちの代わり
    return f"{name} done"

async def main():
    task_a = asyncio.create_task(work("A", 2))
    task_b = asyncio.create_task(work("B", 2))
    print(await task_a, await task_b)  # 2つ合わせて約2秒で終わる

asyncio.run(main())

IT女子 アラ美

asyncioにすれば、どんな処理でも速くなるものだと思っていたわ。

ITアライグマ

速くなるのは待ち時間の多い処理です。計算が重い処理はマルチプロセスを使います。

asyncioの基本的な使い方

asyncioの基本的な使い方を理解することが、非同期プログラミングの第一歩です。
ここでは、実践的な使用方法を解説します。

async/await構文の基礎

async/awaitは、非同期処理を記述するための構文です。
async defで非同期関数を定義し、awaitで非同期操作の完了を待ちます。

複数のAPIを並行で呼び出す基本形は、次のようになります。

import asyncio
import aiohttp

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

async def main():
    urls = ['https://example.com/a', 'https://example.com/b']
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_data(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
    return results

asyncio.run(main())

ClientSessionは呼び出しのたびに作らず、1つを作って使い回すのがポイントです（理由は次の章で説明します）。

asyncio.gatherによる並行実行

asyncio.gatherは、複数のコルーチンを並行実行する関数です。
すべてのタスクが完了するまで待ち、結果を渡した順のリストで返します。

既定では、どれか1つが例外を出すと、その例外がそのままgatherから送られてきます。
1つが失敗してもほかの結果を受け取りたいときは、return_exceptions=Trueを付けると、例外も結果のリストに入ります。

タイムアウトとキャンセル処理

タイムアウトは、長時間実行されるタスクを制限する重要な機能です。
asyncio.wait_forを使用して、タイムアウト時間を設定できます。Python 3.11以降ではasync with asyncio.timeout(秒数):の書き方も使えます。

try:
    text = await asyncio.wait_for(fetch_data(session, url), timeout=5)
except TimeoutError:  # Python 3.11以降。3.10以前は asyncio.TimeoutError
    text = None

時間切れになると、待っていたタスクはキャンセルされます。
応答が遅い外部APIがあっても、そこで全体が止まらないようにしておけます。

IT女子 アラ美

1つのAPIが返ってこなくて、処理全体が止まってしまったのよ。

ITアライグマ

外部APIの呼び出しには、wait_forなどで必ずタイムアウトを付けておきましょう。

aiohttpによる非同期HTTP通信

aiohttpは、asyncioベースの非同期HTTPクライアント・サーバーライブラリです。
ここでは、aiohttpを活用した実践的な実装を解説します。

ClientSessionの使い方

ClientSessionは、HTTP接続を管理するクラスです。
セッションの中で接続を使い回す（コネクションプーリング）ので、同じサーバーへの2回目以降のリクエストでは、接続を張り直す手間を省けます。
aiohttpの公式ドキュメントでも、リクエストごとにセッションを作らず、アプリケーション全体で1つのセッションを使うよう勧めています。

並行リクエストの制御

並行リクエスト数を制御することで、サーバーへの負荷を調整できます。
asyncio.Semaphoreを使用して、同時実行数を制限します。

sem = asyncio.Semaphore(10)  # 同時に10件まで

async def fetch_limited(session, url):
    async with sem:
        return await fetch_data(session, url)

1000件のURLを一度に投げると、相手のサーバーに負荷をかけたり、アクセスを拒否されたりします。
同時に何件まで送ってよいかは、相手のAPIの利用規約やレート制限に合わせて決めます。

エラーハンドリングとリトライ

エラーハンドリングは、非同期処理の信頼性を高めます。
try-except構文とリトライロジックを組み合わせることで、一時的なエラーに対応できます。

async def fetch_with_retry(session, url, retries=3):
    for attempt in range(retries):
        try:
            return await fetch_data(session, url)
        except aiohttp.ClientError:
            if attempt == retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)  # 1秒、2秒…と待ち時間を倍にする

失敗するたびに待ち時間を倍にする（指数バックオフ）と、一時的に混み合ったサーバーに続けて負荷をかけずに済みます。

IT女子 アラ美

並行にできるなら、1000件まとめて一気に投げたほうが早いでしょう？

ITアライグマ

相手のサーバーに負荷がかかります。Semaphoreで同時の数を絞りましょう。

データベース操作の非同期化

データベース操作の非同期化は、アプリケーションのパフォーマンスを大きく向上させます。
ここでは、非同期データベースライブラリの活用方法を解説します。

asyncpgによるPostgreSQL接続

asyncpgは、PostgreSQL用の非同期ドライバです。
asyncio向けに作られていて、通信部分はCython（C拡張）で実装されています。

import asyncpg

async def main():
    pool = await asyncpg.create_pool(dsn="postgresql://user:pass@localhost/db")
    async with pool.acquire() as conn:
        rows = await conn.fetch("SELECT id, name FROM users WHERE active = $1", True)
    await pool.close()

プレースホルダーは$1、$2の形で書きます。

aiomysqlによるMySQL接続

aiomysqlは、MySQL用の非同期ドライバです。
PyMySQLをもとに作られていて、APIも似ているため、PyMySQLを使ったコードから移りやすいのが特徴です。

トランザクション管理

トランザクションは、データの整合性を保証する重要な機能です。
非同期環境でも、適切なトランザクション管理が必要です。

async with pool.acquire() as conn:
    async with conn.transaction():  # 例外が出ると自動でロールバック
        await conn.execute("UPDATE accounts SET balance = balance - 100 WHERE id = $1", 1)
        await conn.execute("UPDATE accounts SET balance = balance + 100 WHERE id = $1", 2)

注意したいのは、1つの接続で複数のクエリを同時に投げられないことです。
クエリを並行で実行したいときは、プールから接続をクエリごとに借りて使います。

IT女子 アラ美

1つの接続でクエリをgatherしたら、エラーになってしまったわ。

ITアライグマ

1つの接続は同時に1クエリまでです。並行にするならプールから借りましょう。

パフォーマンス最適化テクニック

パフォーマンス最適化は、非同期プログラミングの効果を最大化します。
ここでは、実践的な最適化テクニックを解説します。

コネクションプーリング

コネクションプーリングは、接続の再利用により、オーバーヘッドを削減します。
データベースやHTTP接続で特に効果的です。
前の章のaiohttpのClientSessionやasyncpgのcreate_poolのように、アプリケーションの起動時にプールを作り、終了まで使い回します。

バッチ処理の活用

バッチ処理は、複数の操作をまとめて実行することで、効率を高めます。
データベースの一括挿入や更新で効果を発揮します。

asyncpgなら、executemanyで同じ形のINSERTをまとめて送れます。大量のデータを入れるなら、PostgreSQLのCOPYを使うcopy_records_to_tableも選べます。
1件ずつawaitして送ると、そのたびにデータベースとの往復の待ちが入るので、まとめるほど待ちの回数を減らせます。

メモリ使用量の最適化

メモリ使用量は、大量のタスクを扱う際に重要です。
ジェネレータやストリーミング処理を活用することで、メモリを節約できます。

たとえば10万件のURLに対して一度にタスクを作ると、タスクと結果がすべてメモリに載ります。
数百件ずつに区切って処理するか、asyncio.Queueに入れて決まった数の作業役（ワーカー）で順に取り出すと、メモリの使用量を一定に保てます。
大きなファイルのダウンロードも、response.content.iter_chunked()で少しずつ読み、そのままファイルに書き込みます。

IT女子 アラ美

タスクを10万個まとめて作ったら、メモリがあふれてしまったのよね。

ITアライグマ

件数を区切るか、Queueと決まった数のワーカーで処理すると安定します。

実践的なユースケースと導入事例

ユースケースを理解することで、非同期プログラミングの適用範囲が明確になります。
ここでは、非同期化が向いている代表的な場面を紹介します。

Webスクレイピングの高速化

Webスクレイピングは、非同期プログラミングの代表的なユースケースです。
ページの取得はほとんどが通信の待ち時間なので、複数のページを並行取得すると、全体の時間を短くできます。

ただし、相手のサイトの利用規約とrobots.txtを守り、Semaphoreで同時の数を絞って、サーバーに負担をかけないようにします。

マイクロサービス間通信

マイクロサービスでは、1つの画面を返すために複数のサービスへ問い合わせることがあります。
お互いに依存しない問い合わせなら、asyncio.gatherで同時に投げると、応答時間を「全部の合計」から「いちばん遅いサービスの時間」に近づけられます。

リアルタイムデータ処理

リアルタイムデータ処理は、ストリーミングデータを継続的に処理します。
WebSocketやメッセージキューから届くデータを、受け取る役と処理する役に分けてasyncio.Queueでつなぐと、受信が処理に待たされずに済みます。

どのくらいの件数を処理できるかは、1件の処理の重さとマシンの性能で変わるので、実際のデータ量で負荷をかけて確かめます。
処理にCPUを多く使う場合は、その部分だけをマルチプロセスに分けることも検討します。

IT女子 アラ美

3つのサービスを順番に呼んでいたけど、同時に呼んでもよかったのね。

ITアライグマ

お互いの結果を使わないならgatherで同時に呼べます。依存があれば順番です。

よくある質問

asyncioとthreadingはどう使い分ければよいですか？

どちらもI/O待ちの多い処理に向いています。
使いたいライブラリがasyncioに対応しているならasyncio、requestsのように同期のライブラリしかない場合はthreading（concurrent.futures.ThreadPoolExecutorなど）が手軽です。

asyncioでCPUを使う重い計算は速くなりますか？

速くなりません。asyncioは1つのスレッドで待ち時間を切り替える仕組みなので、計算し続ける処理には効きません。
重い計算は、ProcessPoolExecutorなどで別のプロセスに分けます。

非同期の関数の中でrequestsを使ってもよいですか？

使うと、通信を待つ間にイベントループ全体が止まり、ほかのタスクも進まなくなります。
aiohttpやhttpxの非同期クライアントに置き換えるか、どうしても使うならasyncio.to_threadで別スレッドに逃がします。

IT女子 アラ美

async defの中でrequestsを使っていたら、全然並行にならなかったわ。

ITアライグマ

同期の通信はループを止めます。aiohttpに替えるかto_threadで逃がしましょう。

まとめ

Python非同期プログラミング実践ガイドについて、asyncioで処理速度を向上させる実装手法を解説しました。

Python非同期プログラミングの基本概念では、非同期処理が必要な理由、asyncioの基本構造、コルーチンとタスクの理解が重要です。
asyncioの基本的な使い方では、async/await構文の基礎、asyncio.gatherによる並行実行、タイムアウトとキャンセル処理が効果的です。
aiohttpによる非同期HTTP通信では、ClientSessionの使い方、並行リクエストの制御、エラーハンドリングとリトライが重要です。

データベース操作の非同期化では、asyncpgによるPostgreSQL接続、aiomysqlによるMySQL接続、トランザクション管理が役立ちます。
パフォーマンス最適化テクニックでは、コネクションプーリング、バッチ処理の活用、メモリ使用量の最適化が効果的です。
実践的なユースケースと導入事例では、Webスクレイピングの高速化、マイクロサービス間通信、リアルタイムデータ処理が開発効率を向上させます。

Python非同期プログラミングを適切に活用することで、I/O待ち時間を有効活用し、全体の処理時間を短くできます。
まずは、今ある処理のうち「APIやDBの応答を待っている時間」がどれくらいあるかを測るところから始めてみてください。

IT女子 アラ美

まずは今の処理で、待っている時間がどれくらいあるのかを測ってみるわね。

ITアライグマ

はい。待ちが多い処理から非同期にすると、効果を確かめやすいですよ。