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

Python非同期プログラミング入門！aiohttpで高速なHTTPリクエストを実現する方法

Python

最終更新

リンドくん

リンドくん

たなべ先生、PythonでWebサイトからデータを取得するとき、すごく時間がかかることがあるんです。何か早くする方法ってありますか？

たなべ

たなべ

非同期処理を使うことで、複数のHTTPリクエストを同時に処理して、大幅に高速化できるんだよ。
今日はaiohttpというライブラリを使った方法を教えるね。

プログラミングを学んでいる方なら、Webサイトからデータを取得する際に「思ったよりも時間がかかる...」と感じたことがあるのではないでしょうか？

特に複数のAPIを呼び出す必要がある場合や、大量のデータを処理する際には、従来の同期処理では処理時間が長くなってしまいがちです。

そんな問題を解決してくれるのが、Pythonの非同期処理です。
今回は、非同期HTTPリクエストを簡単に実現できる「aiohttp」というライブラリについて、初心者の方でもわかりやすく解説していきます。

この記事を読めば、あなたのPythonプログラムがこれまでよりもずっと高速になり、効率的なWeb開発ができるようになるはずです！

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

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

✓ 質問し放題

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

HackATAの詳細を見る

aiohttpとは？非同期処理の基本概念

リンドくん

リンドくん

そもそも「非同期処理」って何ですか？普通のPythonとどう違うんでしょうか？

たなべ

たなべ

わかりやすく例えると、レストランでの注文みたいなものかな。
従来の方法だと一つの注文が終わってから次の注文を聞く感じだけど、非同期処理だと複数の注文を同時に受けて、効率的に処理できるんだ。

aiohttpの基本概念

aiohttpは、Pythonで非同期HTTPクライアント・サーバー処理を行うためのライブラリです。
従来のrequestsライブラリとは異なり、複数のHTTPリクエストを並行して処理することができます。

従来の同期処理の問題点

  • 1つのリクエストが完了するまで次のリクエストが開始できない
  • 待機時間が積み重なって全体の処理時間が長くなる
  • CPU資源を有効活用できない

aiohttpを使った非同期処理の利点

  • 複数のリクエストを同時に並行処理できる
  • 待機時間を有効活用して全体の処理速度を向上
  • リソースを効率的に使用できる

速度の違いを具体的に比較

例えば、10個のAPIを呼び出すとき、それぞれが1秒かかる場合を考えてみましょう。

  • 同期処理の場合: 10秒（1秒 × 10回）
  • 非同期処理の場合: 約1秒（10個を同時実行）

この違いは、処理する件数が多くなればなるほど顕著に現れます。まさに革命的な速度向上を実現できるのです。

aiohttpの主な特徴

aiohttpには以下のような特徴があります。

  • async/await構文を使った直感的な非同期プログラミング
  • クライアント機能（HTTPリクエストの送信）
  • サーバー機能（Webアプリケーションの構築）
  • WebSocketサポート
  • セッション管理機能

これらの機能により、従来では困難だった高性能なWebアプリケーションやAPIクライアントを比較的簡単に構築することができます。

aiohttpのインストールと基本的な使い方

リンドくん

リンドくん

実際にはどうやって使うんですか？インストールから教えてください！

たなべ

たなべ

まずはインストールから始めよう。pipを使えば簡単にインストールできるよ。
そのあと、基本的な使い方を一緒に見ていこう。

aiohttpのインストール方法

aiohttpのインストールは非常に簡単です。以下のコマンドを実行するだけで完了します。

pip install aiohttp
# uvの場合はuv add aiohttp

最もシンプルな使用例

まずは、最もシンプルなGETリクエストから始めてみましょう。

import aiohttp
import asyncio

async def fetch_data():
    async with aiohttp.ClientSession() as session:
        async with session.get('https://httpbin.org/json') as response:
            data = await response.json()
            print(data)

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

このコードの重要なポイントを解説します。

  • async def: 非同期関数を定義するキーワード
  • async with: 非同期コンテキストマネージャー（リソースの自動管理）
  • await: 非同期処理の完了を待つキーワード
  • asyncio.run(): 非同期関数を実行するための関数

複数のリクエストを並行実行する例

aiohttpの真価は、複数のリクエストを同時に処理することにあります。

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 fetch_multiple_urls():
    urls = [
        'https://httpbin.org/delay/1',
        'https://httpbin.org/delay/1', 
        'https://httpbin.org/delay/1',
        'https://httpbin.org/delay/1',
        'https://httpbin.org/delay/1'
    ]
    
    start_time = time.time()
    
    async with aiohttp.ClientSession() as session:
        # 複数のリクエストを同時実行
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
    
    end_time = time.time()
    
    print(f"処理時間: {end_time - start_time:.2f}秒")
    print(f"取得件数: {len(results)}件")

asyncio.run(fetch_multiple_urls())

この例では、通常であれば5秒かかる処理（1秒 x 5回）が、約1秒で完了します。これが非同期処理の威力です！

エラーハンドリングの基本

実際の開発では、エラーハンドリングも重要です。

import aiohttp
import asyncio

async def safe_fetch(session, url):
    try:
        async with session.get(url) as response:
            if response.status == 200:
                return await response.json()
            else:
                print(f"エラー: ステータスコード {response.status}")
                return None
    except aiohttp.ClientError as e:
        print(f"リクエストエラー: {e}")
        return None

async def main():
    async with aiohttp.ClientSession() as session:
        result = await safe_fetch(session, 'https://httpbin.org/json')
        if result:
            print("データ取得成功:", result)

asyncio.run(main())

このように、適切なエラーハンドリングを組み込むことで、安定したプログラムを作成できます。

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

リンドくん

リンドくん

実際にどのくらい速くなるんですか？測定方法とか、さらに速くするコツがあれば教えてください！

たなべ

たなべ

いい視点だね！実際にパフォーマンスを測定して、どれだけ改善されるか数字で見ることは重要なんだ。
さらに高速化するためのテクニックも紹介するよ。

同期処理vs非同期処理の速度比較

実際にrequests（同期）とaiohttp（非同期）の処理速度を比較してみましょう。

import requests
import aiohttp
import asyncio
import time

# 同期処理（requests）
def sync_fetch_multiple():
    urls = ['https://httpbin.org/delay/1' for _ in range(5)]
    start_time = time.time()
    
    results = []
    for url in urls:
        response = requests.get(url)
        results.append(response.status_code)
    
    end_time = time.time()
    return results, end_time - start_time

# 非同期処理（aiohttp）
async def async_fetch_multiple():
    urls = ['https://httpbin.org/delay/1' for _ in range(5)]
    start_time = time.time()
    
    async with aiohttp.ClientSession() as session:
        async def fetch_one(url):
            async with session.get(url) as response:
                return response.status
        
        tasks = [fetch_one(url) for url in urls]
        results = await asyncio.gather(*tasks)
    
    end_time = time.time()
    return results, end_time - start_time

# 比較実行
async def compare_performance():
    print("=== パフォーマンス比較 ===")
    
    # 同期処理の測定
    sync_results, sync_time = sync_fetch_multiple()
    print(f"同期処理: {sync_time:.2f}秒 (結果: {len(sync_results)}件)")
    
    # 非同期処理の測定
    async_results, async_time = await async_fetch_multiple()
    print(f"非同期処理: {async_time:.2f}秒 (結果: {len(async_results)}件)")
    
    # 速度向上率を計算
    improvement = sync_time / async_time
    print(f"速度向上: {improvement:.1f}倍高速!")

asyncio.run(compare_performance())

コネクション管理と最適化

パフォーマンスをさらに向上させるための最適化テクニックをご紹介します。

import aiohttp
import asyncio

# 最適化された設定でのクライアントセッション
async def optimized_requests():
    # コネクションプールの設定
    connector = aiohttp.TCPConnector(
        limit=100,  # 同時接続数の上限
        limit_per_host=30,  # ホスト毎の同時接続数上限
        ttl_dns_cache=300,  # DNS キャッシュの有効期限
        use_dns_cache=True,  # DNS キャッシュを使用
    )
    
    # タイムアウトの設定
    timeout = aiohttp.ClientTimeout(
        total=30,  # 全体のタイムアウト
        connect=10,  # 接続タイムアウト
        sock_read=10  # 読み込みタイムアウト
    )
    
    async with aiohttp.ClientSession(
        connector=connector,
        timeout=timeout
    ) as session:
        
        urls = [f'https://httpbin.org/delay/1?id={i}' for i in range(20)]
        
        async def fetch_with_retry(url, max_retries=3):
            for attempt in range(max_retries):
                try:
                    async with session.get(url) as response:
                        if response.status == 200:
                            return await response.json()
                        else:
                            print(f"HTTP {response.status} for {url}")
                except asyncio.TimeoutError:
                    print(f"タイムアウト (試行 {attempt + 1}/{max_retries}): {url}")
                    if attempt == max_retries - 1:
                        return None
                    await asyncio.sleep(0.5 * (attempt + 1))  # 指数バックオフ
                except Exception as e:
                    print(f"エラー: {e}")
                    return None
            return None
        
        # セマフォを使った同時実行数の制限
        semaphore = asyncio.Semaphore(10)  # 同時に10個まで
        
        async def limited_fetch(url):
            async with semaphore:
                return await fetch_with_retry(url)
        
        start_time = time.time()
        tasks = [limited_fetch(url) for url in urls]
        results = await asyncio.gather(*tasks)
        end_time = time.time()
        
        successful_results = [r for r in results if r is not None]
        print(f"成功: {len(successful_results)}/{len(urls)} 件")
        print(f"処理時間: {end_time - start_time:.2f}秒")

asyncio.run(optimized_requests())

メモリ効率の改善

大量のデータを扱う際のメモリ効率改善テクニックです。

import aiohttp
import asyncio
import weakref

class MemoryEfficientFetcher:
    def __init__(self, max_concurrent=10):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.session = None
    
    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.session.close()
    
    async def fetch_stream(self, url):
        """大きなファイルを効率的にダウンロード"""
        async with self.semaphore:
            async with self.session.get(url) as response:
                if response.status == 200:
                    content = b''
                    async for chunk in response.content.iter_chunked(1024):
                        content += chunk
                        # メモリ使用量を監視しながら処理
                        if len(content) > 10 * 1024 * 1024:  # 10MB制限
                            break
                    return len(content)
                return 0
    
    async def batch_process(self, urls, batch_size=5):
        """バッチ処理でメモリ効率を向上"""
        results = []
        
        for i in range(0, len(urls), batch_size):
            batch = urls[i:i + batch_size]
            batch_tasks = [self.fetch_stream(url) for url in batch]
            batch_results = await asyncio.gather(*batch_tasks, return_exceptions=True)
            results.extend(batch_results)
            
            # バッチ間で少し待機（サーバーへの負荷軽減）
            await asyncio.sleep(0.1)
        
        return results

# 使用例
async def memory_efficient_example():
    urls = [f'https://httpbin.org/bytes/{1000}' for _ in range(50)]
    
    async with MemoryEfficientFetcher(max_concurrent=5) as fetcher:
        results = await fetcher.batch_process(urls, batch_size=10)
        
        successful_downloads = [r for r in results if isinstance(r, int) and r > 0]
        print(f"成功したダウンロード: {len(successful_downloads)}件")
        print(f"合計データ量: {sum(successful_downloads):,} bytes")

asyncio.run(memory_efficient_example())

これらの最適化テクニックを活用することで、さらに効率的で安定したプログラムを作成することができます。
特に大規模なデータ処理や、高負荷な環境での使用を想定している場合は、これらの手法が非常に有効です。

まとめ

リンドくん

リンドくん

aiohttpすごいですね！これまでのプログラムがとても遅く感じちゃいます...

たなべ

たなべ

そうなんだよ！非同期処理を理解すると、プログラミングの可能性がグッと広がるんだ。
最初は少し難しく感じるかもしれないけど、慣れてしまえば手放せなくなるよ。

この記事では、Pythonの非同期HTTPクライアントライブラリ「aiohttp」について、基本概念から実践的な活用方法まで詳しく解説してきました。

aiohttpの主なメリットを改めて整理しましょう。

  • 劇的な速度向上 - 複数のリクエストを並行処理することで、処理時間を大幅に短縮
  • リソース効率 - CPUやメモリを効率的に活用できる
  • スケーラビリティ - 大量のリクエストを扱うアプリケーションに最適
  • モダンな構文 - async/await を使った読みやすいコード

特に、Web API との連携や、大量のデータ収集が必要なプロジェクトでは、その真価を発揮します。
従来の同期処理では数十秒かかっていた処理が、数秒で完了するケースも珍しくありません。

aiohttpをマスターすることで、あなたのPythonプログラミングスキルは確実に次のレベルに到達するでしょう。
最初は新しい概念に戸惑うかもしれませんが、その努力は必ず報われます。

この記事をシェア

関連するコンテンツ
PythonのhttpxライブラリでAPI通信を効率化！初心者向け入門ガイド

PythonのhttpxライブラリでAPI通信を効率化！初心者向け入門ガイド

Python

2025/09/26

PythonのrequestsライブラリでHTTP通信を簡単マスター！初心者向けガイド

PythonのrequestsライブラリでHTTP通信を簡単マスター！初心者向けガイド

Python

2025/09/24

Pythonのmultiprocessingで並列処理入門！初心者でもわかるCPU並列処理の基本とサンプルコード

Pythonのmultiprocessingで並列処理入門！初心者でもわかるCPU並列処理の基本とサンプルコード

Python

2025/09/23

Pythonのthreading入門！マルチスレッド処理で効率的なプログラムを作ろう

Pythonのthreading入門！マルチスレッド処理で効率的なプログラムを作ろう

Python

2025/09/22

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

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

Python

2025/09/19

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

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

Python

2025/09/18