フリーキーズ | 独学プログラミング
技術コンテンツ転職フリーランスツール用語集ゲームタグ一覧

Pythonのasyncioで非同期プログラミング入門！初心者でもわかる基本と実践

Python

最終更新

リンドくん

リンドくん

先生、Pythonの「asyncio」ってこれは何なんですか？普通のプログラミングと何が違うんでしょうか？

たなべ

たなべ

asyncioは非同期プログラミングを可能にするPythonの標準ライブラリなんだ。
簡単に言うと、複数の処理を効率的に同時進行させることができるんだよ。従来の方法よりもずっと効率的にプログラムを動かせるんだ。

プログラミングを学んでいると、「処理が重くて時間がかかる...」「複数のAPIから同時にデータを取得したい...」といった課題に直面することがあるのではないでしょうか？

そんな時に威力を発揮するのが、Pythonのasyncioを使った非同期プログラミングです。
非同期プログラミングを理解することで、より効率的で高速なプログラムを作成できるようになります。

この記事では、asyncio初心者の方でも理解できるよう、基本概念から実践的な使い方まで、段階的に解説していきます。

プログラミング学習でお悩みの方へ

HackATAは、エンジニアを目指す方のためのプログラミング学習コーチングサービスです。 経験豊富な現役エンジニアがあなたの学習をサポートします。

✓ 質問し放題

✓ β版公開中（2025年内の特別割引）

HackATAの詳細を見る

非同期プログラミングとは何か？

リンドくん

リンドくん

そもそも「非同期」って何ですか？普通のプログラムとどう違うんでしょうか？

たなべ

たなべ

レストランの例で考えてみよう。
同期処理は一人のウェイターが一つの注文を完全に処理してから次の注文を受ける方法。
非同期処理は複数の注文を同時に管理して、効率的にサービスを提供する方法なんだ。

同期処理と非同期処理の違い

通常のPythonプログラム（同期処理）では、コードは上から下へ順番に実行されます。
一つの処理が完了するまで、次の処理は待機状態になります。

import time

def sync_example():
    print("処理1開始")
    time.sleep(2)  # 2秒待機
    print("処理1完了")
    
    print("処理2開始")
    time.sleep(2)  # 2秒待機
    print("処理2完了")
    
    print("すべて完了")

# 実行すると合計4秒かかる
sync_example()

これに対して、非同期処理では複数の処理を並行して実行できます。

import asyncio

async def async_task(name, duration):
    print(f"{name}開始")
    await asyncio.sleep(duration)  # 非同期での待機
    print(f"{name}完了")

async def async_example():
    # 複数のタスクを同時に実行
    await asyncio.gather(
        async_task("処理1", 2),
        async_task("処理2", 2)
    )
    print("すべて完了")

# 実行すると約2秒で完了
asyncio.run(async_example())

この例では、2つの処理が並行して実行されるため、合計時間が大幅に短縮されます。

非同期プログラミングのメリット

非同期プログラミングの主なメリットは以下の通りです。

  • 処理時間の短縮 → 複数の処理を並行実行することで全体の実行時間を削減
  • リソースの効率的活用 → CPUやメモリを無駄なく使用
  • レスポンシブなアプリケーション → ユーザーインターフェースがフリーズしない
  • スケーラビリティの向上 → 大量のリクエストを効率的に処理

特にWebアプリケーション開発や、外部APIとの連携、ファイルのダウンロードなど、待機時間が発生する処理において、その威力を発揮します。

async/awaitの基本的な使い方

リンドくん

リンドくん

asyncとawaitってキーワードがよく出てきますが、これらはどういう意味なんですか？

たなべ

たなべ

asyncは「この関数は非同期関数です」という宣言で、awaitは「ここで他の処理が完了するまで待ちます」という指示なんだ。
この2つのキーワードが非同期プログラミングの基本になるよ。

基本的な構文

非同期プログラミングでは、主に以下の2つのキーワードを使用します。

async - 非同期関数を定義する際に使用

async def my_async_function():
    # 非同期関数の内容
    pass

await - 非同期処理の完了を待つ際に使用

async def example():
    result = await some_async_function()
    return result

シンプルな非同期関数の例

まずは基本的な非同期関数の書き方を見てみましょう。

import asyncio

async def greet_after_delay(name, delay):
    """指定した時間後に挨拶を表示する非同期関数"""
    print(f"{name}さん、{delay}秒後にメッセージを表示します")
    await asyncio.sleep(delay)  # 非同期で待機
    print(f"こんにちは、{name}さん！")
    return f"完了: {name}"

async def main():
    """メイン処理"""
    # 単一のタスクを実行
    result = await greet_after_delay("たなべ", 1)
    print(result)

# 非同期関数の実行
asyncio.run(main())

複数の非同期処理を並行実行

複数の処理を同時に実行したい場合は、asyncio.gather()やasyncio.create_task()を使用します。

import asyncio

async def download_file(filename, size):
    """ファイルダウンロードをシミュレート"""
    print(f"{filename}のダウンロード開始 ({size}MB)")
    
    # サイズに応じてダウンロード時間を設定
    download_time = size * 0.1
    await asyncio.sleep(download_time)
    
    print(f"{filename}のダウンロード完了!")
    return f"{filename} ({size}MB)"

async def main():
    """複数ファイルを並行してダウンロード"""
    tasks = [
        download_file("document.pdf", 5),
        download_file("image.jpg", 2),
        download_file("video.mp4", 10)
    ]
    
    # すべてのタスクを並行実行
    results = await asyncio.gather(*tasks)
    
    print("すべてのダウンロードが完了しました:")
    for result in results:
        print(f"  - {result}")

asyncio.run(main())

この例では、3つのファイルが並行してダウンロードされるため、個別に実行するよりもはるかに高速に処理が完了します。

asyncioの活用例

リンドくん

リンドくん

実際のプロジェクトでは、asyncioをどんな風に使うんですか？

たなべ

たなべ

WebAPI連携やWebスクレイピング、データベース操作など、実際の開発では外部リソースとの通信でasyncioが大活躍するんだ。具体例を見てみよう！

例1 複数WebAPIからのデータ取得

実際の開発でよくあるシナリオとして、複数のAPIから同時にデータを取得する例を見てみましょう。

import asyncio
import aiohttp  # 非同期HTTP通信ライブラリ（pip install aiohttp）
import time

async def fetch_api_data(session, url, api_name):
    """APIからデータを取得する非同期関数"""
    try:
        print(f"{api_name} API からデータ取得開始...")
        async with session.get(url) as response:
            data = await response.json()
            print(f"{api_name} API からデータ取得完了!")
            return {api_name: data}
    except Exception as e:
        print(f"{api_name} API でエラー: {e}")
        return {api_name: None}

async def get_multiple_api_data():
    """複数のAPIから並行してデータを取得"""
    start_time = time.time()
    
    apis = [
        ("https://jsonplaceholder.typicode.com/posts/1", "Posts"),
        ("https://jsonplaceholder.typicode.com/users/1", "Users"),
        ("https://jsonplaceholder.typicode.com/albums/1", "Albums")
    ]
    
    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_api_data(session, url, name) 
            for url, name in apis
        ]
        
        results = await asyncio.gather(*tasks)
    
    end_time = time.time()
    execution_time = end_time - start_time
    
    print(f"\n実行時間: {execution_time:.2f}秒")
    print("取得したデータ:")
    for result in results:
        for api_name, data in result.items():
            if data:
                print(f"  {api_name}: {list(data.keys())}")

# 実行
asyncio.run(get_multiple_api_data())

例2 Webスクレイピングの高速化

複数のWebページから情報を収集する際にも、asyncioが威力を発揮します。

import asyncio
import aiohttp
from bs4 import BeautifulSoup

async def scrape_page(session, url):
    """単一ページをスクレイピング"""
    try:
        async with session.get(url) as response:
            html = await response.text()
            soup = BeautifulSoup(html, 'html.parser')
            
            # タイトルを取得（例）
            title = soup.find('title')
            title_text = title.text.strip() if title else "タイトル不明"
            
            return {
                'url': url,
                'title': title_text,
                'status': response.status
            }
    except Exception as e:
        return {
            'url': url,
            'title': f"エラー: {e}",
            'status': 'error'
        }

async def scrape_multiple_pages(urls):
    """複数ページを並行スクレイピング"""
    print(f"{len(urls)}個のページをスクレイピング開始...")
    
    async with aiohttp.ClientSession() as session:
        tasks = [scrape_page(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
    
    print("スクレイピング完了!")
    return results

# 使用例
async def main():
    urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/2",
        "https://httpbin.org/delay/1"
    ]
    
    start_time = time.time()
    results = await scrape_multiple_pages(urls)
    end_time = time.time()
    
    print(f"実行時間: {end_time - start_time:.2f}秒")
    
    for result in results:
        print(f"URL: {result['url']}")
        print(f"ステータス: {result['status']}")
        print("---")

asyncio.run(main())

例3 ファイル処理の並行実行

大量のファイル処理も非同期で効率化できます。

import asyncio
import aiofiles  # 非同期ファイル操作（pip install aiofiles）
import os

async def process_file(filepath):
    """ファイルを非同期で処理"""
    try:
        async with aiofiles.open(filepath, 'r', encoding='utf-8') as file:
            content = await file.read()
            
        # ファイル処理をシミュレート（例：行数カウント）
        line_count = len(content.splitlines())
        word_count = len(content.split())
        
        return {
            'file': os.path.basename(filepath),
            'lines': line_count,
            'words': word_count
        }
    except Exception as e:
        return {
            'file': os.path.basename(filepath),
            'error': str(e)
        }

async def process_multiple_files(file_paths):
    """複数ファイルを並行処理"""
    tasks = [process_file(path) for path in file_paths]
    results = await asyncio.gather(*tasks)
    
    print("ファイル処理結果:")
    for result in results:
        if 'error' in result:
            print(f"  {result['file']}: エラー - {result['error']}")
        else:
            print(f"  {result['file']}: {result['lines']}行, {result['words']}語")

# 使用例（実際のファイルパスに置き換えてください）
file_list = ["file1.txt", "file2.txt", "file3.txt"]
asyncio.run(process_multiple_files(file_list))

これらの例では、同期処理と比較して大幅な処理時間の短縮が期待できます。特に外部リソースとの通信が多い場合、その効果は顕著に現れます。

asyncioのよくある注意点と解決法

リンドくん

リンドくん

asyncioを使っているとエラーが出ることがあるんですが、初心者が気をつけるべきことはありますか？

たなべ

たなべ

asyncioには独特のルールがあるから、最初は戸惑うことも多いよね。
よくあるパターンを知っておけば、多くの問題は事前に避けられるんだ。

よくある間違い1 async関数内でのtime.sleep()使用

間違った例

import asyncio
import time

async def bad_example():
    print("開始")
    time.sleep(2)  # ❌ 同期的なsleepを使用
    print("完了")

# この場合、非同期の利点が失われる

正しい例

import asyncio

async def good_example():
    print("開始")
    await asyncio.sleep(2)  # ✅ 非同期のsleepを使用
    print("完了")

よくある間違い2 awaitを忘れる

間違った例

async def fetch_data():
    await asyncio.sleep(1)
    return "データ"

async def bad_main():
    result = fetch_data()  # ❌ awaitを忘れている
    print(result)  # <coroutine object> が表示される

正しい例:

async def good_main():
    result = await fetch_data()  # ✅ awaitを使用
    print(result)  # "データ" が表示される

よくある間違い3 非同期ライブラリの選択ミス

外部ライブラリを使用する際は、非同期対応のものを選ぶ必要があります。

推奨ライブラリの組み合わせ

# HTTP通信
import aiohttp  # requests の代わり

# データベース操作
import asyncpg  # PostgreSQL用
import aiomysql  # MySQL用

# ファイル操作
import aiofiles  # 標準のopen() の代わり

# Redis操作
import aioredis  # redis-py の代わり

エラーハンドリングのベストプラクティス

非同期処理でのエラーハンドリングは特に重要です。

import asyncio

async def risky_operation(task_id):
    """エラーが発生する可能性のある処理"""
    await asyncio.sleep(1)
    if task_id == 2:
        raise ValueError(f"タスク{task_id}でエラー発生")
    return f"タスク{task_id}完了"

async def safe_operation(task_id):
    """エラーハンドリングを含む安全な処理"""
    try:
        result = await risky_operation(task_id)
        print(result)
        return result
    except Exception as e:
        error_msg = f"タスク{task_id}でエラー: {e}"
        print(error_msg)
        return error_msg

async def main_with_error_handling():
    """複数タスクのエラーハンドリング"""
    tasks = [safe_operation(i) for i in range(1, 4)]
    
    # すべてのタスクを実行（一部エラーでも他は継続）
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    print("すべてのタスクが完了しました")
    print("結果:", results)

asyncio.run(main_with_error_handling())

パフォーマンス最適化のコツ

import asyncio

async def optimized_example():
    """パフォーマンスを意識した非同期処理"""
    
    # 1. タスクの数を制限（同時実行数の制御）
    semaphore = asyncio.Semaphore(5)  # 最大5つまで同時実行
    
    async def limited_task(task_id):
        async with semaphore:
            await asyncio.sleep(1)
            return f"タスク{task_id}完了"
    
    # 2. create_task()を使用してタスクを事前作成
    tasks = [
        asyncio.create_task(limited_task(i)) 
        for i in range(20)
    ]
    
    # 3. 結果の取得
    results = await asyncio.gather(*tasks)
    return results

# 実行
results = asyncio.run(optimized_example())
print(f"処理完了: {len(results)}件")

これらの注意点を理解しておくことで、asyncioを使った開発がよりスムーズに進むでしょう。

まとめ

リンドくん

リンドくん

asyncioって最初は難しそうに見えましたが、基本を理解すれば案外使えそうですね！

たなべ

たなべ

そうなんだ！最初は戸惑うかもしれないけど、async/awaitの基本パターンを覚えてしまえば、あとは応用していくだけ。
特にWebアプリ開発では必須のスキルになってきているから、ぜひマスターしてほしいな。

この記事では、Pythonのasyncioを使った非同期プログラミングについて、基本概念から実践的な活用例まで解説しました。

重要なポイントをおさらいしましょう。

  • 非同期プログラミングは複数の処理を効率的に並行実行する手法
  • async/awaitの基本構文をマスターすることが第一歩
  • 外部API連携やWebスクレイピングで真価を発揮
  • 適切なライブラリ選択とエラーハンドリングが成功の鍵

非同期プログラミングは、現代のWebアプリケーション開発において必須のスキルとなっています。
最初は慣れないかもしれませんが、基本パターンを覚えてしまえば、従来の同期処理では実現できない高速で効率的なプログラムを作成できるようになります。

ぜひ今回紹介したサンプルコードを実際に動かしてみて、asyncioの威力を体感してみてください。そして、自分のプロジェクトでも積極的に活用して、より良いプログラムを作り上げていきましょう！

この記事をシェア

関連するコンテンツ
Pythonのunittest.mockで依存をモックしてテストを書く方法！初心者向けガイド

Pythonのunittest.mockで依存をモックしてテストを書く方法！初心者向けガイド

Python

2025/09/18

Python初心者向けunittestの使い方！標準テストフレームワークで始めるテスト駆動開発

Python初心者向けunittestの使い方！標準テストフレームワークで始めるテスト駆動開発

Python

2025/09/17

Python型ヒント入門！typingモジュールで静的解析を活用してコードの品質を向上させよう

Python型ヒント入門！typingモジュールで静的解析を活用してコードの品質を向上させよう

Python

2025/09/16

Pythonのリスト・辞書内包表記でコードを簡潔に！初心者向け基本から応用まで

Pythonのリスト・辞書内包表記でコードを簡潔に！初心者向け基本から応用まで

Python

2025/09/15

Pythonイテレータとイテラブルの違いとは？初心者向けに解説

Pythonイテレータとイテラブルの違いとは？初心者向けに解説

Python

2025/09/12

Pythonコンテキストマネージャを自作してwith文を極める！初心者でもわかる実装方法と活用術

Pythonコンテキストマネージャを自作してwith文を極める！初心者でもわかる実装方法と活用術

Python

2025/09/11