Python asyncio モジュール

asyncioは Python 標準ライブラリ内のモジュールで、非同期 I/O 操作のコードを記述するために使用されます。

asyncio は、並行タスクを処理するための効率的な方法を提供し、ネットワークリクエストやファイル読み書きなどの I/O 集中型操作に特に適しています。

使用することでasyncio、シングルスレッドで複数のタスクを同時に処理でき、マルチスレッドやマルチプロセスを使用する必要はありません。

なぜ asyncio が必要なのか？

従来の同期プログラミングでは、タスクが I/O 操作（ネットワークリクエストなど）の完了を待つ必要がある場合、プログラムは操作が完了するまでブロックされます。これにより、特に大量の I/O 操作を処理する必要がある場合、プログラムの効率が低下します。

asyncio非同期プログラミングモデルを導入することで、プログラムが I/O 操作の待機中に他のタスクを実行し続けることができ、それによってプログラムの並行性と効率が向上します。

あなたがレストランを経営していると想像してください：

  • 同期モード（通常の関数）：あなたには料理人が1人しかいません。客Aがステーキを注文し、料理人はステーキを焼き始めます（これには5分かかります）。この5分間、料理人は完全に占有され、他のことは何もできません。客Bがコップ一杯の水を注文したいだけでも、ただ待つしかありません。
  • 非同期モード（asyncio）：あなたには複数の料理人がいます（実際には1人だけですが、非常に賢いです）。料理人は客Aのステーキを焼き始めた後、待つ必要があることに気づくと、すぐにそのステーキを「準備中」とマークし、振り返って客Bに水を注ぎに行きます。水を注ぎ終えて戻ったら、ステーキが焼き上がったかどうかを確認し、まだ焼けていなければ、客Cの注文を処理することもできます。こうして、I/O（ステーキを焼く、ネットワークリクエスト、ファイルの読み書きなど）を待っている時間に、料理人（CPU）はずっと効率的に働いています。

asyncio は、この賢い作業モードを実現するために Python が提供する標準ライブラリです。シングルスレッド並行処理のコードを記述でき、Web クローラー、Web サーバー、マイクロサービスなどの I/O 集中型シナリオに特に適しています。

その中核は、イベントループ、コルーチン、そしてタスクです。

asyncio の中心概念

1. コルーチン（Coroutine）

コルーチンはasyncioの中心概念の一つです。これは特別な関数であり、実行中に一時停止し、後で実行を再開できます。コルーチンはasync defキーワードで定義され、awaitキーワードで実行を一時停止し、非同期操作の完了を待ちます。

例

import asyncio

async def say_hello():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

2. イベントループ（Event Loop）

イベントループはasyncioの中核コンポーネントであり、コルーチンのスケジューリングと実行を担当します。実行すべきタスクが常にあるかどうかをチェックし、タスク完了時に対応するコールバック関数を呼び出します。

例

async def main():
    await say_hello()

asyncio.run(main())

3. タスク（Task）

タスクはコルーチンをラップしたものであり、実行中またはこれから実行されるコルーチンを表します。あなたはasyncio.create_task()関数を使用してタスクを作成し、それをイベントループに追加できます。

例

async def main():
    task = asyncio.create_task(say_hello())
    await task

4. Future

Futureは、非同期操作の結果を表すオブジェクトです。通常は低レベル API で使用され、まだ完了していない操作を表します。あなたはawaitキーワードでFuture完了を待ちます。

例

async def main():
    future = asyncio.Future()
    await future

基本的な使い方とコード例

上記の概念を理解するために、複数のURLに並行アクセスする古典的な例を見てみましょう。

3つの異なるURLのコンテンツを取得する必要があるとします。同期方式では順番に実行され、総所要時間は3回のリクエスト時間の合計になります。asyncio、これら3つのリクエストを同時に送信でき、総所要時間は最も遅いリクエストにほぼ等しくなります。

同期バージョン（比較用）

例

import time
import requests

def fetch_url(url):
    """時間のかかるネットワークリクエストをシミュレートする（同期版）"""
    print(f"开始获取: {url}")
    time.sleep(2)  # 模拟 2 秒网络延迟
    print(f"完成获取: {url}")
    return f"{url} からのデータ"

def main_sync():
    urls = ['https://example.com/1', 'https://example.com/2', 'https://example.com/3']
    results = []
    start = time.time()
   
    for url in urls:
        result = fetch_url(url)  # 必须等上一个完成才能开始下一个
        results.append(result)
   
    end = time.time()
    print(f"同步版本总耗时: {end - start:.2f} 秒")
    print(f"结果: {results}")

if __name__ == "__main__":
    main_sync()

期待される実行結果：

开始获取: https://example.com/1
完成获取: https://example.com/1
开始获取: https://example.com/2
完成获取: https://example.com/2
开始获取: https://example.com/3
完成获取: https://example.com/3
同步版本总耗时: 6.00 秒
结果: [‘来自 https://example.com/1 的数据‘, ‘来自 https://example.com/2 的数据‘, ‘来自 https://example.com/2 的数据‘]

合計で約6秒かかりました。

非同期バージョン（asyncio を使用）

私たちはaiohttpライブラリを使用して置き換え、requests非同期HTTPリクエストを実行します。まずそれをインストールします：pip install aiohttp。

例

import asyncio
import aiohttp
import time

async def fetch_url_async(session, url):
    """時間のかかるネットワークリクエストをシミュレートする（非同期版）"""
    print(f"开始异步获取: {url}")
    # 注意：ここでは aiohttp の非同期 get メソッドを使用し、await で待機します。
    async with session.get(url) as response:
        # 模拟处理响应也需要时间
        await asyncio.sleep(2)  # 使用 asyncio.sleep 模拟 I/O 等待，它不会阻塞线程
        text = await response.text()
        print(f"完成异步获取: {url}")
        return f"{url} からのデータ（長さ：{len(text)}）"

async def main_async():
    urls = ['https://httpbin.org/get', 'https://httpbin.org/delay/1', 'https://httpbin.org/headers']
   
    async with aiohttp.ClientSession() as session:  # 创建异步 HTTP 会话
        # URLごとにタスク（Task）を作成する
        tasks = []
        for url in urls:
            # create_task 会将协程加入事件循环，立即开始调度
            task = asyncio.create_task(fetch_url_async(session, url))
            tasks.append(task)
       
        print("所有任务已创建，开始并发执行...")
       
        # asyncio.gather を使用してすべてのタスクを並行実行し、それらがすべて完了するのを待ちます
        # gather は結果のリストを返します。順序は渡されたタスクの順序と一致します。
        results = await asyncio.gather(*tasks)
       
        return results

if __name__ == "__main__":
    start = time.time()
    # asyncio.run() はイベントループを起動し、トップレベルのコルーチンを実行する簡単な方法です
    final_results = asyncio.run(main_async())
    end = time.time()
   
    print(f"\n"异步版本总耗时: {end - start:.2f} 秒")
    for res in final_results:
        print(res)

期待される実行結果：

所有任务已创建，开始并发执行...
开始异步获取: https://httpbin.org/get
开始异步获取: https://httpbin.org/delay/1
开始异步获取: https://httpbin.org/headers
（大约 2 秒后，所有请求几乎同时完成）
完成异步获取: https://httpbin.org/headers
完成异步获取: https://httpbin.org/get
完成异步获取: https://httpbin.org/delay/1

异步版本总耗时: 2.10 秒  # 注意！总耗时远小于 6 秒
来自 https://httpbin.org/get 的数据 (长度: 274)
来自 https://httpbin.org/delay/1 的数据 (长度: 392)
来自 https://httpbin.org/headers 的数据 (长度: 177)

コード解説：

  • async def：コルーチン関数を定義しています。fetch_url_asyncとmain_async。
  • await：`fetch` 内fetch_url_asyncでは、私たちはawait session.get()とawait response.text()、これはイベントループに「このネットワークリクエストは時間がかかるので、他の準備ができたタスクを先に実行して」と伝えています。
  • asyncio.create_task()：`fetch(url)` をfetch_url_asyncコルーチンとしてラップし、Task、イベントループによってスケジュールされ、並行処理を実現します。
  • asyncio.gather(*tasks)：非常に実用的な関数で、渡されたすべてのコルーチン/タスクを並行実行し、それらがすべて完了するのを待って、最終的にすべての結果を収集します。
  • asyncio.run(main_async())：Python 3.7+ で推奨される方法で、イベントループの作成、コルーチンの実行、ループのクローズを担当します。

主要な関数とパラメータの説明

以下に表の形式でasyncioの中で最もよく使われる高級関数をいくつか示します：

関数 主な役割 よく使うパラメータの説明
asyncio.run(coro, *, debug=False) トップレベルのコルーチンを実行し、イベントループのライフサイクルを管理します。プログラムのメインエントリです。 coro：実行するコルーチンオブジェクト。
debug：`True` に設定するとTrueイベントループのデバッグモードを有効にできます。
asyncio.create_task(coro, *, name=None) コルーチンをラップしてTaskオブジェクトにし、イベントループに登録してスケジュールを待ちます。これが並行処理を実現する主要な方法です。 coro：ラップするコルーチンオブジェクト。
name：（Python 3.8+）タスクに名前を指定して、デバッグを容易にします。
asyncio.gather(*aws, return_exceptions=False) 並行実行複数の非同期タスク（awsコルーチン、タスクなどを受け入れます）、すべての完了を待って、結果のリストを返します。 *aws：可変長引数で、複数の非同期オブジェクトを渡します。
return_exceptions：デフォルトはFalse、任意のタスクが例外をスローすると、すぐにgatherの呼び出し元に伝播します。`True` に設定した場合、True例外は結果の一部として返されます。
asyncio.sleep(delay, result=None) 非同期的に指定秒数だけスリープします。これはtime.sleep（ブロッキング）との重要な違いです。 delayスリープする秒数。
resultスリープ終了後に返される値。
asyncio.wait(aws, *, timeout=None, return_when=ALL_COMPLETED) タスクを並行実行し、指定した条件が満たされるのを待ちます。2つのコレクションを返します(done, pending)、それぞれ完了したタスクと未完了のタスクです。 aws: 非同期オブジェクトのコレクション。
timeout: タイムアウト時間（秒）。
return_when: いつ返すかを決定します。選択肢：FIRST_COMPLETED（最初に完了したとき）、FIRST_EXCEPTION（最初の例外が発生したとき）、ALL_COMPLETED（すべて完了したとき、デフォルト）。
asyncio.to_thread(func, /, *args, **kwargs) （Python 3.9+）通常の、ブロックする可能性のある同期関数を別のスレッドで実行し、await可能なawaitコルーチンを返します。CPU 密集型またはブロッキング I/O を処理するために使用します。 func: スレッドで実行する同期関数。
*args, **kwargs: 関数に渡す引数。

視覚的に理解する：非同期タスクのスケジューリングの流れ

図解説明： このフローチャートは、イベントループがどのようにスケジューラのように機能するかを示しています。イベントループはタスクキューを維持し、あるタスクがawait（例：ネットワーク応答の待機）に達すると、そのタスクは一時停止され、イベントループはすぐにキューから次に実行可能（準備完了）なタスクを見つけて実行します。一時停止中のタスクの I/O 操作が完了すると、イベントループは通知を受け取り、そのタスクの状態を準備完了に戻し、将来のある時点で実行を継続します。このようにして、I/O 待機中に CPU を他のタスクの実行に最大限活用し、シングルスレッド内での並行処理を実現します。

asyncio の基本的な使い方

1. コルーチンを実行する

コルーチンを実行するには、asyncio.run()関数を使用できます。この関数はイベントループを作成し、指定されたコルーチンを実行します。

例

import asyncio

async def main():
    print("Start")
    await asyncio.sleep(1)
    print("End")

asyncio.run(main())

2. 複数のタスクを並行実行する

使用できるのはasyncio.gather()関数で複数のコルーチンを並行実行し、それらがすべて完了するのを待つことです。

例

import asyncio

async def task1():
    print("Task 1 started")
    await asyncio.sleep(1)
    print("Task 1 finished")

async def task2():
    print("Task 2 started")
    await asyncio.sleep(2)
    print("Task 2 finished")

async def main():
    await asyncio.gather(task1(), task2())

asyncio.run(main())

3. タイムアウト制御

使用できるのはasyncio.wait_for()関数でコルーチンにタイムアウト時間を設定する方法です。指定した時間内にコルーチンが完了しない場合、asyncio.TimeoutError例外が発生します。

例

import asyncio

async def long_task():
    await asyncio.sleep(10)
    print("Task finished")

async def main():
    try:
        await asyncio.wait_for(long_task(), timeout=5)
    except asyncio.TimeoutError:
        print("Task timed out")

asyncio.run(main())

asyncio の適用シーン

asyncio特に以下のようなシーンに適しています：

  1. ネットワークリクエスト：HTTP リクエスト、WebSocket 通信など。
  2. ファイル I/O：ファイルの非同期読み書きなど。
  3. データベース操作：データベースへの非同期アクセスなど。
  4. リアルタイムデータ処理：リアルタイムメッセージキューの処理など。

よく使うクラス、メソッド、関数

1. コア関数

メソッド/関数 説明 例
asyncio.run(coro) 非同期メイン関数を実行（Python 3.7+） asyncio.run(main())
asyncio.create_task(coro) タスクを作成してイベントループに追加 task = asyncio.create_task(fetch_data())
asyncio.gather(*coros) 複数のコルーチンを並行実行 await asyncio.gather(task1, task2)
asyncio.sleep(delay) 非同期待機（非ブロッキング） await asyncio.sleep(1)
asyncio.wait(coros) タスクの完了方法を制御 done, pending = await asyncio.wait([task1, task2])

2. イベントループ（Event Loop）

メソッド 説明 例
loop.run_until_complete(future) タスクが完了するまで実行 loop.run_until_complete(main())
loop.run_forever() イベントループを永久に実行 loop.run_forever()
loop.stop() イベントループを停止 loop.stop()
loop.close() イベントループを閉じる loop.close()
loop.call_soon(callback) コールバック関数の即時実行を予約 loop.call_soon(print, "Hello")
loop.call_later(delay, callback) コールバックを遅延実行 loop.call_later(5, callback)

3. コルーチン（Coroutine）とタスク（Task）

メソッド/デコレータ 説明 例
@asyncio.coroutine コルーチンデコレータ（旧版、Python 3.4-3.7） @asyncio.coroutine
def old_coro():
async def コルーチンの定義（Python 3.5+） async def fetch():
task.cancel() タスクをキャンセル task.cancel()
task.done() タスクが完了したかどうかを確認 if task.done():
task.result() タスクの結果を取得（タスクの完了が必要） data = task.result()

4. 同期プリミティブ（類似：threading）

类 説明 例
asyncio.Lock() 非同期ミューテックスロック lock = asyncio.Lock()
async with lock:
asyncio.Event() イベント通知 event = asyncio.Event()
await event.wait()
asyncio.Queue() 非同期キュー queue = asyncio.Queue()
await queue.put(item)
asyncio.Semaphore() セマフォ sem = asyncio.Semaphore(5)
async with sem:

5. ネットワークとサブプロセス

メソッド/クラス 説明 例
asyncio.open_connection() TCP接続を確立 reader, writer = await asyncio.open_connection('host', 80)
asyncio.start_server() TCPサーバーを作成 server = await asyncio.start_server(handle, '0.0.0.0', 8888)
asyncio.create_subprocess_exec() サブプロセスを作成 proc = await asyncio.create_subprocess_exec('ls')

6. ユーティリティ

メソッド 説明 例
asyncio.current_task() 現在のタスクを取得 task = asyncio.current_task()
asyncio.all_tasks() すべてのタスクを取得 tasks = asyncio.all_tasks()
asyncio.shield(coro) タスクをキャンセルから保護 await asyncio.shield(critical_task)
asyncio.wait_for(coro, timeout) タイムアウト付き待機 try: await asyncio.wait_for(task, 5)

例

基本的なコルーチンの例

例

import asyncio

async def hello():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

asyncio.run(hello())  # Python 3.7+

2. タスクの並行実行

例

async def fetch(url):
    print(f"Fetching {url}")
    await asyncio.sleep(2)
    return f"Data from {url}"

async def main():
    results = await asyncio.gather(
        fetch("url1.com"),
        fetch("url2.com")
    )
    print(results)

asyncio.run(main())

3. 非同期キューの使用

例

async def producer(queue):
    for i in range(5):
        await queue.put(i)
        await asyncio.sleep(0.1)

async def consumer(queue):
    while True:
        item = await queue.get()
        print(f"Consumed {item}")
        queue.task_done()

async def main():
    queue = asyncio.Queue()
    await asyncio.gather(
        producer(queue),
        consumer(queue)
    )

注意事項

  1. Python バージョン：一部の機能は Python 3.7+ が必要です（例：asyncio.run()）。

  2. ブロッキング操作：コルーチン内で同期ブロッキングコードを使用しないでください（例：time.sleep()）。

  3. デバッグ：設定PYTHONASYNCIODEBUG=1環境変数でデバッグモードを有効にできます。

  4. タスクのキャンセル：キャンセルされたタスクはCancelledErrorを発生させるため、適切に処理する必要があります。

その他の拡張