サイトアイコン アマチュア無線局JS2IIU

【Python】非同期プログラミング入門：asyncioを使った効率的なコーディング

JS2IIU

こんにちは、JS2IIUです。

非同期処理を実現するasyncioについて事例を紹介します。今回もよろしくお願いします。

1. はじめに

Pythonで複数の処理を同時に進めたいとき、非同期プログラミングが有効です。特にI/O操作（API通信、ファイル入出力、データベースアクセスなど）では、待ち時間を有効に使い、パフォーマンスを向上させることができます。

本記事では、Pythonの標準ライブラリであるasyncioを使った非同期プログラミングを、実践的なユースケースを交えながら解説します。

2. asyncioの基礎知識

2.1 asyncとawaitの基本

asyncとawaitは非同期処理を記述するためのPythonのキーワードです。

2.2 簡単な非同期関数の例

Python
import asyncio

async def say_hello():
    await asyncio.sleep(2)
    print("Hello, asyncio!")

asyncio.run(say_hello())

このコードでは、asyncio.sleep(2)で2秒間待機し、その後メッセージを出力します。

asyncioモジュールのインポート

Python
import asyncio

非同期関数の定義

Python
async def say_hello():

コルーチン（Coroutine）とは？

  • 一時停止と再開が可能な特殊な関数です。
  • Pythonのasync/await構文はこのコルーチンを操作します。

awaitによる非同期処理

Python
await asyncio.sleep(2)

ポイント

  • awaitを使うことで、他のタスクに処理を譲りながら待機できます。
  • これにより、CPUがアイドル状態になるのを防ぎ、リソースを有効活用できます。

メッセージの出力

Python
print("Hello, asyncio!")

asyncio.run()でコルーチンを実行

Python
asyncio.run(say_hello())

イベントループとは？

  • 非同期処理を管理・実行する仕組みです。
  • asyncio.run()は内部でイベントループを生成し、コルーチンを実行します。

プログラムの実行フロー

  1. asyncio.run(say_hello())でイベントループが作成され、say_hello()が実行開始。
  2. await asyncio.sleep(2)で2秒間の非同期待機を行う（この間、他のタスクがあれば実行可能）。
  3. 2秒経過後にprint("Hello, asyncio!")が実行され、コンソールにメッセージが表示される。
  4. コルーチンが終了し、イベントループが閉じられる。

3. ユースケース1：複数のAPIからデータを取得

複数のAPIを順番に呼び出すと時間がかかりますが、asyncioを使うと同時に実行でき、時間を節約できます。

3.1 同期処理の場合

Python
import requests
import time

def fetch_url(url):
    response = requests.get(url)
    return response.text

start = time.time()

for url in ["https://example.com", "https://example.org"]:
    print(fetch_url(url))

end = time.time()
print(f"Elapsed time: {end - start:.2f} seconds")

このコードを実行すると、各URLから取得した内容と処理時間が以下のように表示されます。

Bash
<!doctype html>
<html>...</html>
<!doctype html>
<html>...</html>
Elapsed time: 4.02 seconds

3.2 非同期処理の場合

aiohttpを使用して非同期でAPIを取得します。

Python
import aiohttp
import asyncio
import time

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

async def main():
    urls = ["https://example.com", "https://example.org"]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        for result in results:
            print(result)

start = time.time()
asyncio.run(main())
end = time.time()
print(f"Elapsed time: {end - start:.2f} seconds")

このコードでは、aiohttp.ClientSession()を使って複数のURLを並行して取得しています。

ライブラリのインポート

Python
import aiohttp
import asyncio
import time

URLからデータを取得する関数

Python
async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.text()
関数の仕組み
  1. async def
  1. session.get(url)
  1. await response.text()

💡 ポイント
この関数は1つのURLからデータを取得する処理を非同期で行います。

メイン関数

Python
async def main():
    urls = ["https://example.com", "https://example.org"]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        for result in results:
            print(result)
各行の説明
  1. urlsリストの定義
Python
   urls = ["https://example.com", "https://example.org"]
  1. async with aiohttp.ClientSession()
Python
   async with aiohttp.ClientSession() as session:
  1. タスクの生成
Python
   tasks = [fetch_url(session, url) for url in urls]
  1. 非同期実行の開始
Python
   results = await asyncio.gather(*tasks)
  1. 結果の出力
Python
   for result in results:
       print(result)

💡 重要なポイント

  • すべてのURLを同時にリクエストするため、リクエスト時間の最も長いURLに依存した待ち時間になります。
  • 同期処理では各URLを順番に処理するため、トータル時間が増えますが、ここでは並列で処理するため高速です。

時間計測とプログラムの実行

Python
start = time.time()
asyncio.run(main())
end = time.time()
print(f"Elapsed time: {end - start:.2f} seconds")
処理の流れ
  1. start = time.time()
  1. asyncio.run(main())
  1. end = time.time()
  1. 処理時間を表示
Python
   print(f"Elapsed time: {end - start:.2f} seconds")

実行結果

通常、2つのURLからデータを取得し、処理時間を表示します。

Plaintext
<!DOCTYPE html>
<html>...</html>
<!DOCTYPE html>
<html>...</html>
Elapsed time: 1.02 seconds

4. ユースケース2：大量のファイルを同時に処理

asyncio.to_thread()を使えば、CPUバウンドの処理も非同期で実行できます。

4.1 ファイルの非同期読み込み

Python
import asyncio

def read_file(file_path):
    with open(file_path, 'r') as file:
        return file.read()

async def main():
    file_paths = ["file1.txt", "file2.txt"]
    tasks = [asyncio.to_thread(read_file, path) for path in file_paths]
    results = await asyncio.gather(*tasks)
    for content in results:
        print(content)

asyncio.run(main())

read_file関数

Python
def read_file(file_path):
    with open(file_path, 'r') as file:
        return file.read()
説明
ポイント

main関数

Python
async def main():
    file_paths = ["file1.txt", "file2.txt"]
    tasks = [asyncio.to_thread(read_file, path) for path in file_paths]
    results = await asyncio.gather(*tasks)
    for content in results:
        print(content)
説明
  1. ファイルパスのリスト作成
Python
   file_paths = ["file1.txt", "file2.txt"]
  1. 非同期タスクの作成
Python
   tasks = [asyncio.to_thread(read_file, path) for path in file_paths]
  1. 複数の非同期タスクを並行実行
Python
   results = await asyncio.gather(*tasks)
  1. ファイル内容を出力
Python
   for content in results:
       print(content)

5. ユースケース3：チャットボットの非同期実装

複数のユーザーからの入力を非同期に処理するチャットボットを作成します。

5.1 非同期チャットボットの例

Python
import asyncio

async def chat_bot(user, delay):
    await asyncio.sleep(delay)
    print(f"{user}: Hello!")

async def main():
    users = [("User1", 3), ("User2", 1), ("User3", 2)]
    tasks = [chat_bot(user, delay) for user, delay in users]
    await asyncio.gather(*tasks)

asyncio.run(main())

このプログラムは複数のユーザー入力を並行して処理します。

chat_bot関数

Python
async def chat_bot(user, delay):
    await asyncio.sleep(delay)
    print(f"{user}: Hello!")
説明
  1. 非同期に待機： await asyncio.sleep(delay)
    • awaitを使って指定された秒数だけ一時停止。
    • 他のタスクはこの間も並行して実行されるため、処理が効率的に進む。
  2. メッセージ出力：
    python print(f"{user}: Hello!")
    • 待機時間が終了すると、userに対応するメッセージを出力。

main関数

Python
async def main():
    users = [("User1", 3), ("User2", 1), ("User3", 2)]
    tasks = [chat_bot(user, delay) for user, delay in users]
    await asyncio.gather(*tasks)
説明
  1. ユーザーと遅延時間のリストを作成：
Python
   users = [("User1", 3), ("User2", 1), ("User3", 2)]
  1. 非同期タスクの作成：
Python
   tasks = [chat_bot(user, delay) for user, delay in users]
  1. 複数のタスクを同時実行：
Python
   await asyncio.gather(*tasks)

プログラムの実行

Python
asyncio.run(main())
説明

6. エラーハンドリングとデバッグ

非同期処理でも例外処理を適切に行うことが重要です。

6.1 asyncio.TimeoutErrorの処理

Python
import asyncio

async def slow_task():
    await asyncio.sleep(5)

async def main():
    try:
        await asyncio.wait_for(slow_task(), timeout=3)
    except asyncio.TimeoutError:
        print("Timeout occurred")

asyncio.run(main())

7. ベストプラクティス

8. まとめと次のステップ

本記事では、Pythonのasyncioを使った非同期プログラミングの基礎と、実践的なユースケースを解説しました。

さらなる学習リソース

非同期プログラミングをマスターして、より効率的なPythonコードを目指しましょう！

最後に、書籍のPRです。
24年9月に出版された「ハイパーモダンPython-信頼性の高いワークフローを構築するモダンテクニック」、Claudio Jolowicz著、嶋田、鈴木訳。開発環境の構築、プロジェクトの管理、テストに関して実践的な内容でとても参考になる一冊です。ぜひ手に取ってみてください。

最後まで読んでいただきありがとうございます。

モバイルバージョンを終了