プロキシを使用してレート制限を回避する方法：アーキテクチャ、IPローテーション、開発者向けコード例

レート制限は、スクレイパーがクラッシュしたり、API統合が中断されたり、自動化スクリプトが 429 Too Many Requests ステータスを受け取る最も一般的な理由の一つです。サーバーは、1つのIPからのリクエストが多すぎると判断し、応答を停止します。この記事では、バンやエラーなしでリクエスト制限を回避するためのプロキシインフラを正しく構築する方法を、PythonとNode.jsの実際のコード例を交えて解説します。

レート制限とは何か、そして通常の遅延が役に立たない理由

レート制限（リクエスト頻度制限）は、サーバーを保護するメカニズムで、特定の時間枠内に1つのソースからのリクエスト数を制限します。ソースは通常IPアドレスですが、高度なシステムは認証トークン、User-Agent、クッキー、さらには行動パターンも考慮します。

スクリプトが制限を超えると、サーバーは次のいずれかの応答を返します:

• 429 Too Many Requests — レート制限のための標準HTTPステータス
• 503 Service Unavailable — 時々429の代わりに使用されます
• 403 Forbidden — IPがすでにブロックリストに登録されている場合
• 空の応答またはタイムアウト — 攻撃的なブロック時

多くの開発者の最初の考えは、リクエスト間に time.sleep(1) を追加することです。これは非常に緩やかな制限（例えば、1分間に60リクエスト）の場合にのみ機能します。しかし、実際のシナリオはより複雑です:

マーケットプレイスから100,000の商品カードを収集したり、リアルタイムで価格を監視したりする必要がある場合、遅延は役に立ちません。別のアーキテクチャが必要です。そして、ここでプロキシがオプションではなく、必要不可欠になります。

重要なことは、レート制限がIPアドレスに関連付けられていることを理解することです。異なる100のIPがあれば、実質的に100の独立した「クォータ」があるということです。これがプロキシを通じて制限を回避するための鍵となる原則です。

プロキシがリクエスト制限の問題をどのように解決するか

メカニズムは簡単です: 目的のサーバーへの各リクエストは異なるIPアドレスから送信されます。サーバーの観点から見ると、これは異なるユーザーです。それぞれのクォータはほとんど消費されないため、ブロックは発生しません。

プロキシなしでの作業とプロキシプールを使用した場合の違いを具体的な例で考えてみましょう。サーバーが1つのIPから1分間に10リクエストを許可していると仮定します:

帯域幅のスケーリングに加えて、プロキシはレート制限に関していくつかの利点を提供します:

• セッションの分離 — 1つのIPがブロックされても、他は引き続き動作します
• 地理的分散 — リクエストが異なる地域から行われるため、疑わしさが減少します
• スティッキーセッション — 多段階シナリオ（認証 + アクション）のために1つのIPに「くっつく」機能
• 負荷の制御 — 各IPへのリクエストを正確に調整し、制限を超えないようにできます

あなたのタスクに適したプロキシの種類

すべてのプロキシがレート制限に対して同じように効果的ではありません。タイプの選択は、ターゲットサイト、リクエストのボリューム、および予算によって異なります。3つの主要なタイプを見てみましょう:

レジデンシャルプロキシ

これは実際の家庭ユーザーのIPアドレスです。通常のインターネットトラフィックのように見え、ブロックされることは非常に稀です。レジデンシャルプロキシは、攻撃的な保護を持つサイトに最適な選択です: マーケットプレイス（Wildberries、Ozon）、ソーシャルネットワーク、Cloudflare保護リソース。主な欠点は、データセンターのプロキシと比較して価格が高いことです。

モバイルプロキシ

モバイルオペレーターのIPアドレス（3G/4G/5G）。特徴は、1つのIPが同時に何千人もの実際の加入者によって使用される可能性があるため、そのようなアドレスをブロックすることは非常に望ましくないということです。モバイルプロキシは、レジデンシャルプロキシがブロックされ始める場所で最良の結果を示します — 例えば、高頻度のInstagramスクレイピングや、接続タイプを分析するプラットフォームのAPIとの作業時です。

データセンタープロキシ

サーバーデータセンターからの高速で安価なIP。深刻な保護がないサイトのスクレイピングに最適です: オープンAPI、ニュースアグリゲーター、公開データベース。レート制限のあるタスクには、より多くのプロキシが必要ですが（ブロックリストに載ることが多いため）、適切なローテーションを行えば、大量のリクエストにうまく対処できます。詳細はデータセンタープロキシのページをご覧ください。

IPのローテーション戦略: per-request, sticky sessions, round-robin

プロキシが存在するだけでは問題は解決しません — 正しく管理することが重要です。いくつかのローテーション戦略があり、それぞれが特定のシナリオに適しています。

Per-requestローテーション（各リクエストごとに新しいIP）

各HTTPリクエストは新しいIPアドレスを介して送信されます。これはレート制限を回避するための最も攻撃的な戦略です — サーバーは物理的に1つのIPのカウンターを蓄積する時間がありません。次のようなシナリオに適しています:

• 商品カードのスクレイピング（各カードは別々のリクエスト）
• 検索エンジンからのデータ収集
• セッションを必要としない任意のステートレスリクエスト

スティッキーセッション（セッション中は固定IP）

1つのIPがセッション全体（通常は1〜30分）使用されます。認証が必要なシナリオでは非常に重要です: アカウントにログインし、アクションを実行し、ログアウトします。ステップ間でIPが変更されると、サーバーはセッションを疑わしいものとしてブロックする可能性があります。

リクエスト制限付きのラウンドロビン

最も正確な戦略です。サーバーの制限（例えば、1分間に10リクエスト）を知っており、リクエストをプロキシプールに分配して、各IPがこの閾値を超えないようにします。各IPの最後のリクエストの時間を考慮したキューの実装が必要です。

Pythonのコード例: requests, aiohttp, Scrapy

実践に移りましょう。以下は、最も人気のある3つのPythonツールのための準備されたテンプレートです。

1. requests + 手動プロキシローテーション

最も単純なオプションは、プロキシのリストと各リクエストごとにランダムに選択することです:

import requests
import random
import time

PROXIES = [
    "http://user:[email protected]:8080",
    "http://user:[email protected]:8080",
    "http://user:[email protected]:8080",
    # ... 必要な数を追加
]

def get_random_proxy():
    proxy = random.choice(PROXIES)
    return {"http": proxy, "https": proxy}

def fetch_with_retry(url, max_retries=3):
    for attempt in range(max_retries):
        proxy = get_random_proxy()
        try:
            response = requests.get(
                url,
                proxies=proxy,
                timeout=10,
                headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
            )
            if response.status_code == 429:
                print(f"{proxy}でレート制限に達しました、切り替えます...");
                time.sleep(1)
                continue
            return response
        except requests.RequestException as e:
            print(f"試行 {attempt+1} が失敗しました: {e}")
            time.sleep(2)
    return None

# 使用法
urls = ["https://example.com/item/1", "https://example.com/item/2"]
for url in urls:
    result = fetch_with_retry(url)
    if result:
        print(f"OK: {url} — {len(result.text)} バイト")

2. レート制限を考慮したスマートプロキシプール

より高度なオプションは、各IPの最後の使用時間を追跡し、設定された制限を超えないようにするクラス ProxyPool です:

import requests
import time
from collections import defaultdict
from threading import Lock

class ProxyPool:
    def __init__(self, proxies, rate_limit=10, window=60):
        """
        proxies: 'http://user:pass@host:port'形式の文字列のリスト
        rate_limit: ウィンドウ秒あたりの1つのIPからの最大リクエスト数
        window: 秒単位の時間ウィンドウ
        """
        self.proxies = proxies
        self.rate_limit = rate_limit
        self.window = window
        self.usage = defaultdict(list)  # proxy -> [timestamps]
        self.lock = Lock()

    def get_available_proxy(self):
        now = time.time()
        with self.lock:
            for proxy in self.proxies:
                # 古いタイムスタンプをクリア
                self.usage[proxy] = [
                    t for t in self.usage[proxy]
                    if now - t < self.window
                ]
                if len(self.usage[proxy]) < self.rate_limit:
                    self.usage[proxy].append(now)
                    return {"http": proxy, "https": proxy}
        return None  # すべてのプロキシが制限に達しました

    def fetch(self, url, **kwargs):
        proxy = self.get_available_proxy()
        if proxy is None:
            print("すべてのプロキシがレート制限に達しました、待機中...");
            time.sleep(5)
            return self.fetch(url, **kwargs)
        
        try:
            response = requests.get(url, proxies=proxy, timeout=10, **kwargs)
            return response
        except requests.RequestException as e:
            print(f"リクエストが失敗しました: {e}")
            return None

# 使用法
pool = ProxyPool(
    proxies=[
        "http://user:[email protected]:8080",
        "http://user:[email protected]:8080",
    ],
    rate_limit=10,  # IPあたり1分間に10リクエスト
    window=60
)

for i in range(100):
    r = pool.fetch(f"https://example.com/page/{i}")
    if r:
        print(f"ページ {i}: {r.status_code}")

3. 非同期スクレイピングのためのaiohttp

非同期アプローチを使用すると、スレッドをブロックせずに数十のプロキシを同時に使用できます:

import asyncio
import aiohttp
import itertools

PROXIES = [
    "http://user:[email protected]:8080",
    "http://user:[email protected]:8080",
    "http://user:[email protected]:8080",
]

proxy_cycle = itertools.cycle(PROXIES)

async def fetch(session, url, proxy):
    try:
        async with session.get(
            url,
            proxy=proxy,
            timeout=aiohttp.ClientTimeout(total=10)
        ) as response:
            if response.status == 429:
                await asyncio.sleep(2)
                return None
            return await response.text()
    except Exception as e:
        print(f"エラー: {e}")
        return None

async def main(urls):
    connector = aiohttp.TCPConnector(limit=50)
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [
            fetch(session, url, next(proxy_cycle))
            for url in urls
        ]
        results = await asyncio.gather(*tasks)
        return results

urls = [f"https://example.com/item/{i}" for i in range(200)]
results = asyncio.run(main(urls))
print(f"収集したページ数: {sum(1 for r in results if r is not None)}")

4. middlewareを介したScrapyのローテーション

Scrapyには、既存のソリューションである scrapy-rotating-proxies があります。しかし、自分のミドルウェアを書くこともできます:

# middlewares.py
import random

class RotatingProxyMiddleware:
    def __init__(self, proxies):
        self.proxies = proxies

    @classmethod
    def from_crawler(cls, crawler):
        return cls(proxies=crawler.settings.getlist("PROXY_LIST"))

    def process_request(self, request, spider):
        proxy = random.choice(self.proxies)
        request.meta["proxy"] = proxy

    def process_response(self, request, response, spider):
        if response.status == 429:
            spider.logger.warning(f"レート制限に達しました, プロキシ: {request.meta.get('proxy')}")
            # 問題のあるプロキシを除外するロジックを追加できます
        return response

# settings.py
PROXY_LIST = [
    "http://user:[email protected]:8080",
    "http://user:[email protected]:8080",
]
DOWNLOADER_MIDDLEWARES = {
    "myproject.middlewares.RotatingProxyMiddleware": 350,
}
AUTOTHROTTLE_ENABLED = True
AUTOTHROTTLE_TARGET_CONCURRENCY = 10

Node.jsのコード例: axios, got, Puppeteer

Node.jsは、ブラウザの自動化やAPIとの作業に人気の選択肢です。以下は、プロキシを使用するための準備されたパターンです。

1. axiosによるプロキシのローテーション

const axios = require('axios');
const { HttpsProxyAgent } = require('https-proxy-agent');

const proxies = [
  'http://user:[email protected]:8080',
  'http://user:[email protected]:8080',
  'http://user:[email protected]:8080',
];

let proxyIndex = 0;

function getNextProxy() {
  const proxy = proxies[proxyIndex % proxies.length];
  proxyIndex++;
  return proxy;
}

async function fetchWithProxy(url, retries = 3) {
  for (let i = 0; i < retries; i++) {
    const proxyUrl = getNextProxy();
    const agent = new HttpsProxyAgent(proxyUrl);
    
    try {
      const response = await axios.get(url, {
        httpsAgent: agent,
        httpAgent: agent,
        timeout: 10000,
        headers: {
          'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)',
        },
      });
      return response.data;
    } catch (error) {
      if (error.response?.status === 429) {
        console.log(`レート制限に達しました、プロキシを切り替えます...`);
        await new Promise(r => setTimeout(r, 1000));
        continue;
      }
      console.error(`試行 ${i + 1} が失敗しました:`, error.message);
    }
  }
  return null;
}

// 使用法
(async () => {
  const urls = Array.from({length: 50}, (_, i) => `https://example.com/item/${i}`);
  
  const results = await Promise.allSettled(
    urls.map(url => fetchWithProxy(url))
  );
  
  const successful = results.filter(r => r.status === 'fulfilled' && r.value).length;
  console.log(`成功: ${successful}/${urls.length}`);
})();

2. Puppeteerによるプロキシとレート制限の回避

JavaScriptレンダリングとCloudflare保護のあるサイトには、ヘッドレスブラウザが必要です:

const puppeteer = require('puppeteer');

const proxies = [
  'proxy1.example.com:8080',
  'proxy2.example.com:8080',
];

async function scrapeWithProxy(url, proxyHost) {
  const browser = await puppeteer.launch({
    args: [
      `--proxy-server=${proxyHost}`,
      '--no-sandbox',
      '--disable-setuid-sandbox',
    ],
    headless: true,
  });

  const page = await browser.newPage();
  
  // プロキシの認証
  await page.authenticate({
    username: 'user',
    password: 'pass',
  });

  // リアルなUser-Agentを設定
  await page.setUserAgent(
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
  );

  try {
    await page.goto(url, { waitUntil: 'networkidle2', timeout: 30000 });
    
    // レート制限を確認
    const status = await page.evaluate(() => document.title);
    if (status.includes('429') || status.includes('Too Many')) {
      console.log('レート制限に達しました、プロキシを切り替える必要があります');
      return null;
    }
    
    const data = await page.evaluate(() => {
      return document.querySelector('.price')?.textContent || null;
    });
    
    return data;
  } finally {
    await browser.close();
  }
}

// タスクごとのローテーション
(async () => {
  const urls = ['https://example.com/product/1', 'https://example.com/product/2'];
  
  for (let i = 0; i < urls.length; i++) {
    const proxy = proxies[i % proxies.length];
    const result = await scrapeWithProxy(urls[i], proxy);
    console.log(`${urls[i]}: ${result}`);
    await new Promise(r => setTimeout(r, 500)); // 小さな遅延
  }
})();

高度なテクニック: ヘッダー, フィンガープリンティング, Cloudflareの回避

IPの変更は必要ですが、常に十分ではありません。現代の保護システムは、リクエストの数十のパラメータを分析します。考慮すべき他の要素を見てみましょう。

HTTPヘッダー: 最小限の必須セット

正常なヘッダーなしのリクエストは、IPを変更してもボットのように見えます。常に追加してください:

headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
    "Accept-Language": "ja-JP,ja;q=0.9,en-US;q=0.8,en;q=0.7",
    "Accept-Encoding": "gzip, deflate, br",
    "Connection": "keep-alive",
    "Upgrade-Insecure-Requests": "1",
    "Sec-Fetch-Dest": "document",
    "Sec-Fetch-Mode": "navigate",
    "Sec-Fetch-Site": "none",
    "Cache-Control": "max-age=0",
}

Retry-Afterヘッダーの処理

429の応答時、サーバーは待機すべき時間を示すことがよくあります。このヘッダーを正しく処理することで、無駄にリクエストを消費せずに済みます:

def handle_rate_limit(response):
    if response.status_code == 429:
        retry_after = response.headers.get("Retry-After")
        if retry_after:
            wait_time = int(retry_after)
            print(f"レート制限に達しました。{wait_time}秒待機します...");
            time.sleep(wait_time + 1)  # +1秒のバッファ
        else:
            # ヘッダーがない場合は指数的な遅延
            time.sleep(min(2 ** attempt, 60))
        return True
    return False

TLSフィンガープリンティングとその回避方法

高度なシステム（Cloudflare、Akamai、PerimeterX）は、TLS接続のユニークな「フィンガープリント」を分析します。標準ライブラリ requests は簡単に認識できるフィンガープリンティングを持っています。解決策:

• curl_cffi（Python） — TLSレベルでChrome/Firefoxのフィンガープリンティングをエミュレートします
• tls-client（Go/Python） — 異なるブラウザプロファイルをサポートする同様のツール
• Playwright/Puppeteer — 実際のブラウザで、デフォルトで理想的なフィンガープリンティング

# pip install curl-cffi
from curl_cffi import requests as cffi_requests

response = cffi_requests.get(
    "https://cloudflare-protected-site.com/api/data",
    impersonate="chrome120",  # Chrome 120をエミュレート
    proxies={"https": "http://user:[email protected]:8080"}
)
print(response.json())

クッキーとセッションの管理

サイトがセッションを追跡するためにクッキーを使用している場合、クッキーを変更せずにIPを変更しても意味がありません。プロキシを切り替える際は、常に新しいセッションを作成してください:

import requests

def create_fresh_session(proxy_url):
    """各プロキシのために新しいクッキーで新しいセッションを作成します"""
    session = requests.Session()
    session.proxies = {"http": proxy_url, "https": proxy_url}
    session.headers.update({
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)...",
    })
    # 前のセッションからクッキーは移行されません
    return session

# 各新しいIPごとに新しいセッション
for proxy in proxies:
    session = create_fresh_session(proxy)
    response = session.get("https://example.com/protected-page")
    # 応答を処理します...

プロキシとレート制限に関する一般的な間違い

正しく設定されたプロキシを使用しても、開発者は定期的に同じ過ちを繰り返します。ここでは、最も一般的な間違いとそれを避ける方法を紹介します。

エラー1: 「死んだ」プロキシの使用

プールに追加する前にプロキシを常に確認し、作業中も定期的に確認してください。1つの動作しないプロキシがループ内にあると、リクエストとタイムアウトが失われます:

def check_proxy(proxy_url, test_url="https://httpbin.org/ip", timeout=5):
    try:
        r = requests.get(
            test_url,
            proxies={"http": proxy_url, "https": proxy_url},
            timeout=timeout
        )
        return r.status_code == 200
    except:
        return False

# スタート前に動作するプロキシをフィルタリング
working_proxies = [p for p in PROXIES if check_proxy(p)]
print(f"動作するプロキシ: {len(working_proxies)}/{len(PROXIES)}")

エラー2: プロトコルの種類を無視する

HTTPプロキシは、HTTPSトラフィックを直接プロキシできません（CONNECTを介してのみ）。SOCKS5プロキシはトランスポートレベルで動作し、任意のプロトコルをサポートします。ほとんどの現代のサイトでは、SOCKS5またはHTTPSプロキシを使用してください:

# requestsでのSOCKS5プロキシ（pip install requests[socks]が必要）
proxies = {
    "http": "socks5://user:[email protected]:1080",
    "https": "socks5://user:[email protected]:1080",
}

# HTTPSプロキシ
proxies = {
    "http": "https://user:[email protected]:8080",
    "https": "https://user:[email protected]:8080",
}

エラー3: 指数的バックオフの欠如

429の後にすぐにリクエストを再試行すると、状況が悪化するだけです。正しい戦略は、ジッター（ランダムな偏差）を伴う指数的な遅延です:

import random

def exponential_backoff(attempt, base=1, max_wait=60):
    """
    attempt: 試行番号（0から始まる）
    base: 基本的な遅延（秒単位）
    max_wait: 最大遅延
    """
    wait = min(base * (2 ** attempt), max_wait)
    # ジッター ±25%でthundering herdを防止
    jitter = wait * 0.25 * random.uniform(-1, 1)
    return wait + jitter

# リトライロジックでの使用
for attempt in range(5):
    response = requests.get(url, proxies=proxy)
    if response.status_code == 429:
        wait = exponential_backoff(attempt)
        print(f"レート制限に達しました。{wait:.1f}秒待機します（試行 {attempt+1}）")
        time.sleep(wait)
    else:
        break

エラー4: すべてのプロキシに1つのスレッド

50のプロキシがある場合でも、実行スレッドが1つしかないと、同時に1つのプロキシしか使用できません。ThreadPoolExecutorや非同期アプローチを使用して、プール全体を並行して使用してください:

from concurrent.futures import ThreadPoolExecutor, as_completed

def fetch_url(args):
    url, proxy = args
    try:
        r = requests.get(url, proxies={"https": proxy}, timeout=10)
        return url, r.status_code, len(r.text)
    except Exception as e:
        return url, None, str(e)

# すべてのプロキシを並行して使用
tasks = [(url, proxies[i % len(proxies)]) for i, url in enumerate(urls)]

with ThreadPoolExecutor(max_workers=len(proxies)) as executor:
    futures = {executor.submit(fetch_url, task): task for task in tasks}
    for future in as_completed(futures):
        url, status, size = future.result()
        print(f"{url}: {status} ({size})")

結論と推奨事項

レート制限は、体系的にアプローチすれば解決可能な問題です。このガイドからの重要な結論は次のとおりです:

• プロキシプール、単一のプロキシではなく — 真剣な作業のための最小単位です。プロキシの数は、次の公式で決定されます: 目標速度 ÷ IPあたりのサーバー制限。
• ローテーション戦略が重要 — statelessリクエストにはper-request、認証シナリオにはスティッキーセッション。
• IPは唯一のパラメータではない — ヘッダー、クッキー、TLSフィンガープリンティング、行動パターンも保護システムによって分析されます。
• 429を正しく処理する — 指数的バックオフ、Retry-Afterヘッダー、ブロック時のプロキシ切り替え。
• プロキシの種類は目的に応じて異なる — オープンAPIにはデータセンタープロキシ、マーケットプレイスにはレジデンシャルプロキシ、最大の保護が必要な場合はモバイルプロキシを使用します。

マーケットプレイス（Wildberries、Ozon）のスクレイピング、保護されたAPIからのデータ収集、高速での自動化に取り組んでいる場合は、レジデンシャルプロキシから始めることをお勧めします: 匿名性と速度の最適なバランスを提供し、そのIPアドレスはほとんどブロックリストに載ることがありません。高頻度のリクエストでブロックに対する最大の耐性が必要なタスクには、モバイルプロキシを検討する価値があります — そのIPは何千人もの実際のユーザーによって共有されており、どのサイトにとってもブロックは非常に望ましくありません。