【第1回】Python asyncio 入門 — async/awaitとイベントループの基本を理解しよう

プログラミング
2026.02.23

最近、API呼び出しをまとめて処理するコードを書いていて「同期処理だと遅すぎる…」という壁にぶつかりました。調べていくうちに asyncio にたどり着いたんですが、最初は async・await・コルーチン・タスク・イベントループ……と知らない概念が一気に出てきて正直面食らいました。

このシリーズでは、そんな自分が「あ、そういうことか」と腑に落ちた順番で asyncio を丁寧に解説していきます。全4回構成で、第1回の今回は基礎中の基礎 ― async/await の使い方とイベントループの仕組みを中心に見ていきます。

  • 第1回：async/await とイベントループの基本（← 今回）
  • 第2回：タスクの並行実行（create_task / gather / TaskGroup）
  • 第3回：非同期 HTTP リクエスト（aiohttp 実践編）
  • 第4回：非同期処理のよくある落とし穴とデバッグ Tips

この記事でわかること

  • 非同期処理が必要な理由（同期処理との違い）
  • asyncio の全体像（コルーチン・タスク・イベントループの関係）
  • async def と await の基本的な書き方・使い方
  • asyncio.run() でプログラムを動かす方法
  • 同期処理と非同期処理の実行時間の違いを体感するコード例

そもそも非同期処理ってなんで必要なの？

まず「なぜ非同期が必要か」というところから整理しておきます。通常の Python コードは同期処理で動いていて、1行ずつ順番に実行されます。前の処理が終わるまで次の処理は始まりません。

たとえば、API を3回叩くとして、それぞれ2秒かかるとしたら…同期だと合計6秒待つことになります。でもよく考えると、3つのリクエストは互いに依存していないので、並行して実行できるはずですよね。

# 同期処理のイメージ（合計6秒かかる）
import time

def fetch_data(name):
    print(f"{name}: 取得開始")
    time.sleep(2)  # ← ここでプログラム全体が止まる
    print(f"{name}: 取得完了")

fetch_data("API-A")
fetch_data("API-B")
fetch_data("API-C")

time.sleep(2) の間、プログラムは何もできません。これが「ブロッキング」な同期処理の問題です。

非同期処理を使えば、待ち時間の間に別の処理を走らせることができます。料理中に電子レンジをスタートして、チンと鳴るまでの間に別の作業をするイメージです（この例、いろんなところで使われてますが、本当にわかりやすい）。

非同期処理が特に有効なシーン

asyncio は I/Oバウンドな処理（待ち時間が支配的な処理）に特に効果を発揮します。具体的には：

  • Web API の呼び出し（HTTP リクエスト）
  • データベースへのクエリ
  • ファイルの読み書き
  • WebSocket によるリアルタイム通信

逆に、計算処理が重い CPU バウンドなタスクには asyncio はあまり向きません。そっちは multiprocessing の領域になります（それはまた別の話）。

asyncio の全体像を把握する

asyncio を理解するうえで最初に押さえておきたい登場人物が3つあります。これがふわっとしたまま進むと後でコードが読めなくなるので、最初にしっかり整理しておきます。

① コルーチン（Coroutine）

async def で定義した関数のことです。普通の関数と違って、途中で処理を一時停止して、他の処理に制御を渡せるという特性があります。呼び出しても即実行はされず、コルーチンオブジェクトが返ってきます。

import asyncio

# async def でコルーチン関数を定義
async def greet(name):
    print(f"こんにちは、{name}！")
    await asyncio.sleep(1)  # ← 1秒待つ間、他の処理に譲れる
    print(f"{name}さん、処理完了！")

② タスク（Task）

コルーチンをイベントループに登録して「ちゃんと実行してね」とスケジューリングしたものです。asyncio.create_task() で作成します。タスクにすることで、複数の処理を並行して動かせるようになります。

③ イベントループ（Event Loop）

非同期処理の司令塔です。登録されたタスクを監視して、「このタスクは今 I/O 待ちだから、あっちのタスクを先に進めよう」という切り替えを自動でやってくれます。

自分たちは asyncio.run() を呼ぶだけでイベントループが起動されるので、普段はあまり意識しなくていいんですが、「裏でこういう仕組みが動いている」という感覚は持っておくと理解が深まります。

整理するとこういう関係です：

コルーチン（async def）
  ↓ create_task() でスケジュール登録
タスク
  ↓ 管理・実行
イベントループ（asyncio.run() で起動）

async/await の基本的な書き方・使い方

では実際にコードを書いていきましょう。まずは最もシンプルな例から。

Hello World 的なやつ

import asyncio

async def main():
    print("処理スタート")
    await asyncio.sleep(1)  # 1秒待機（非同期）
    print("1秒後に処理再開")

# asyncio.run() でイベントループを起動してコルーチンを実行
asyncio.run(main())

ポイント解説：

  • async def main()：コルーチン関数の定義。async をつけるだけで非同期関数になります
  • await asyncio.sleep(1)：非同期の待機。time.sleep() と違い、待機中にイベントループが他のタスクを処理できます
  • asyncio.run(main())：Python 3.7 で追加された関数で、非同期コードを実行します。イベントループの作成・起動からコルーチンの実行、最後のクリーンアップまで一気にやってくれるので、基本はこれをエントリーポイントにするのがわかりやすいです

余談ですが、昔の asyncio のチュートリアルを見ると loop = asyncio.get_event_loop() 的な書き方が出てきて混乱した記憶があります。最近のドキュメントでも、コルーチン内では get_event_loop() より get_running_loop() が推奨されていたり、状況によっては get_event_loop() が警告を出したりするので、古い記事を読むときは注意です。

await が使える場所に注意

await が使えるのは async 関数の中だけです。これを間違えると「SyntaxError: ‘await’ outside function」というエラーが出ます。

# ❌ NG：通常の関数の中で await は使えない
def bad_example():
    await asyncio.sleep(1)  # SyntaxError！

# ✅ OK：async def の中でのみ使える
async def good_example():
    await asyncio.sleep(1)  # 問題なし

また await に渡せるのは「awaitable」なオブジェクト（コルーチン・タスク・Future）だけです。普通の関数や値を await しようとするとエラーになります。

同期 vs 非同期：実行時間を比較してみる

ここが一番わかりやすい部分だと思います。実際に時間を計って、非同期処理の効果を体感してみましょう。

同期処理バージョン

import time

def fetch(name, delay):
    print(f"[{name}] 開始")
    time.sleep(delay)  # ブロッキング待機
    print(f"[{name}] 完了（{delay}秒かかった）")

start = time.time()
fetch("タスクA", 2)
fetch("タスクB", 2)
fetch("タスクC", 2)
elapsed = time.time() - start
print(f"\n合計時間: {elapsed:.1f}秒")
# → 合計時間: 6.0秒

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

import asyncio
import time

async def fetch(name, delay):
    print(f"[{name}] 開始")
    await asyncio.sleep(delay)  # ノンブロッキング待機
    print(f"[{name}] 完了（{delay}秒かかった）")

async def main():
    start = time.time()

    # 3つのコルーチンを同時にスケジュール
    task_a = asyncio.create_task(fetch("タスクA", 2))
    task_b = asyncio.create_task(fetch("タスクB", 2))
    task_c = asyncio.create_task(fetch("タスクC", 2))

    # 全タスクの完了を待つ
    await task_a
    await task_b
    await task_c

    elapsed = time.time() - start
    print(f"\n合計時間: {elapsed:.1f}秒")
    # → 合計時間: 2.0秒

asyncio.run(main())

実行結果のイメージ：

[タスクA] 開始
[タスクB] 開始
[タスクC] 開始
[タスクA] 完了（2秒かかった）
[タスクB] 完了（2秒かかった）
[タスクC] 完了（2秒かかった）

合計時間: 2.0秒

同期だと6秒かかっていたものが、非同期では2秒で完了しました。3つのタスクが「待ちの間は他に譲る」ことで並行して進んでいるのがわかります。

asyncio.create_task() でタスクを作成した時点でイベントループへの登録が行われ、await で実際に完了を待ちます。このあたりの詳しい話（gather や TaskGroup など）は第2回で掘り下げています。

よくあるハマりポイント

自分が最初に引っかかったミスをいくつか紹介しておきます。

time.sleep() を使ってしまう

time.sleep() は同期関数で処理をブロック（停止）しますが、asyncio.sleep() は非同期的に待機します。他のタスクがある場合、この待機時間の間に他の処理を実行できます。

import asyncio
import time

async def bad():
    time.sleep(2)      # ❌ イベントループ全体が2秒止まる

async def good():
    await asyncio.sleep(2)  # ✅ 待ち時間に他タスクが動ける

コルーチンを await せずに呼ぶ

コルーチン関数を呼び出しただけでは実行されません。await をつけるか、create_task() でタスク化する必要があります。

async def main():
    greet("テスト")         # ❌ コルーチンオブジェクトが返るだけで実行されない
    await greet("テスト")   # ✅ これで実行される

await なしで呼ぶと「RuntimeWarning: coroutine ‘greet’ was never awaited」という警告が出ます。これが出たらまず await 忘れを疑ってください。

async def でない関数内で await を使う

await は必ず async def の中で使う必要があります。「async の感染」みたいなイメージで、await を使いたい関数は全部 async def にしていく必要があります。これが最初ちょっとめんどくさく感じるところかもしれません。

まとめ

第1回の内容を整理します：

  • 非同期処理は I/O バウンドな処理（API 呼び出しや DB アクセス）での「待ち時間の有効活用」に効果的
  • asyncio の3大概念：コルーチン（async def）、タスク（create_task）、イベントループ（asyncio.run）
  • async def でコルーチン関数を定義し、await で非同期待機する
  • asyncio.run() がエントリーポイント。Python 3.7 で追加された、推奨されがちな書き方
  • 非同期コードの中では time.sleep() ではなく asyncio.sleep() を使う
  • await は async def の中でしか使えない

ここまで読んでくれてありがとうございます。自分の理解整理がメインですが、少しでも参考になれば！

📚 Python asyncio 非同期処理入門 シリーズ

▶ 今読んでいる記事：【第1回】async/awaitとイベントループの基本を理解しよう

→ 次回：公開されたらここにリンクが入ります

関連記事

参考になったらクリックしてもらえると嬉しいです!

Blogmura ProgrammingProgramming Ranking
タイトルとURLをコピーしました