大规模爬取的代理轮换策略

代用旋转为什么对大范围擦拭至关重要

当你从数以百计的要求扩大到数以百万的要求时,一个单一的代理IP就成了一项负债. 网站跟踪每个IP的请求模式,并会节流或禁止超过正常浏览行为的地址. 代理旋转 将您的请求分布在许多IP上,这样不会使单个地址积累足够的活动来触发检测.

天真轮换方式与精心设计的策略的区别可能意味着95%的成功率与40%的成功率的区别. 本指南涵盖了四大轮换策略,每次何时使用,以及如何用工作代码实例执行.

这篇文章是我们 完整网页搜索代理指南 组合。 如果您需要基础代理概念, 请从那里开始 。

战略1:按请求轮换

最简单的做法: 每个请求都获得一个新的IP在每一个请求都是独立的——物价查询、SERP查询、产品页检索——无国籍人数据收集是理想的。

何时使用

• 绘制每个 URL 独立的大型目录
• SERP 跨越许多关键字的监测
• 不需要 cookie 或会话状态的任何任务

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",
])

节点.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页)
• 需要 cookie 或会话持久性的任务
• 模拟现实浏览模式

执行模式

使用 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")

节点.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)

以失败执行

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. 地理分布旋转 指定来自目标国家或城市的执行伙伴获取准确的地方数据。

何时使用

• ERP刮伤 用于本地搜索排名
• 跨区域价格监测
• 内容提供情况检查(地理限制内容)
• 在特定市场进行临时核查

以国家为目标的执行情况

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']}")

参见关于 代理汉特位置 页面。

综合战略:混合办法

在实践中,大规模刮刮项目结合了多种战略. 以下是一种模式,它使用每个请求旋转进行发现,粘性会话用于深层爬行,以及基于故障的倒置:

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

轮调战略比较

小规模旋转的最佳做法

• 敬重机器人. txt. (中文(简体) ). 轮换并不能免除你作为一个好公民的责任。 请检查date=中的日期值 (帮助)
• 增加现实的延迟。 即使是旋转, 破碎数百个请求每秒看起来像机器人。 在请求之间添加0.5-2次随机延迟。
• 监测成功率。 每个目标网站都有追踪HTTP状态代码. 低于90% 意味着您的旋转需要调谐 。
• 与头旋转结合. 仅仅轮换实施伙伴是不够的。 旋转用户代理字符串和其他信头以避免 指纹检测。 。 。
• 对失败使用后退. 当一个IP被屏蔽时,等待重试. 指示反向(1s,2s,4s,8s)防止对临时敌对目标浪费请求.

要了解你需要多少个IP来支持你的旋转策略,请看 需要多少代理吗?关于全面的刮刮结构概况,请访问 完整网页搜索代理指南。 。 。

准备实施这些战略吗? 检查一下 Python SDK 键盘, (中文). 节点 SDK,或 冲啊 SDK 或探索 代理哈特定价计划 开始吧

经常被问到的问题

网络刮刮的最佳代理旋转策略是什么?

按请求旋转是大多数刮刮任务最安全的默认值. 它确保每个请求使用不同的IP,使模式检测更加困难. 对于需要会话持久性(pagination,登录流量)的任务,使用粘度会话代替.

我应该多快地旋转代理?

对于每个请求旋转,每个请求都会自动获得一个新的IP. 对于粘性会话,5-10分钟是一个很好的默认. 最佳持续时间取决于目标——攻击性场地可能需要缩短会议时间(1-2分钟),而宽大场地则容忍30+分钟。

我能把不同的轮换策略结合起来吗?

是的,你应该为复杂的项目。 使用每个请求旋转进行发现和URL收集,粘度会话进行深爬行,失败旋转作为IP被屏蔽时的倒置. 本指南中的混合方法说明了方法.

代理服务器自动处理旋转吗 ?

对 通过ProxyHat网关(gate.proxyhat.com:8080)的每一个请求都会自动接收与住宅池不同的IP. 对于粘贴的会话,在您的证书中添加会话参数. 不需要人工IP列表管理.