【Python】async/awaitで非同期処理を実装する基本

2026年1月14日

概要

Pythonの asyncio モジュールを使用した「非同期処理（Asynchronous Processing）」の基本的な書き方を解説します。

従来の同期処理（time.sleep 等）は処理が終わるまでその場でプログラム全体が止まってしまいますが、非同期処理（await asyncio.sleep 等）を使うと、待ち時間の間に別の処理を効率よく進めることが可能になります。

仕様（入出力）

入力: 非同期関数（コルーチン）の定義と呼び出し
出力: 非同期に実行されたタスクの結果
構文:
- async def: 非同期関数（コルーチン）を定義する。
- await: 非同期処理の結果を待つ（処理完了まで制御を返す）。
- asyncio.run(): 非同期処理のエントリーポイントとして実行を開始する。

基本の使い方

関数の前に async を付け、中で待機する処理の前に await を付けます。そして実行には asyncio.run() を使います。

1. 同期処理（従来の書き方）

処理が完了するまでブロック（停止）します。

import time

def fetch_data_sync():
    print("データ取得開始（同期）...")
    time.sleep(3) # ここで完全に止まる
    print("データ取得完了")
    return "完了"

# 実行
print(fetch_data_sync())

2. 非同期処理（async/await）

待ち時間の間に制御をイベントループに返すため、効率的です。

import asyncio

async def fetch_data_async():
    print("データ取得開始（非同期）...")
    await asyncio.sleep(3) # ノンブロッキング待機
    print("データ取得完了")
    return "完了"

# 実行
if __name__ == "__main__":
    result = asyncio.run(fetch_data_async())
    print(result)

コード全文

非同期関数を定義し、それをメイン関数から呼び出して実行する完全なサンプルコードです。

import asyncio
import time

# --- 非同期関数の定義 (async def) ---
async def fetch_data():
    """データを取得するシミュレーションを行うコルーチン"""
    print("[Task] データ取得リクエスト送信...")
    
    # 非同期スリープ (await asyncio.sleep)
    # time.sleep(3) を使うと非同期の意味がなくなるので注意
    await asyncio.sleep(3)
    
    print("[Task] データ受信完了")
    return "Sample Data"

# --- メインコルーチンの定義 ---
async def main():
    print("--- 処理開始 ---")
    
    # await でコルーチンの完了を待機し、戻り値を受け取る
    # この行で3秒間待ちますが、スレッドをブロックしているわけではありません
    result = await fetch_data()
    
    print(f"取得結果: {result}")
    print("--- 処理終了 ---")

# --- 実行 ---
if __name__ == "__main__":
    # asyncio.run() でイベントループを開始し、main()を実行
    asyncio.run(main())

カスタムポイント

構文と意味のまとめ

構文	意味・役割	注意点
`async def 関数名():`	コルーチン関数の定義。この関数を呼ぶと、実行結果ではなく「コルーチンオブジェクト」が返されます。	通常の関数呼び出しのように `func()` と書いても中身は実行されません。
`await コルーチン`	実行待ち。指定した非同期処理が完了するまで待ち、結果を受け取ります。	`async def` の中でしか使えません。
`asyncio.run(コルーチン)`	実行開始。イベントループを作成し、最上位のコルーチン（mainなど）を実行します。	プログラムのエントリーポイント（開始地点）で1回だけ呼びます。

注意点

呼び出し方の違い
- async def で定義した関数は、単に func() と書いても実行されません。「コルーチンオブジェクト」が作られるだけです。
- 必ず await func()（async関数内）または asyncio.run(func())（エントリーポイント）で実行する必要があります。
time.sleep の禁止
- async def の中で time.sleep() を使うと、非同期処理のメリットが消え、プログラム全体がその場でフリーズします。
- 必ず await asyncio.sleep() を使用してください。
Jupyter Notebookでの実行
- Jupyter NotebookやGoogle Colabは既にイベントループが動いているため、asyncio.run() を使うとエラーになります。
- その場合は、単に await main() と書くだけで実行できます。

応用

複数の非同期処理を「並行して（同時に）」実行し、全体の待ち時間を短縮する asyncio.gather の例です。

import asyncio

async def task(name, seconds):
    print(f"{name} 開始 ({seconds}秒待機)")
    await asyncio.sleep(seconds)
    print(f"{name} 完了")
    return f"{name}の結果"

async def main():
    print("並行実行開始")
    # 2つのタスク（2秒と3秒）を同時に開始
    # 合計時間は 2+3=5秒 ではなく、長い方の 3秒 になります
    results = await asyncio.gather(
        task("Task-A", 2),
        task("Task-B", 3)
    )
    print(f"全ての結果: {results}")

if __name__ == "__main__":
    asyncio.run(main())