プロキシ経由でのタイムアウトエラーを修正する方法

リクエストがハング状態になり、スクリプトがTimeoutErrorエラーで落ちてデータが取得できない。こんな経験ありませんか？プロキシ経由のタイムアウトエラーはスクレイピングと自動化の最も一般的な問題の1つです。原因を分析し、具体的なソリューションを提供します。

タイムアウトエラーが発生する理由

タイムアウトは1つの問題ではなく、症状です。治療する前に、原因を理解する必要があります：

プロキシサーバーが遅い。過負荷のサーバーまたは地理的に遠いプロキシは、各リクエストに遅延を追加します。タイムアウトが10秒で、プロキシが12秒で応答する場合、エラーが発生します。

ターゲットサイト側でのブロック。サイトは明示的な拒否の代わりに、疑わしいリクエストを意図的に「ハング」させることがあります。これはボット対策の戦術で、接続を無期限に開いたままにします。

DNS問題。プロキシはドメインを解決する必要があります。プロキシのDNSサーバーが遅いか利用不可の場合、リクエストは接続段階でハング状態になります。

タイムアウト設定が不正。すべてに1つの共通タイムアウトを使用するのは一般的な誤りです。接続タイムアウトと読み取りタイムアウトは異なるもので、別々に設定する必要があります。

ネットワークの問題。パケット損失、プロキシの不安定な接続、ルーティングの問題など、すべてがタイムアウトにつながります。

タイムアウトの種類と設定

ほとんどのHTTPライブラリは複数のタイムアウトタイプをサポートしています。それらの違いを理解することが、正しい設定の鍵です。

接続タイムアウト

プロキシとターゲットサーバーとのTCP接続を確立する時間。プロキシが利用不可またはサーバーが応答しない場合、このタイムアウトが発動します。推奨値：5～10秒。

読み取りタイムアウト

接続確立後のデータ待機時間。サーバーは接続しましたが沈黙している場合、読み取りタイムアウトが発動します。通常のページの場合：15～30秒。重いAPIの場合：60秒以上。

合計タイムアウト

開始から終了までのリクエスト全体の時間。ハング接続からの保護。通常：接続 + 読み取り + 余裕。

Pythonのrequestsライブラリでの設定例：

import requests

proxies = {
    "http": "http://user:pass@proxy.example.com:8080",
    "https": "http://user:pass@proxy.example.com:8080"
}

# タプル：(接続タイムアウト、読み取りタイムアウト)
timeout = (10, 30)

try:
    response = requests.get(
        "https://target-site.com/api/data",
        proxies=proxies,
        timeout=timeout
    )
except requests.exceptions.ConnectTimeout:
    print("プロキシまたはサーバーへの接続に失敗しました")
except requests.exceptions.ReadTimeout:
    print("サーバーが時間内にデータを送信しませんでした")

aiohttp（非同期Python）の場合：

import aiohttp
import asyncio

async def fetch_with_timeout():
    timeout = aiohttp.ClientTimeout(
        total=60,      # 合計タイムアウト
        connect=10,    # 接続用
        sock_read=30   # データ読み取り用
    )
    
    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.get(
            "https://target-site.com/api/data",
            proxy="http://user:pass@proxy.example.com:8080"
        ) as response:
            return await response.text()

リトライロジック：正しいアプローチ

タイムアウトは常に致命的なエラーではありません。多くの場合、再試行は成功します。ただし、リトライは慎重に行う必要があります。

指数バックオフ

一時停止なしで再試行リクエストでサーバーを叩かないでください。指数バックオフを使用してください：次の試行ごとに増加する遅延があります。

import requests
import time
import random

def fetch_with_retry(url, proxies, max_retries=3):
    """リトライと指数バックオフを伴うリクエスト"""
    
    for attempt in range(max_retries):
        try:
            response = requests.get(
                url,
                proxies=proxies,
                timeout=(10, 30)
            )
            response.raise_for_status()
            return response
            
        except (requests.exceptions.Timeout, 
                requests.exceptions.ConnectionError) as e:
            
            if attempt == max_retries - 1:
                raise  # 最後の試行 - エラーをスロー
            
            # 指数バックオフ：1s、2s、4s...
            # + リクエストの波を作らないようにランダムなジッターを追加
            delay = (2 ** attempt) + random.uniform(0, 1)
            print(f"試行 {attempt + 1} が失敗しました：{e}")
            print(f"{delay:.1f}秒後に再試行します...")
            time.sleep(delay)

tenacityライブラリ

本番環境のコードには、既成のソリューションを使用する方が便利です：

from tenacity import retry, stop_after_attempt, wait_exponential
from tenacity import retry_if_exception_type
import requests

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
    retry=retry_if_exception_type((
        requests.exceptions.Timeout,
        requests.exceptions.ConnectionError
    ))
)
def fetch_data(url, proxies):
    response = requests.get(url, proxies=proxies, timeout=(10, 30))
    response.raise_for_status()
    return response.json()

タイムアウト時のプロキシローテーション

1つのプロキシが常にタイムアウトを与える場合、問題はそのプロキシにあります。論理的なソリューション：別のプロキシに切り替えます。

import requests
from collections import deque
from dataclasses import dataclass, field
from typing import Optional
import time

@dataclass
class ProxyManager:
    """失敗した試行を追跡するプロキシマネージャー"""
    
    proxies: list
    max_failures: int = 3
    cooldown_seconds: int = 300
    _failures: dict = field(default_factory=dict)
    _cooldown_until: dict = field(default_factory=dict)
    
    def get_proxy(self) -> Optional[str]:
        """動作中のプロキシを取得"""
        current_time = time.time()
        
        for proxy in self.proxies:
            # クールダウン中のプロキシをスキップ
            if self._cooldown_until.get(proxy, 0) > current_time:
                continue
            return proxy
        
        return None  # すべてのプロキシがクールダウン中

    def report_failure(self, proxy: str):
        """失敗したリクエストを報告"""
        self._failures[proxy] = self._failures.get(proxy, 0) + 1
        
        if self._failures[proxy] >= self.max_failures:
            # プロキシをクールダウン状態に設定
            self._cooldown_until[proxy] = time.time() + self.cooldown_seconds
            self._failures[proxy] = 0
            print(f"プロキシ {proxy} がクールダウン状態に設定されました")
    
    def report_success(self, proxy: str):
        """成功時にエラーカウンターをリセット"""
        self._failures[proxy] = 0


def fetch_with_rotation(url, proxy_manager, max_attempts=5):
    """エラー時に自動的にプロキシを切り替えるリクエスト"""
    
    for attempt in range(max_attempts):
        proxy = proxy_manager.get_proxy()
        
        if not proxy:
            raise Exception("利用可能なプロキシがありません")
        
        proxies = {"http": proxy, "https": proxy}
        
        try:
            response = requests.get(url, proxies=proxies, timeout=(10, 30))
            response.raise_for_status()
            proxy_manager.report_success(proxy)
            return response
            
        except (requests.exceptions.Timeout, 
                requests.exceptions.ConnectionError):
            proxy_manager.report_failure(proxy)
            print(f"{proxy}経由のタイムアウト、別のプロキシを試します...")
            continue
    
    raise Exception(f"{max_attempts}回の試行後、データを取得できませんでした")

レジデンシャルプロキシを自動ローテーション機能で使用する場合、このロジックは簡略化されます。プロバイダーが各リクエストごと、または指定された間隔でIPを自動的に切り替えます。

タイムアウト制御を伴う非同期リクエスト

大規模なスクレイピングでは、同期リクエストは非効率です。非同期アプローチにより、数百のURLを並行処理できますが、タイムアウトの慎重な処理が必要です。

import aiohttp
import asyncio
from typing import List, Tuple

async def fetch_one(
    session: aiohttp.ClientSession, 
    url: str,
    semaphore: asyncio.Semaphore
) -> Tuple[str, str | None, str | None]:
    """タイムアウト処理を伴う1つのURLの読み込み"""
    
    async with semaphore:  # 並行性を制限
        try:
            async with session.get(url) as response:
                content = await response.text()
                return (url, content, None)
                
        except asyncio.TimeoutError:
            return (url, None, "timeout")
        except aiohttp.ClientError as e:
            return (url, None, str(e))


async def fetch_all(
    urls: List[str],
    proxy: str,
    max_concurrent: int = 10
) -> List[Tuple[str, str | None, str | None]]:
    """タイムアウトと並行性制御を伴う大量読み込み"""
    
    timeout = aiohttp.ClientTimeout(total=45, connect=10, sock_read=30)
    semaphore = asyncio.Semaphore(max_concurrent)
    
    connector = aiohttp.TCPConnector(
        limit=max_concurrent,
        limit_per_host=5  # 1つのホストに対して最大5接続
    )
    
    async with aiohttp.ClientSession(
        timeout=timeout,
        connector=connector
    ) as session:
        # すべてのリクエストにプロキシを設定
        tasks = [
            fetch_one(session, url, semaphore) 
            for url in urls
        ]
        results = await asyncio.gather(*tasks)
    
    # 統計情報
    success = sum(1 for _, content, _ in results if content)
    timeouts = sum(1 for _, _, error in results if error == "timeout")
    print(f"成功：{success}、タイムアウト：{timeouts}")
    
    return results


# 使用方法
async def main():
    urls = [f"https://example.com/page/{i}" for i in range(100)]
    results = await fetch_all(
        urls, 
        proxy="http://user:pass@proxy.example.com:8080",
        max_concurrent=10
    )

asyncio.run(main())

重要：並行性を高すぎる値に設定しないでください。1つのプロキシ経由で50～100の同時リクエストは既に多いです。複数のプロキシで10～20の方が良いです。

診断：原因を特定する方法

設定を変更する前に、問題の原因を特定してください。

ステップ1：プロキシを直接確認

# curlを使用した簡単なテストと時間測定
curl -x http://user:pass@proxy:8080 \
     -w "Connect: %{time_connect}s\nTotal: %{time_total}s\n" \
     -o /dev/null -s \
     https://httpbin.org/get

time_connectが5秒以上の場合、プロキシまたはそこへのネットワークに問題があります。

ステップ2：直接リクエストと比較

import requests
import time

def measure_request(url, proxies=None):
    start = time.time()
    try:
        r = requests.get(url, proxies=proxies, timeout=30)
        elapsed = time.time() - start
        return f"OK: {elapsed:.2f}s, status: {r.status_code}"
    except Exception as e:
        elapsed = time.time() - start
        return f"FAIL: {elapsed:.2f}s, error: {type(e).__name__}"

url = "https://target-site.com"
proxy = {"http": "http://proxy:8080", "https": "http://proxy:8080"}

print("直接：", measure_request(url))
print("プロキシ経由：", measure_request(url, proxy))

ステップ3：異なるタイプのプロキシを確認

タイムアウトはプロキシのタイプに依存する場合があります：

ステップ4：詳細をログに記録

import logging
import requests
from requests.adapters import HTTPAdapter

# デバッグログを有効化
logging.basicConfig(level=logging.DEBUG)
logging.getLogger("urllib3").setLevel(logging.DEBUG)

# これでリクエストのすべてのステージが表示されます：
# - DNS解決
# - 接続確立
# - リクエスト送信
# - レスポンス受信

タイムアウトエラー解決チェックリスト

タイムアウトが発生した場合の簡潔なアクション手順：

1. タイムアウトのタイプを特定～接続か読み取りか？これらは異なる問題です。
2. プロキシを個別に確認～実際に動作していますか？遅延はどのくらいですか？
3. タイムアウトを増加～値がプロキシのタイプに対して攻撃的すぎる可能性があります。
4. バックオフを伴うリトライを追加～単一のタイムアウトは正常で、重要なのは回復力です。
5. ローテーションを設定～問題が発生したときに別のプロキシに自動的に切り替えます。
6. 並行性を制限～同時リクエストが多すぎるとプロキシが過負荷になります。
7. ターゲットサイトを確認～リクエストをブロックまたはスロットルしている可能性があります。

結論

プロキシ経由のタイムアウトエラーは解決可能な問題です。ほとんどの場合、プロキシのタイプに合わせてタイムアウトを正しく設定し、リトライロジックを追加し、障害時のローテーションを実装するだけで十分です。高い安定性要件のタスクには、自動ローテーション機能を備えたレジデンシャルプロキシを使用してください。詳細はproxycove.comをご覧ください。