Pythonの非同期処理入門｜async / await構文とasyncioの使い方をやさしく解説！

はじめに｜非同期処理ってなに？なんで必要？

みなさん、Pythonでプログラムを書いていると「処理に時間がかかるな〜」と思ったこと、ありませんか？

たとえば、ネットからデータをダウンロードしたり、大きなファイルを読み込んだりすると、その処理が終わるまでプログラム全体が“待ち”の状態になりますよね。その間、ほかの処理が止まってしまうと、とてももったいないし非効率です。

そんなときに使えるのが「非同期処理（ひどうきしょり）」です！

非同期処理ってなに？

非同期処理とは、今行っている処理の完了を待たずに、次の処理に進むことができるしくみです。

たとえば料理でいうと、スープを煮込んでいる間にサラダを作る、みたいなイメージ。煮込みが終わるのをじーっと待つのではなく、「待ってる間にできることをやっちゃおう！」という考え方なんです。

Pythonでは、asyncやawaitというキーワードを使うことで、こうした非同期処理をカンタンに実装できるようになりました。

なんで非同期処理が必要なの？

非同期処理は、特に**時間のかかる処理（I/O処理）**で大活躍します。

たとえばこんな場面です：

• ネットから複数ファイルをダウンロードしたい
• APIを使ってデータを連続取得したい
• Webサーバーで複数のリクエストを同時に処理したい

こうした処理は、ひとつずつ待っていたら非常に時間がかかりますが、非同期にすることで複数の処理を“同時に”進められるようになります。

---

1. 非同期処理の基本とasync/await構文の役割

さて、ここからはいよいよ本題。「Pythonで非同期処理をどうやって書くのか？」を見ていきましょう！

Pythonには、非同期処理を簡単に扱うためのキーワードがあります。それが async と await です。

この2つを使うことで、従来の面倒だった非同期処理がとてもシンプルに書けるようになりました。

---

async defで非同期関数（コルーチン）を作ろう

まず、Pythonでは非同期処理をする関数のことを**コルーチン（coroutine）**と呼びます。コルーチンは、async defで定義します。

import asyncio

async def hello():
    print("こんにちは！")
    await asyncio.sleep(1)
    print("1秒待ちました")

ここでのポイントは、await asyncio.sleep(1)の部分。

これは、1秒間「待つ」処理ですが、普通のtime.sleep(1)と違って、他の処理を止めずに「待つ」ことができるんです。

---

awaitでコルーチンを呼び出す

コルーチンは、awaitを使って呼び出す必要があります。awaitは「この処理が終わるまでちょっと待ってね」という合図です。

await hello()

ただし、これだけではエラーになります。なぜなら、awaitはコルーチンの中でしか使えないからです。

そのため、最初に実行する処理を含んだコルーチンをasync defで定義し、asyncio.run()で実行します。

async def main():
    await hello()

asyncio.run(main())

このように書くことで、「イベントループ」と呼ばれる処理の土台が自動的に作られ、非同期処理がちゃんと動くようになります。

---

通常の関数との違いは？

ここまで見て「普通の関数と何が違うの？」と思うかもしれませんが、非同期関数には以下の特徴があります。

通常の関数	非同期関数（async def）
defで定義する	async defで定義する
すぐに実行される	実行にはawaitが必要
1つずつ順番に進む	複数の処理を同時に進められる

---

2. 実際に書いてみよう｜基本の非同期関数とsleep処理

非同期処理の雰囲気がつかめてきたところで、今度は実際にコードを書いて試してみましょう！

ここでは、複数の処理を同時に実行するという非同期処理の強みを実感できるように、asyncio.sleep()を使った簡単な例を紹介します。

---

同時に2つの処理を走らせるとどうなる？

まずは、「1秒待ってメッセージを表示する」非同期関数を2つ作ってみます。

import asyncio

async def task1():
    print("タスク1：開始")
    await asyncio.sleep(1)
    print("タスク1：完了")

async def task2():
    print("タスク2：開始")
    await asyncio.sleep(1)
    print("タスク2：完了")

async def main():
    await task1()
    await task2()

asyncio.run(main())

このコードでは、task1が終わってからtask2が始まるため、合計で約2秒かかります。

---

同時実行するには「create_task()」を使う！

それでは、この2つのタスクを並行して実行してみましょう。

async def main():
    t1 = asyncio.create_task(task1())
    t2 = asyncio.create_task(task2())

    await t1
    await t2

今度は、create_task()でそれぞれのコルーチンをタスク化しています。これにより、2つの処理が同時に動くようになり、全体で約1秒しかかかりません！

タスク1：開始
タスク2：開始
（1秒待つ）
タスク1：完了
タスク2：完了

このように、同じ待ち時間の処理を一緒に実行すれば、待ち時間を節約できるんです！

---

time.sleep()とasyncio.sleep()の違いに注意！

ここで気をつけたいのが、time.sleep()との違いです。

import time

def blocking_task():
    print("ブロッキング開始")
    time.sleep(1)
    print("ブロッキング終了")

---

3. タスクと並行処理｜asyncio.create_task()とgather()

前回は asyncio.create_task() を使って複数の処理を同時に実行する方法を紹介しましたね。ここでは、そのタスクをもっと便利に扱う方法を深掘りしていきましょう！

---

asyncio.create_task()でもう一度おさらい

asyncio.create_task()は、非同期関数（コルーチン）を「タスク」という実行可能な形に変換してくれます。

タスクは、イベントループに登録された後、裏で勝手に動き出すのがポイント。

t1 = asyncio.create_task(task1())
t2 = asyncio.create_task(task2())

await t1
await t2

このように書くことで、両方のタスクが同時に動き始めるんですね。

---

asyncio.gather()で複数タスクを一気に処理！

タスクが複数あるときに便利なのが asyncio.gather()。

これは、複数のコルーチンやタスクをまとめて実行し、結果をリストで返してくれる便利な関数です。

results = await asyncio.gather(task1(), task2())

このコードでは、task1()とtask2()が同時に実行されて、それぞれの戻り値がresultsにリストで入ります。

---

create_task()とgather()の違いは？

特徴	create_task()	gather()
処理の開始	即座にイベントループに登録して動く	実行時に内部でタスク化されて動く
戻り値の扱い	.result()で個別に取得（後述）	awaitするとリストで一括取得できる
主な用途	明示的にタスクを作りたいとき	複数タスクをまとめて一括処理したいとき

💡どちらを使うか迷ったら、「戻り値が必要 → gather()」「細かくタスクを管理したい → create_task()」が目安になります！

---

タスクの戻り値を使いたいときは？

以下のように、各タスクの戻り値を受け取りたいときには gather()が便利です：

async def say_hello(name):
    await asyncio.sleep(1)
    return f"Hello, {name}!"

async def main():
    greetings = await asyncio.gather(
        say_hello("Alice"),
        say_hello("Bob"),
        say_hello("Charlie")
    )
    print(greetings)

asyncio.run(main())

実行結果：

['Hello, Alice!', 'Hello, Bob!', 'Hello, Charlie!']

---

タスクをリスト化してループ処理もOK！

もちろん、リスト内包表記を使って大量のタスクを一気に処理することも可能です。

names = ["Tom", "Jerry", "Spike"]
tasks = [say_hello(name) for name in names]
results = await asyncio.gather(*tasks)

このように、gather(*リスト)の形にすれば、一気に全部並行処理できますよ。

---

4. 通常関数の非同期化｜run_in_executor()の使い方

ここまでで、Pythonの非同期処理では async と await を使ってコルーチンを実行する方法を学びました。

でも、現実のアプリ開発では time.sleep() や画像処理・計算処理など、**非同期じゃない普通の関数（同期関数）**を使う場面もたくさんあります。

このような関数をそのまま使うと、非同期処理が止まってしまって逆効果です。

そこで登場するのが、**run_in_executor()**という機能です！

---

同期関数を非同期で動かすには？

たとえば、次のようなブロッキング関数があるとします。

import time

def heavy_task():
    print("重い処理開始")
    time.sleep(2)
    print("重い処理終了")

これを非同期処理に混ぜてしまうと、2秒間ほかの処理が止まってしまいます。

そこで、run_in_executor()でこの関数を別スレッドで動かすようにします。

---

run_in_executor()の書き方

import asyncio

def heavy_task():
    print("重い処理開始")
    time.sleep(2)
    print("重い処理終了")

async def main():
    loop = asyncio.get_running_loop()
    await loop.run_in_executor(None, heavy_task)

asyncio.run(main())

これで、heavy_task()は非同期処理に影響を与えず、別スレッドで実行されます！

---

引数付きの関数を使いたいときは？

もちろん、引数付きの関数も渡せます！

def greet(name):
    time.sleep(1)
    return f"Hello, {name}"

async def main():
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(None, greet, "Alice")
    print(result)

asyncio.run(main())

実行結果：

Hello, Alice

✅ Noneを指定すると、デフォルトのスレッドプールが使われます。
必要に応じて concurrent.futures.ThreadPoolExecutor や ProcessPoolExecutor を使うこともできます。

---

補足：本当に重い処理にはProcessPoolExecutorも検討しよう

画像処理やAI推論のようなCPU負荷が高い処理を非同期で扱いたいときは、スレッドではなく**プロセスベースの並列化（multiprocessing）**が向いています。

この場合は以下のように書きます：

from concurrent.futures import ProcessPoolExecutor

async def main():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as executor:
        result = await loop.run_in_executor(executor, heavy_task)

---

5. 処理をキャンセルしたい！タイムアウト処理の例

非同期処理では「思ったより時間がかかる…」「一定時間以上かかったらやめたい」といった場面に出くわすことがあります。

そんなときに活躍するのが、asyncio.wait_for() という便利な関数です！

---

asyncio.wait_for()とは？

wait_for()は、ある処理（コルーチン）にタイムリミットを設定できる関数です。

await asyncio.wait_for(コルーチン, timeout=秒数)

指定した時間内にコルーチンが終わらなければ、**asyncio.TimeoutError**が発生し、処理が強制キャンセルされます。

---

実際の使い方を見てみよう

import asyncio

async def slow_task():
    print("ゆっくりした処理を開始...")
    await asyncio.sleep(5)
    print("完了！")

async def main():
    try:
        await asyncio.wait_for(slow_task(), timeout=2)
    except asyncio.TimeoutError:
        print("タイムアウトしました！")

asyncio.run(main())

このコードでは、5秒かかる処理に対して2秒のタイムリミットを設定しています。結果はこうなります：

ゆっくりした処理を開始...
タイムアウトしました！

await asyncio.sleep(5)が終わる前に、タイムアウトで中断されたのがわかりますね。

---

タイムアウト後に行いたい処理は？

タイムアウト後もエラーでプログラムが止まらないよう、try / exceptブロックを使うのがポイントです。

また、タイムアウト後にログを残したり、他の処理に切り替えたりすることもできます。

try:
    await asyncio.wait_for(api_call(), timeout=3)
except asyncio.TimeoutError:
    await log_error("APIが応答しませんでした")

---

タイムアウトをうまく使えば…

• 重たいAPI呼び出しが止まっても他の処理に影響なし！
• ユーザーの待ち時間を制御できる！
• エラー対応も柔軟にできる！

---

6. 実例：複数ファイルの非同期ダウンロード処理

ここまで非同期処理の基本を学んできましたが、「実際のアプリケーションでどう使うの？」という疑問が出てきますよね。

今回は、複数のファイルを同時にダウンロードする非同期処理の例を紹介します！

---

実用例：URLから複数ファイルを並行して取得

たとえば、以下のような処理を考えてみましょう：

• 画像やテキストなどのファイルを複数ダウンロードしたい
• それぞれの処理を1つずつ待っていると時間がもったいない
• 非同期でまとめて処理できたら速く終わる！

---

aiohttpを使った非同期HTTPダウンロード

PythonのHTTPリクエストライブラリとして有名なのはrequestsですが、これは同期処理専用です。

非同期でHTTPリクエストを行うなら、aiohttpライブラリがおすすめです！

pip install aiohttp

---

サンプルコード｜複数のURLを同時にダウンロード

import asyncio
import aiohttp

urls = [
    "https://example.com/file1.txt",
    "https://example.com/file2.txt",
    "https://example.com/file3.txt",
]

async def download_file(session, url):
    async with session.get(url) as response:
        content = await response.text()
        print(f"{url} をダウンロードしました（{len(content)}文字）")

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [download_file(session, url) for url in urls]
        await asyncio.gather(*tasks)

asyncio.run(main())

このコードでは、

• 非同期HTTPセッションを開く（aiohttp.ClientSession()）
• 各ファイルのダウンロード処理を非同期関数download_file()で定義
• asyncio.gather()で一気に並列ダウンロード！

という流れになっています。

---

実行結果（例）

https://example.com/file1.txt をダウンロードしました（1560文字）
https://example.com/file3.txt をダウンロードしました（1489文字）
https://example.com/file2.txt をダウンロードしました（1521文字）

順番はバラバラになることもありますが、全体の処理時間は1ファイル分とほぼ同じになります！

---

Windowsでエラーが出たら？

一部のWindows環境では、RuntimeError: Event loop is closed というエラーが出る場合があります。

その場合は、以下のコードを追加して回避できます。

import asyncio
import sys

if sys.platform.startswith("win"):
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

---

まとめ｜Pythonの非同期処理でアプリをもっと快適に！

ここまで、Pythonの非同期処理について「async / await」の基本から、実践的な使い方までを紹介してきました。

今回のポイントを振り返ってみましょう。

---

✅ 非同期処理のキホンまとめ

• async def で非同期関数（コルーチン）を定義
• await を使ってコルーチンの実行を待機
• asyncio.run() で非同期処理を開始
• asyncio.create_task() や asyncio.gather() で複数タスクを同時実行
• run_in_executor() で通常関数も非同期風に扱える
• asyncio.wait_for() でタイムアウト制御もできる

---

✅ 非同期処理が活きる場面

非同期処理は、特に時間のかかるI/O操作やネットワーク処理で効果を発揮します。

• Web APIをたくさん叩くとき
• 複数ファイルをダウンロード／アップロードしたいとき
• Webアプリで複数のリクエストを高速にさばきたいとき

こうしたケースで非同期処理を導入すれば、処理の無駄を大幅に減らし、アプリのレスポンスを改善できます！

---

あわせて読みたい

• 【Python入門8】for文とwhile文の違いと使い方をやさしく解説｜初心者でもすぐに理解できる！
• 【初心者向け】Pythonで九九を自動生成する方法｜for文でかんたん繰り返し処理を学ぼう
• 【Python入門】while文の使い方とbreak・continueの違いをやさしく解説！
• 【Python入門】MeCabで形態素解析！自然言語処理を始めよう【初心者向け】

---

よくある質問（Q&A）