大規模スクレイピングのためのプロキシローテーション戦略

プロキシの回転が大規模スクレイピングに不可欠である理由

数百〜数百万の要求をスケールアップすると、単一のプロキシIPは責任となります。 ウェブサイトは、IPごとのパターンを追跡し、通常の閲覧行動を超えるアドレスをスロットルまたは禁止します。 プロキシの回転 単一のアドレスが検出をトリガーするのに十分な活動を蓄積しないように、多くのIP間でリクエストを配布します。

ナイブ・ローテーションのアプローチとよく設計された戦略の違いは、95%の成功率と40%の1の違いを意味することができます。 このガイドでは、4つの主要な回転戦略を、それぞれ使用するとき、および作業コード例でそれらを実装する方法について説明します。

この記事は私たちの一部です ウェブスクレイピングプロキシの完全なガイド クラスター。 基礎的なプロキシの概念を必要としたらそこにを始めて下さい。

戦略1:リクエスト回転

最も単純なアプローチ: すべてのリクエストは新しい IP を取得します。. これは、各リクエストが独立しているステートレスデータ収集に理想的です — 価格検索, SERPクエリ, 製品ページフェッチ.

いつ使うか

• 各URLが独立した大型カタログをスクレイピング
• さまざまなキーワードでSERP監視
• クッキーやセッションの状態を必要としないタスク

Pythonの実装

import requests
PROXY = "http://USERNAME:PASSWORD@gate.proxyhat.com:8080"
def fetch_with_rotation(urls: list[str]) -> list[str]:
    """Each request automatically gets a fresh IP via the rotating gateway."""
    results = []
    session = requests.Session()
    session.proxies = {"http": PROXY, "https": PROXY}
    for url in urls:
        try:
            resp = session.get(url, timeout=30)
            resp.raise_for_status()
            results.append(resp.text)
        except requests.RequestException as e:
            print(f"Failed {url}: {e}")
            results.append(None)
    return results
# Each request through gate.proxyhat.com uses a different IP
pages = fetch_with_rotation([
    "https://example.com/product/1",
    "https://example.com/product/2",
    "https://example.com/product/3",
])

Node.js 実装

const HttpsProxyAgent = require('https-proxy-agent');
const fetch = require('node-fetch');
const agent = new HttpsProxyAgent('http://USERNAME:PASSWORD@gate.proxyhat.com:8080');
async function fetchWithRotation(urls) {
  const results = [];
  for (const url of urls) {
    try {
      const res = await fetch(url, { agent, timeout: 30000 });
      results.push(await res.text());
    } catch (err) {
      console.error(`Failed ${url}: ${err.message}`);
      results.push(null);
    }
  }
  return results;
}

導入事例

package main
import (
    "fmt"
    "io"
    "net/http"
    "net/url"
    "time"
)
func fetchWithRotation(urls []string) []string {
    proxyURL, _ := url.Parse("http://USERNAME:PASSWORD@gate.proxyhat.com:8080")
    client := &http.Client{
        Transport: &http.Transport{Proxy: http.ProxyURL(proxyURL)},
        Timeout:   30 * time.Second,
    }
    results := make([]string, len(urls))
    for i, u := range urls {
        resp, err := client.Get(u)
        if err != nil {
            fmt.Printf("Failed %s: %v\n", u, err)
            continue
        }
        body, _ := io.ReadAll(resp.Body)
        resp.Body.Close()
        results[i] = string(body)
    }
    return results
}

戦略2:タイムド・ローテーション(スティッキー・セッション)

いくつかのスクレイピングタスクは、関連する一連のリクエストに対して同じIPを必要とする - 多段のチェックアウトをナビゲートしたり、記録されたセッションを維持したり、ページネーションリストを閲覧したりする。 時間回転 (または粘着セッション) 定義された期間に割り当てられた同じ IP を保持します。, 通常 1-30 分.

いつ使うか

• 開始クローリング (ページ 1, 2, 3... 結果)
• クッキーやセッションの永続を必要とするタスク
• リアルなブラウジングパターンのシミュレーション

実装パターン

ProxyHat を使うと、スティッキーセッションは、資格情報内のセッションパラメータで制御されます。 各セッション ID は、設定された期間に同じ IP を保持します。

import requests
import uuid
def create_sticky_session(duration_label: str = "10m"):
    """Create a session that maintains the same IP."""
    session_id = uuid.uuid4().hex[:8]
    proxy = f"http://USERNAME-session-{session_id}:PASSWORD@gate.proxyhat.com:8080"
    session = requests.Session()
    session.proxies = {"http": proxy, "https": proxy}
    return session
# All requests through this session use the same IP
session = create_sticky_session()
page1 = session.get("https://example.com/listings?page=1")
page2 = session.get("https://example.com/listings?page=2")
page3 = session.get("https://example.com/listings?page=3")

Node.js スティッキーセッション

const HttpsProxyAgent = require('https-proxy-agent');
const fetch = require('node-fetch');
const crypto = require('crypto');
function createStickyAgent() {
  const sessionId = crypto.randomBytes(4).toString('hex');
  return new HttpsProxyAgent(
    `http://USERNAME-session-${sessionId}:PASSWORD@gate.proxyhat.com:8080`
  );
}
async function crawlPaginated(baseUrl, pages) {
  const agent = createStickyAgent(); // Same IP for all pages
  const results = [];
  for (let page = 1; page <= pages; page++) {
    const res = await fetch(`${baseUrl}?page=${page}`, { agent });
    results.push(await res.text());
  }
  return results;
}

戦略3:失敗ベースの回転

あらゆる要求またはタイマーの回転の代り、 故障ベースの回転 ブロックされるまで IP を使用して、スイッチを切り替えます。 動作する限り、各IPの値を最大化します。

いつ使うか

• 予測不可能な境界を持つターゲット
• IP ごとの最大の要求を望む予算意識スクレイピング
• 長時間走るクロールは、いくつかのIPが時間と他の分を持続する

自動故障による実装

import requests
import uuid
from time import sleep
class FailureBasedRotator:
    """Rotates proxy only when the current IP fails."""
    BLOCK_SIGNALS = [403, 429, 503]
    MAX_RETRIES = 3
    def __init__(self):
        self.session_id = None
        self.requests_on_current_ip = 0
        self._new_session()
    def _new_session(self):
        self.session_id = uuid.uuid4().hex[:8]
        self.requests_on_current_ip = 0
        proxy = f"http://USERNAME-session-{self.session_id}:PASSWORD@gate.proxyhat.com:8080"
        self.session = requests.Session()
        self.session.proxies = {"http": proxy, "https": proxy}
    def fetch(self, url: str) -> str | None:
        for attempt in range(self.MAX_RETRIES):
            try:
                resp = self.session.get(url, timeout=30)
                if resp.status_code in self.BLOCK_SIGNALS:
                    print(f"Blocked (HTTP {resp.status_code}) after "
                          f"{self.requests_on_current_ip} requests. Rotating...")
                    self._new_session()
                    sleep(1)
                    continue
                resp.raise_for_status()
                self.requests_on_current_ip += 1
                return resp.text
            except requests.RequestException:
                self._new_session()
                sleep(1)
        return None
# Usage
rotator = FailureBasedRotator()
for url in urls:
    html = rotator.fetch(url)

Failoverで実装する

package main
import (
    "crypto/rand"
    "encoding/hex"
    "fmt"
    "io"
    "net/http"
    "net/url"
    "time"
)
type FailureRotator struct {
    client    *http.Client
    sessionID string
    reqCount  int
}
func NewFailureRotator() *FailureRotator {
    r := &FailureRotator{}
    r.rotate()
    return r
}
func (r *FailureRotator) rotate() {
    b := make([]byte, 4)
    rand.Read(b)
    r.sessionID = hex.EncodeToString(b)
    r.reqCount = 0
    proxyStr := fmt.Sprintf("http://USERNAME-session-%s:PASSWORD@gate.proxyhat.com:8080", r.sessionID)
    proxyURL, _ := url.Parse(proxyStr)
    r.client = &http.Client{
        Transport: &http.Transport{Proxy: http.ProxyURL(proxyURL)},
        Timeout:   30 * time.Second,
    }
}
func (r *FailureRotator) Fetch(target string) (string, error) {
    for attempt := 0; attempt < 3; attempt++ {
        resp, err := r.client.Get(target)
        if err != nil {
            r.rotate()
            time.Sleep(time.Second)
            continue
        }
        defer resp.Body.Close()
        if resp.StatusCode == 403 || resp.StatusCode == 429 || resp.StatusCode == 503 {
            fmt.Printf("Blocked after %d requests. Rotating...\n", r.reqCount)
            r.rotate()
            time.Sleep(time.Second)
            continue
        }
        body, _ := io.ReadAll(resp.Body)
        r.reqCount++
        return string(body), nil
    }
    return "", fmt.Errorf("all retries exhausted for %s", target)
}

戦略4:ジオ分散ローテーション

ローカライズされたコンテンツをスクレイピングするとき — 検索結果、価格設定、可用性 — 特定の地理的な場所から IP が必要です。 ジオ分布回転 ターゲット国や都市からIPを割り当て、正確なローカルデータを取得します。

いつ使うか

• SERPスクレイピング ローカル検索ランキング
• 地域全体の価格監視
• コンテンツの可用性チェック(制限コンテンツ)
• 特定の市場における広告検証

カントリーターゲティングによる実装

import requests
from concurrent.futures import ThreadPoolExecutor
COUNTRIES = ["us", "gb", "de", "fr", "jp"]
def fetch_localized(url: str, country: str) -> dict:
    """Fetch URL through a proxy in the specified country."""
    proxy = f"http://USERNAME-country-{country}:PASSWORD@gate.proxyhat.com:8080"
    try:
        resp = requests.get(url, proxies={"http": proxy, "https": proxy}, timeout=30)
        return {"country": country, "status": resp.status_code, "body": resp.text}
    except requests.RequestException as e:
        return {"country": country, "status": 0, "error": str(e)}
def scrape_all_regions(url: str) -> list[dict]:
    """Fetch the same URL from multiple countries in parallel."""
    with ThreadPoolExecutor(max_workers=len(COUNTRIES)) as executor:
        futures = [executor.submit(fetch_localized, url, c) for c in COUNTRIES]
        return [f.result() for f in futures]
# Get localized pricing from 5 countries simultaneously
results = scrape_all_regions("https://example.com/product/pricing")
for r in results:
    print(f"{r['country'].upper()}: HTTP {r['status']}")

利用可能なターゲティングオプションを参照してください。 ProxyHat の場所 サイトマップ

戦略の融合:ハイブリッドアプローチ

練習では、大規模なスクレイピングプロジェクトは複数の戦略を組み合わせます。 ここでは、発見、深いクローリングのための粘りのあるセッション、および失敗ベースのフォールバックに使用するパターンです。

import requests
import uuid
from enum import Enum
class RotationMode(Enum):
    PER_REQUEST = "per_request"
    STICKY = "sticky"
    FAILURE_BASED = "failure_based"
class HybridRotator:
    def __init__(self, mode: RotationMode = RotationMode.PER_REQUEST):
        self.mode = mode
        self.session_id = None
        self.failure_count = 0
        self._init_session()
    def _init_session(self):
        if self.mode == RotationMode.PER_REQUEST:
            proxy = "http://USERNAME:PASSWORD@gate.proxyhat.com:8080"
        else:
            self.session_id = self.session_id or uuid.uuid4().hex[:8]
            proxy = f"http://USERNAME-session-{self.session_id}:PASSWORD@gate.proxyhat.com:8080"
        self.session = requests.Session()
        self.session.proxies = {"http": proxy, "https": proxy}
    def force_rotate(self):
        """Force a new IP regardless of mode."""
        self.session_id = uuid.uuid4().hex[:8]
        self.failure_count = 0
        self._init_session()
    def fetch(self, url: str) -> str | None:
        try:
            resp = self.session.get(url, timeout=30)
            if resp.status_code in [403, 429, 503]:
                self.failure_count += 1
                if self.failure_count >= 2:
                    self.force_rotate()
                return None
            self.failure_count = 0
            return resp.text
        except requests.RequestException:
            self.failure_count += 1
            if self.failure_count >= 2:
                self.force_rotate()
            return None
# Discovery phase: rotate every request
discovery = HybridRotator(RotationMode.PER_REQUEST)
sitemap_urls = [discovery.fetch(url) for url in seed_urls]
# Deep crawl phase: sticky sessions per site section
crawler = HybridRotator(RotationMode.STICKY)
for section_url in section_urls:
    pages = [crawler.fetch(f"{section_url}?page={i}") for i in range(1, 11)]
    crawler.force_rotate()  # New IP for next section

回転戦略の比較

スケールでの回転のためのベストプラクティス

• Robots.txt のリスペクト 回転は、良い市民であることからあなたを免除しません。 ルールをチェックし、craw-delayディレクティブに敬意を表します。
• 現実的な遅延を追加します。. 回転しても、毎秒数百のリクエストを破るのがロボティックです。 リクエスト間で0.5-2秒のランダム遅延を追加します。
• 成功率を監視して下さい。 ターゲットサイトの HTTP ステータスコードを追跡します。 90%未満の低下は、回転が調整されることを意味します。
• ヘッダーの回転と結合して下さい。 一人でIPを回転させるだけでは十分ではありません。 ユーザーエージェントの文字列やその他のヘッダーを回転させ、回避 指紋ベースの検出. .
• 故障時にバックオフを使用してください。 IP がブロックされると、再試行の前に待ちます。 暫定的なバックオフ(1秒、2秒、4秒、8秒)は、一時的に敵対的ターゲットに対する浪費要求を防ぎます。

回転戦略をサポートするために必要なIP数を理解するには、 あなたがスクレイピングのために必要とする多くのプロキシは?。 包括的なスクレイピングアーキテクチャの概要については、 ウェブスクレイピングプロキシの完全なガイド. .

これらの戦略を実装する準備は? チェックアウト Python SDK, ノードSDKまたは SDKについて 生産準備のプロキシ統合のため、または探検して下さい ProxyHatの価格設定プラン はじめに。

よくある質問

ウェブスクレイピングに最適なプロキシローテーション戦略は何ですか?

Per-request の回転はほとんどのスクレイピングタスクの最も安全なデフォルトです。 各リクエストは異なるIPを使用しており、パターンの検出をはるかに困難にします。 セッションの永続性(ペジネーション、ログインフロー)を必要とするタスクは、代わりにスティッキーセッションを使用します。

どのように高速でプロキシを回転する必要がありますか?

リクエスト回転ごとに、すべてのリクエストは自動的に新しいIPを取得します。 スティッキーセッションの場合、5-10分はデフォルトです。 最適な期間は、ターゲットに依存します。攻撃的なサイトでは、より短いセッション(1-2分)を必要とする場合があります。

異なる回転戦略を組み合わせることはできますか?

はい、複雑なプロジェクトのために必要です。 発見とURLの収集、深いクローリングのための粘りのあるセッション、およびIPがブロックされるときのフォールバックとして失敗ベースの回転をフォールバックとして使用してください。 このガイドのハイブリッドアプローチは、どのようにを示しています。

ProxyHat は自動回転を処理しますか?

はい。 リクエストを介してすべての ProxyHat ゲートウェイ (gate.proxyhat.com:8080) は、住宅プールから異なる IP を自動的に受け取ります。 スティッキーセッションでは、セッションパラメータを資格情報に追加します。 マニュアルIPリスト管理は必要ありません。