スクレイピングレート制限の解説

スクレイピング率制限とは何ですか?

レート制限は、Webサイトがどの単一クライアントが要求をすることができる速度を制御するために構築する見えない壁です。 あなたがサイトを攻撃的に掻くとき、あなたはこれらの壁に当たる - そして、結果は一時的な減速から恒久的なIP禁止の範囲の範囲です。 速度制限がどのように機能するかを理解し、それらがどのように検出するか、そしてそれらの下で滞在する方法は、データを確実に配信するスクレーパーの構築の基礎です。

このガイドは、速度制限、検出信号のウェブサイトの使用、およびあなたのスクレーパーがスムーズに実行し続ける適応回転のための実用的な戦略の背後にあるメカニックについて説明します。

プロキシによるスクレイピングの広範な概要については、当社のを参照してください。 ウェブスクレイピングプロキシの完全なガイド. 一般にブロックを避けるため、読み込み ブロックせずにウェブサイトをScrapeする方法. .

レート制限の仕組み

ウェブサイトは、異なる検出の粒度で、複数のレイヤーでレート制限を実装します。

レイヤー1:IPベースのレート制限

最も一般的なアプローチ。 サーバは、タイムウィンドウ内のIPアドレスごとのリクエストを追跡します。 しきい値を引き出すと、HTTP 429(Too Many Requests)または503レスポンスが受けられます。

# Typical rate limit behavior
Request 1-50:    HTTP 200 (normal)
Request 51:      HTTP 429 (rate limited)
Wait 60 seconds...
Request 52:      HTTP 200 (reset)

レイヤー2:セッション/クッキーベースの制限

セッションまたはブラウザクッキーごとの頻度要求を追跡します。 IP を回転しても、同じセッショントークンがサーバーのスピードを打つと制限がトリガーされます。

レイヤー3:アカウントベースの制限

ログインが必要なサイトの場合、IPに関係なく、制限はユーザーアカウントに結び付けられます。 API と SaaS プラットフォームで共通。

層4:行動分析

Cloudflare、PerimeterX、Akamaiなどの高度なシステムは、動作パターンを分析します。リクエストタイミング、ナビゲーションフロー、マウスの動き(ブラウザのコンテキスト)。 シンプルなカウンターに頼らないため、バイパスするのが一番難しいレイヤーです。

共通の率の限界の検出信号

ウェブサイトは複数の信号を同時に使用し、自動スクレイピングを検出します。

ガイドの検出メカニズムについて詳しく知る アンチボットシステムがプロキシを検出する方法. .

シグナルレート制限のHTTPレスポンスコード

どの HTTP コードがレートの制限を示すかを知ると、適切なリトライロジックを作成するのに役立ちます。

戦略1: 尊敬すべき脅威

最も単純なアプローチ — ターゲットが許すものの下のリクエスト率を十分に保ちます。 これは、より少ない故障、無駄な帯域幅、およびより持続可能なスクレイピングを意味します。

import requests
import time
import random
PROXY = "http://USERNAME:PASSWORD@gate.proxyhat.com:8080"
def respectful_scrape(urls: list[str], rpm_limit: int = 10) -> list[str]:
    """Scrape URLs while respecting a requests-per-minute limit."""
    delay = 60.0 / rpm_limit
    results = []
    for url in urls:
        try:
            resp = requests.get(
                url,
                proxies={"http": PROXY, "https": PROXY},
                timeout=30
            )
            results.append(resp.text if resp.status_code == 200 else None)
        except requests.RequestException:
            results.append(None)
        # Add delay with random jitter (±30%) to look less robotic
        jitter = delay * random.uniform(0.7, 1.3)
        time.sleep(jitter)
    return results

戦略2:適応性のある脅威

固定速度の代わりに、受信した応答に基づいて速度を動的に調整します。 警告サインを見ると、すべてが動作するとスピードアップします。

Pythonの実装

import requests
import time
import random
from dataclasses import dataclass, field
PROXY = "http://USERNAME:PASSWORD@gate.proxyhat.com:8080"
@dataclass
class AdaptiveThrottle:
    """Automatically adjusts request rate based on server responses."""
    base_delay: float = 2.0      # seconds between requests
    min_delay: float = 0.5
    max_delay: float = 30.0
    current_delay: float = 2.0
    success_streak: int = 0
    warning_codes: set = field(default_factory=lambda: {429, 403, 503})
    def on_success(self):
        self.success_streak += 1
        # Speed up after 10 consecutive successes
        if self.success_streak >= 10:
            self.current_delay = max(self.current_delay * 0.85, self.min_delay)
            self.success_streak = 0
    def on_rate_limit(self):
        self.success_streak = 0
        # Double the delay on rate limit
        self.current_delay = min(self.current_delay * 2.0, self.max_delay)
    def on_block(self):
        self.success_streak = 0
        # Aggressive backoff on block
        self.current_delay = min(self.current_delay * 3.0, self.max_delay)
    def wait(self):
        jitter = self.current_delay * random.uniform(0.7, 1.3)
        time.sleep(jitter)
def scrape_adaptive(urls: list[str]) -> list[dict]:
    throttle = AdaptiveThrottle()
    results = []
    for url in urls:
        try:
            resp = requests.get(
                url,
                proxies={"http": PROXY, "https": PROXY},
                timeout=30
            )
            if resp.status_code == 200:
                throttle.on_success()
                results.append({"url": url, "status": 200, "body": resp.text})
            elif resp.status_code == 429:
                throttle.on_rate_limit()
                # Check Retry-After header
                retry_after = int(resp.headers.get("Retry-After", 0))
                if retry_after:
                    time.sleep(retry_after)
                results.append({"url": url, "status": 429, "body": None})
            elif resp.status_code == 403:
                throttle.on_block()
                results.append({"url": url, "status": 403, "body": None})
            else:
                results.append({"url": url, "status": resp.status_code, "body": resp.text})
        except requests.RequestException as e:
            throttle.on_block()
            results.append({"url": url, "status": 0, "error": str(e)})
        throttle.wait()
        print(f"Current delay: {throttle.current_delay:.1f}s")
    return results

Node.js 実装

const HttpsProxyAgent = require('https-proxy-agent');
const fetch = require('node-fetch');
class AdaptiveThrottle {
  constructor() {
    this.currentDelay = 2000; // ms
    this.minDelay = 500;
    this.maxDelay = 30000;
    this.successStreak = 0;
  }
  onSuccess() {
    this.successStreak++;
    if (this.successStreak >= 10) {
      this.currentDelay = Math.max(this.currentDelay * 0.85, this.minDelay);
      this.successStreak = 0;
    }
  }
  onRateLimit() {
    this.successStreak = 0;
    this.currentDelay = Math.min(this.currentDelay * 2, this.maxDelay);
  }
  onBlock() {
    this.successStreak = 0;
    this.currentDelay = Math.min(this.currentDelay * 3, this.maxDelay);
  }
  async wait() {
    const jitter = this.currentDelay * (0.7 + Math.random() * 0.6);
    return new Promise(resolve => setTimeout(resolve, jitter));
  }
}
async function scrapeAdaptive(urls) {
  const throttle = new AdaptiveThrottle();
  const agent = new HttpsProxyAgent('http://USERNAME:PASSWORD@gate.proxyhat.com:8080');
  const results = [];
  for (const url of urls) {
    try {
      const res = await fetch(url, { agent, timeout: 30000 });
      if (res.ok) {
        throttle.onSuccess();
        results.push({ url, status: res.status, body: await res.text() });
      } else if (res.status === 429) {
        throttle.onRateLimit();
        const retryAfter = parseInt(res.headers.get('retry-after') || '0');
        if (retryAfter) await new Promise(r => setTimeout(r, retryAfter * 1000));
        results.push({ url, status: 429, body: null });
      } else if (res.status === 403) {
        throttle.onBlock();
        results.push({ url, status: 403, body: null });
      }
    } catch (err) {
      throttle.onBlock();
      results.push({ url, status: 0, error: err.message });
    }
    await throttle.wait();
    console.log(`Current delay: ${throttle.currentDelay.toFixed(0)}ms`);
  }
  return results;
}

戦略3:分配率制限

複数のスクレーパーインスタンスを並行して実行する場合、すべてのワーカー間でのレート制限を調整します。 協調なし、各ワーカーは独自の限界を尊重しますが、組み合わせたトラフィックはターゲットを圧倒します。

import requests
import time
import threading
class DistributedRateLimiter:
    """Thread-safe rate limiter for multiple scraper workers."""
    def __init__(self, max_rpm: int):
        self.min_interval = 60.0 / max_rpm
        self.lock = threading.Lock()
        self.last_request_time = 0.0
    def acquire(self):
        """Block until it is safe to make the next request."""
        with self.lock:
            now = time.time()
            elapsed = now - self.last_request_time
            if elapsed < self.min_interval:
                time.sleep(self.min_interval - elapsed)
            self.last_request_time = time.time()
# Shared limiter across all threads
limiter = DistributedRateLimiter(max_rpm=30)
PROXY = "http://USERNAME:PASSWORD@gate.proxyhat.com:8080"
def worker(urls: list[str], results: list):
    for url in urls:
        limiter.acquire()
        try:
            resp = requests.get(
                url,
                proxies={"http": PROXY, "https": PROXY},
                timeout=30
            )
            results.append({"url": url, "status": resp.status_code})
        except Exception as e:
            results.append({"url": url, "error": str(e)})

戦略4:優先事項を要求する

複雑なスクレイピングプロジェクトでは、ターゲットドメインごとのレート制限を管理する優先キューを使用します。

import requests
import time
import heapq
import threading
from collections import defaultdict
PROXY = "http://USERNAME:PASSWORD@gate.proxyhat.com:8080"
class DomainRateLimiter:
    """Per-domain rate limiting with priority queue."""
    def __init__(self, default_rpm: int = 10):
        self.default_rpm = default_rpm
        self.domain_limits = {}          # domain -> max RPM
        self.domain_last = defaultdict(float)  # domain -> last request time
        self.lock = threading.Lock()
    def set_limit(self, domain: str, rpm: int):
        self.domain_limits[domain] = rpm
    def wait_for_domain(self, domain: str):
        rpm = self.domain_limits.get(domain, self.default_rpm)
        min_interval = 60.0 / rpm
        with self.lock:
            now = time.time()
            elapsed = now - self.domain_last[domain]
            if elapsed < min_interval:
                time.sleep(min_interval - elapsed)
            self.domain_last[domain] = time.time()
# Configure per-domain limits
limiter = DomainRateLimiter(default_rpm=10)
limiter.set_limit("amazon.com", 3)        # Very conservative for Amazon
limiter.set_limit("example.com", 30)      # Lenient for simple sites
limiter.set_limit("google.com", 5)        # Moderate for Google

読書 Robots.txt レートヒント

多くのサイトがrobots.txtでクロール設定を公開しています。 ザ・オブ・ザ・ Crawl-delay ディレクティブは、リクエスト間の最小秒を指示します。

import requests
from urllib.parse import urlparse
from urllib.robotparser import RobotFileParser
def get_crawl_delay(base_url: str, user_agent: str = "*") -> float | None:
    """Extract Crawl-delay from robots.txt."""
    parsed = urlparse(base_url)
    robots_url = f"{parsed.scheme}://{parsed.netloc}/robots.txt"
    try:
        resp = requests.get(robots_url, timeout=10)
        if resp.status_code != 200:
            return None
        rp = RobotFileParser()
        rp.parse(resp.text.splitlines())
        delay = rp.crawl_delay(user_agent)
        return delay
    except Exception:
        return None
# Check before scraping
delay = get_crawl_delay("https://example.com")
if delay:
    print(f"Site requests {delay}s between requests")
else:
    print("No crawl-delay specified")

共通の率の限界の間違い

• 429応答を無視します。 多くのスクレーパーは、すべての非-200応答を同じ扱います。 429 は、Retry-After ヘッダーを使用して、何が起こったのかを正確に教えてくれます。
• ジッタなしの遅延を修正しました。 2.000秒ごとに正確にリクエストすると、ロボットが見えます。 遅延にランダムなバリエーション(文字)を追加します。
• 並列作業員の調整は行いません。 それぞれ5人の労働者は10のRPMを合計50のRPMを等しくします。 共有レートリミッターを使用します。
• 遅くすることなくIPを回転させます。 IP の回転は時間を買うが、各新しい IP がすぐに場所を槌で打たれば、高度の検出はあなたを押します。 適切な回転で回転を結合します。
• ピーク時のスクレイピング。 サイトは、高トラフィック期間中に制限率でより攻撃的です。 ターゲットのタイムゾーンのピーク時間に重いクロールをスケジュールします。

レート制限スクレイピングをサポートするために必要なプロキシの数を計算するには、を参照してください あなたがスクレイピングのために必要とする多くのプロキシは?. 率の制限を補うプロキシの回転戦略のために、読む 大規模なスクレイピングのためのプロキシ回転戦略. .

適切な速度制限スクレイピングを開始 ProxyHatのPython SDK または探検 プライシングプラン プロジェクトのために。

よくある質問

レート制限を超えるとどうなりますか?

応答はサイトによって異なります。 ほとんどのRetry-AfterヘッダーでHTTP 429を返します。 CAPTCHAs の提供するサービスもあります。 攻撃的なサイトはすぐに403応答でIPをブロックします。 最悪の場合、繰り返し違反は永久的なIP禁止につながります。

サイトの料金制限はどのようになっていますか?

応答コードを監視しながら、遅くなり次第に増加します。 Crawl-delay ディレクティブの robots.txt をチェックします。 X-RateLimit-Limit と X-RateLimit-Remaining フィールドのレスポンスヘッダを観察します。 一部の API は、ドキュメントの制限を公開しています。

プロキシのバイパスレート制限はありますか?

プロキシは複数の IP 間でリクエストを配信するので、各 IP は IP の制限に従っています。 しかし、洗練されたサイトでは、セッションや指紋、行動パターンの追跡も行っています。 プロキシは必要ありませんが、十分ではありません。適切な回転と現実的な要求パターンを組み合わせます。

スクラップのための最も安全な要求率は何ですか。

普遍的な答えはありません。 GoogleやAmazonなどの攻撃的なターゲットの場合、IPごとの1分あたりの1〜5リクエストは安全です。 軽く保護された場所のために、IPごとの20-60 RPMは働くかもしれません。 常に保守的を開始し、観察された成功率に基づいて増加.