IP地址轮换：如何及为何更换代理

请求 1 → 代理 A (185.45.12.34) → 网站看到 185.45.12.34
请求 2 → 代理 B (92.118.45.78) → 网站看到 92.118.45.78
请求 3 → 代理 C (178.62.91.22) → 网站看到 178.62.91.22
请求 4 → 代理 A (185.45.12.34) → 网站看到 185.45.12.34

1. 绕过速率限制 (Rate Limiting)

大多数网站会限制单个IP在单位时间内发出的请求数量。例如，API可能每小时只允许一个IP发送100个请求。使用包含10个IP的轮换池，您可以每小时发送1,000个请求，将负载分散开来。

2. 避免数据抓取时的IP封禁

在收集大量数据（如电子商务价格抓取、价格监控、联系信息收集）时，来自同一IP的频繁请求很快会导致封禁。轮换允许我们将请求分散到多个IP上，使每个IP的请求量看起来像普通用户一样少。

3. 绕过地理位置封锁

许多服务会根据地理位置显示不同的内容或价格。使用来自不同国家的代理进行轮换，可以收集所有地区的无差异数据，而无需身处当地。

4. 隐藏自动化行为

安全系统（如Cloudflare, Akamai, PerimeterX）会分析行为模式。如果短时间内有大量请求来自同一IP，这显然是机器人的标志。轮换有助于营造出多个独立用户访问的假象。

5. 竞争情报

跟踪竞争对手的价格、监控广告活动、分析SEO排名都需要频繁检查。IP轮换可以让我们在不引起竞争对手注意的情况下收集这些数据。

6. 测试和监控

从不同地区检查网站可用性、测试A/B实验、监控不同国家的SEO排名，都需要使用来自不同地点的IP地址。

1. 速率限制 (请求频率分析)

网站会跟踪特定时间段内来自同一IP的请求数量。典型阈值如下：

• 保守型网站： 每分钟10-30个请求
• 中等网站： 每分钟50-100个请求
• API服务： 每小时100-1000个请求（通常在文档中说明）

2. IP信誉分析

存在广泛的数据库对IP地址进行分类：

• 住宅IP (Residential IP) — 来自家庭互联网服务提供商（信誉高）
• 数据中心IP (Datacenter IP) — 来自托管公司（可疑）
• 移动IP (Mobile IP) — 来自移动运营商（信誉高）
• 已知代理IP (Known proxy IP) — 经常被封锁

3. 浏览器指纹 (Browser Fingerprinting)

即使更换了IP，安全系统仍可以通过“浏览器指纹”将请求关联起来：屏幕分辨率、已安装的字体、插件、WebGL指纹、Canvas指纹、音频上下文指纹。

4. 行为分析 (Behavioral Analysis)

现代反机器人系统会分析以下行为：

• 页面滚动速度
• 鼠标移动轨迹
• 点击模式
• 操作间隔时间
• 页面访问顺序

5. TLS指纹

在HTTPS连接中，服务器可以检测到所使用的TLS版本、支持的加密套件和扩展——这些数据会形成一个指纹，即使更换IP也可能被用于跟踪。

典型轮换间隔：

• 每 5 分钟 — 适用于高频率请求的密集抓取
• 每 10-15 分钟 — 大多数任务的标准模式
• 每 30-60 分钟 — 低频率请求的任务
• 每 2-24 小时 — 用于粘性会话 (sticky sessions)

✅ 优点：

• 可预测性 — 您确切知道IP何时会切换
• 易于实现 — 使用定时器即可轻松设置
• 支持粘性会话 — 可以在特定时间内保持会话不变
• 均匀分布 — 负载在时间上均匀分布
• 易于扩展 — 可以为不同的会话设置不同的定时器

❌ 缺点：

• 在负载变化时效率低下 — 如果请求很少，IP仍会不必要地切换
• 可能超出限制 — 如果在短时间内发送了过多请求
• 可预测的模式 — 高级安全系统可能会检测到这种规律性
• 会话丢失 — IP切换可能导致授权丢失或上下文丢失

🎯 最适合：

• 负载可预测的任务
• 需要保持会话（授权、账户操作）的长期任务
• 固定时间间隔的监控任务
• 需要IP在特定时间内保持稳定的情况

10:00 - 代理 A → 检查价格 (50个商品)
10:30 - 代理 B → 检查价格 (50个商品)
11:00 - 代理 C → 检查价格 (50个商品)
11:30 - 代理 D → 检查价格 (50个商品)
12:00 - 代理 A → 检查价格 (50个商品)

实现方式：

1. 每个请求轮换 — 每个请求使用新的IP（最激进的策略）
2. 突发轮换 — 每N个请求后更改IP（例如，每10个请求）
3. 自适应轮换 — 收到特定HTTP代码（429, 403, 503）时更改IP
4. 基于会话 — 在启动新的逻辑会话时更改IP

✅ 优点：

• 最大程度规避速率限制 — 每个IP的请求最少
• 自适应性 — 仅在需要时才轮换
• 高效利用代理池 — 仅在必要时才更换IP
• 快速响应封锁 — 收到错误时可立即更换IP
• 非常适合抓取 — 每个页面都使用新的IP请求

❌ 缺点：

• 无法保持会话 — 持续更换IP会破坏授权
• 调试困难 — 很难重现特定IP出现的问题
• 代理池消耗快 — 高强度工作会迅速耗尽所有IP
• 成本更高 — 需要更大的代理池才能有效运行
• 切换开销 — 每次IP切换都会有微小的延迟

🎯 最适合：

• 大量数据的高强度抓取
• 绕过严格的速率限制
• 不需要保持状态的单次请求
• 无需授权的公共页面抓取
• 每个请求都相互独立的情况

网站类型	推荐请求数	间隔时间
高防护网站（银行、社交媒体）	1-3 个请求	请求间 5-10 秒
电子商务（市场）	5-10 个请求	请求间 2-5 秒
新闻门户	10-20 个请求	请求间 1-3 秒
公共 API	取决于限制	根据文档
静态网站	20-50 个请求	请求间 0.5-2 秒

随机轮换的变体：

• 随机时间间隔 — 在随机时间段内切换IP（例如，3到15分钟之间）
• 随机请求数 — 在随机请求数后切换（例如，5到20个请求之间）
• 随机IP选择 — 从池中随机选择下一个IP，而不是按顺序选择
• 加权随机 — 优先使用信誉高、速度快的IP
• 抖动轮换 (Jittered rotation) — 在固定间隔中增加随机延迟

✅ 优点：

• 不可预测性 — 安全系统更难检测到模式
• 模仿真实用户 — 人类的行为很少是完全规律的
• 灵活性 — 可以与其他策略结合使用
• 自然流量模式 — 更接近有机流量
• 更难被检测 — 即使在大量数据分析中也更难识别

❌ 缺点：

• 更难预测 — 难以估计任务的完成速度
• 可能效率低下 — 随机性可能导致IP更换过于频繁
• 调试复杂 — 由于随机性，重现问题更困难
• 需要更大的池 — 以确保负载均匀分布
• 实现更复杂 — 需要良好的随机数生成算法

🎯 最适合：

• 绕过高级安全系统（如Cloudflare, Akamai）
• 对隐蔽性要求极高的长期项目
• 竞争情报收集
• 抓取具有行为分析的网站
• 需要最大程度模仿人类行为的任务

标准	基于时间	基于请求	随机
实现难度	⭐ 简单	⭐⭐ 中等	⭐⭐⭐ 复杂
可预测性	✅ 高	⚠️ 中等	❌ 低
速率限制规避能力	⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐⭐
隐蔽性	⭐⭐	⭐⭐⭐	⭐⭐⭐⭐⭐
会话支持	✅ 是	❌ 否	⚠️ 部分
代理池利用率	⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐⭐
抓取速度	⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐⭐
所需池大小	小-中等	大	中-大
调试难度	✅ 简单	⚠️ 中等	❌ 困难
成本	💰 低	💰💰💰 高	💰💰 中等

典型粘性会话设置：

• 1-5 分钟 — 用于快速操作的短期会话
• 10-30 分钟 — 大多数任务的标准模式
• 1-2 小时 — 用于账户操作和授权
• 12-24 小时 — 用于长期操作的最大持续时间
• 无限 (Infinite) — IP保持不变直到会话结束

✅ 粘性会话的优点：

• 保持授权 — 可以以登录状态进行操作
• 支持 Cookies — 网站能“记住”您
• 自然行为 — 普通用户在一次会话中通常使用一个IP
• 更少的验证码 — 稳定的IP更不容易引起怀疑
• 操作一致性 — 可以执行多步骤操作
• 易于调试 — 容易重现特定IP出现的问题

❌ 粘性会话的缺点：

• 易受速率限制影响 — 所有请求都来自同一IP
• 整个会话被封禁的风险 — 如果IP被封，所有进度都会丢失
• 可扩展性较差 — 受限于单个IP的速度
• 会话有限 — 会话最终会结束，IP终将改变

✅ 轮换代理的优点：

• 最大程度规避速率限制 — 每个IP的请求最少
• 高抓取速度 — 可以发送数千个并行请求
• 封禁风险低 — 即使一个IP被封，也不影响整体工作
• 高可扩展性 — 轻松增加工作量
• 隐蔽性高 — 看起来像许多独立的访客

❌ 轮换代理的缺点：

• 无法授权 — 持续更换IP会破坏会话
• Cookies无效 — 每个请求都像来自新用户
• 多步骤操作失败 — 购物车、结账等流程无法完成
• 可能导致更多验证码 — 频繁更换IP可能引起怀疑
• 成本更高 — 需要大量的代理IP

标准	粘性会话	轮换代理
授权	✅ 是	❌ 否
Cookies	✅ 有效	❌ 无效
速率限制规避	⚠️ 有限	✅ 优秀
抓取速度	⭐⭐⭐	⭐⭐⭐⭐⭐
封禁风险	⚠️ 中等	✅ 低
成本	💰 较低	💰💰 较高
复杂度	⭐ 简单	⭐⭐ 中等

1. 社交媒体管理

在处理多个Instagram、Facebook、Twitter账户时，必须在会话期间保持同一个IP。频繁更换IP在授权的账户上很容易导致封禁。

建议： 粘性会话 1-2 小时，每个账户对应一个唯一的IP。

2. 电子商务和购物车

将商品添加到购物车、完成订单、结账流程都需要保持会话。IP更换会导致购物车丢失，需要重新开始。

建议： 粘性会话 30-60 分钟，以完成整个购买周期。

3. 表单填写和注册

多步骤表单、网站注册、邮件验证等过程需要IP保持一致。步骤间的IP切换可能引起怀疑或验证失败。

建议： 粘性会话 10-30 分钟，以完成整个流程。

4. Web应用测试

端到端测试、使用Selenium/Puppeteer的自动化、用户场景验证——所有这些都需要保持IP以模仿真实用户体验。

建议： 粘性会话持续整个测试过程（5-60分钟）。

5. 需要身份验证的API

许多API会根据IP地址验证访问令牌的有效性。更换IP将导致令牌失效，需要重新认证。

建议： 粘性会话持续令牌的生命周期（通常1-24小时）。

import requests
from itertools import cycle

# 代理列表
proxies_list = [
    'http://user:pass@185.45.12.34:8000',
    'http://user:pass@92.118.45.78:8000',
    'http://user:pass@178.62.91.22:8000',
    'http://user:pass@45.89.234.56:8000'
]

# 创建无限迭代器
proxy_pool = cycle(proxies_list)

# 发送请求的函数
def make_request(url):
    proxy = next(proxy_pool)
    proxies = {
        'http': proxy,
        'https': proxy
    }

    try:
        response = requests.get(url, proxies=proxies, timeout=10)
        print(f"成功使用 {proxy}: {response.status_code}")
        return response
    except Exception as e:
        print(f"使用 {proxy} 时出错: {e}")
        return None

# 使用
urls = ['https://example.com/page1', 'https://example.com/page2']
for url in urls:
    make_request(url)
    # 每个请求都使用列表中的下一个代理

import requests
import random
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

class ProxyRotator:
    def __init__(self, proxies_list):
        self.proxies = proxies_list
        self.failed_proxies = set()

    def get_random_proxy(self):
        """获取一个随机的可用代理"""
        available = [p for p in self.proxies if p not in self.failed_proxies]
        if not available:
            # 如果所有代理都失败了，则重置列表
            self.failed_proxies.clear()
            available = self.proxies
        return random.choice(available)

    def make_request(self, url, max_retries=3):
        """发送请求，出错时自动轮换代理"""
        session = requests.Session()

        # 设置重试策略
        retry = Retry(
            total=max_retries,
            backoff_factor=0.5,
            status_forcelist=[500, 502, 503, 504]
        )
        adapter = HTTPAdapter(max_retries=retry)
        session.mount('http://', adapter)
        session.mount('https://', adapter)

        for attempt in range(max_retries):
            proxy = self.get_random_proxy()
            proxies = {'http': proxy, 'https': proxy}

            try:
                response = session.get(url, proxies=proxies, timeout=15)

                # 检查速率限制
                if response.status_code == 429:
                    print(f"在 {proxy} 上被限速，正在轮换...")
                    self.failed_proxies.add(proxy)
                    continue

                print(f"✓ 成功使用 {proxy}")
                return response

            except Exception as e:
                print(f"✗ 使用 {proxy} 失败: {e}")
                self.failed_proxies.add(proxy)

        raise Exception(f"所有重试都失败了 {url}")

# 使用
proxies = [
    'http://user:pass@proxy1.com:8000',
    'http://user:pass@proxy2.com:8000',
    'http://user:pass@proxy3.com:8000'
]

rotator = ProxyRotator(proxies)
response = rotator.make_request('https://example.com')

import requests
import time
from datetime import datetime, timedelta

class TimeBasedRotator:
    def __init__(self, proxies_list, rotation_interval=600):
        """
        rotation_interval: 间隔时间（秒，600 = 10分钟）
        """
        self.proxies = proxies_list
        self.rotation_interval = rotation_interval
        self.current_proxy = None
        self.last_rotation = None
        self.current_index = 0

    def get_proxy(self):
        """获取当前代理或在时间到期时轮换"""
        now = datetime.now()

        # 首次运行或时间到期
        if (self.last_rotation is None or
            (now - self.last_rotation).seconds >= self.rotation_interval):

            self.current_proxy = self.proxies[self.current_index]
            self.current_index = (self.current_index + 1) % len(self.proxies)
            self.last_rotation = now
            print(f"🔄 轮换至: {self.current_proxy}")

        return self.current_proxy

    def make_request(self, url):
        proxy = self.get_proxy()
        proxies = {'http': proxy, 'https': proxy}

        response = requests.get(url, proxies=proxies, timeout=10)
        return response

# 使用：每10分钟更换一次IP
rotator = TimeBasedRotator(proxies_list, rotation_interval=600)

for i in range(100):
    response = rotator.make_request('https://example.com')
    print(f"请求 {i}: {response.status_code}")
    time.sleep(2)  # 请求间暂停2秒

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

class ProxyRotator {
  constructor(proxies) {
    this.proxies = proxies;
    this.currentIndex = 0;
  }

  getNextProxy() {
    const proxy = this.proxies[this.currentIndex];
    this.currentIndex = (this.currentIndex + 1) % this.proxies.length;
    return proxy;
  }

  async makeRequest(url, options = {}) {
    const proxy = this.getNextProxy();
    const agent = new HttpsProxyAgent(proxy);

    try {
      const response = await axios.get(url, {
        ...options,
        httpAgent: agent,
        httpsAgent: agent,
        timeout: 10000
      });

      console.log(`✓ 成功使用 ${proxy}: ${response.status}`);
      return response.data;

    } catch (error) {
      console.error(`✗ 使用 ${proxy} 失败: ${error.message}`);
      throw error;
    }
  }
}

// 使用
const proxies = [
  'http://user:pass@proxy1.com:8000',
  'http://user:pass@proxy2.com:8000',
  'http://user:pass@proxy3.com:8000'
];

const rotator = new ProxyRotator(proxies);

async function scrape() {
  const urls = [
    'https://example.com/page1',
    'https://example.com/page2',
    'https://example.com/page3'
  ];

  for (const url of urls) {
    try {
      await rotator.makeRequest(url);
    } catch (error) {
      console.error(`抓取 ${url} 失败`);
    }
  }
}

scrape();

const puppeteer = require('puppeteer');

class PuppeteerProxyRotator {
  constructor(proxies) {
    this.proxies = proxies;
    this.currentIndex = 0;
  }

  getNextProxy() {
    const proxy = this.proxies[this.currentIndex];
    this.currentIndex = (this.currentIndex + 1) % this.proxies.length;
    return proxy;
  }

  async scrapeWithRotation(url) {
    const proxy = this.getNextProxy();

    // 解析 proxy URL
    const proxyUrl = new URL(proxy);

    const browser = await puppeteer.launch({
      headless: true,
      args: [
        `--proxy-server=${proxyUrl.protocol}//${proxyUrl.host}`,
        '--no-sandbox',
        '--disable-setuid-sandbox'
      ]
    });

    try {
      const page = await browser.newPage();

      // 如果代理需要认证
      if (proxyUrl.username && proxyUrl.password) {
        await page.authenticate({
          username: proxyUrl.username,
          password: proxyUrl.password
        });
      }

      await page.goto(url, { waitUntil: 'networkidle2', timeout: 30000 });

      const content = await page.content();
      console.log(`✓ 使用 ${proxy} 抓取 ${url}`);

      await browser.close();
      return content;

    } catch (error) {
      console.error(`✗ ${proxy} 错误: ${error.message}`);
      await browser.close();
      throw error;
    }
  }
}

// 使用
const proxies = [
  'http://user:pass@185.45.12.34:8000',
  'http://user:pass@92.118.45.78:8000'
];

const rotator = new PuppeteerProxyRotator(proxies);

async function scrapeMultiplePages() {
  const urls = ['https://example.com/1', 'https://example.com/2'];

  for (const url of urls) {
    await rotator.scrapeWithRotation(url);
    // 每个页面都使用新的代理打开
  }
}

scrapeMultiplePages();

工作原理：

1. 您连接到一个统一的端点（例如，rotate.proxycove.com:8000）
2. 在每次请求时，网关会自动从池中选择一个随机IP
3. 您无需管理代理列表或编写轮换逻辑
4. 可以通过在用户名中添加会话ID来配置粘性会话

# Python 示例，使用轮换网关
import requests

# 用于轮换：每个请求 = 新IP
proxies = {
    'http': 'http://username:password@rotate.proxycove.com:8000',
    'https': 'http://username:password@rotate.proxycove.com:8000'
}

# 用于粘性会话：添加 session_id
sticky_proxies = {
    'http': 'http://username-session-abc123:password@rotate.proxycove.com:8000',
    'https': 'http://username-session-abc123:password@rotate.proxycove.com:8000'
}

# 轮换：每个请求都是新IP
for i in range(10):
    r = requests.get('https://api.ipify.org', proxies=proxies)
    print(f"请求 {i}: IP = {r.text}")  # 每次都不同

# 粘性：所有请求都是同一个IP
for i in range(10):
    r = requests.get('https://api.ipify.org', proxies=sticky_proxies)
    print(f"请求 {i}: IP = {r.text}")  # 始终相同

1. ProxyBroker

用于查找、验证和使用代理的库，支持自动轮换。

2. rotating-proxies (Scrapy 中间件)

Scrapy 的中间件，支持自动轮换和黑名单管理。

3. requests-ip-rotator

requests 库的扩展，支持通过 AWS API Gateway 进行IP轮换。

1. proxy-chain

用于创建具有轮换和隧道功能的HTTP代理服务器的库。

2. puppeteer-extra-plugin-proxy-rotation

Puppeteer 的插件，为每个页面自动轮换代理。

1. 加权轮换 (Weighted Rotation)

信誉和速度更好的代理使用频率更高。例如，住宅IP权重为0.6，数据中心IP权重为0.4。

2. 地理位置定向轮换

根据目标URL自动选择特定国家/城市的代理。例如，访问.de域名时使用德国代理。

3. 健康检查与自动移除

定期检查代理健康状况，并自动将故障IP移出池外。在“冷却”期后恢复使用。

4. 请求率自适应轮换

轮换频率根据收到的HTTP状态码自动调整。收到429错误时，请求速度加快。

典型粘性会话设置：

• 1-5 分钟 — 用于快速操作的短期会话
• 10-30 分钟 — 大多数任务的标准模式
• 1-2 小时 — 用于账户操作和授权
• 12-24 小时 — 用于长期操作的最大持续时间
• 无限 (Infinite) — IP保持不变直到会话结束

✅ 粘性会话的优点：

• 保持授权 — 可以以登录状态进行操作
• 支持 Cookies — 网站能“记住”您
• 自然行为 — 普通用户在一次会话中通常使用一个IP
• 更少的验证码 — 稳定的IP更不容易引起怀疑
• 操作一致性 — 可以执行多步骤操作
• 易于调试 — 容易重现特定IP出现的问题

❌ 粘性会话的缺点：

• 易受速率限制影响 — 所有请求都来自同一IP
• 整个会话被封禁的风险 — 如果IP被封，所有进度都会丢失
• 可扩展性较差 — 受限于单个IP的速度
• 会话有限 — 会话最终会结束，IP终将改变

✅ 轮换代理的优点：

• 最大程度规避速率限制 — 每个IP的请求最少
• 高抓取速度 — 可以发送数千个并行请求
• 封禁风险低 — 即使一个IP被封，也不影响整体工作
• 高可扩展性 — 轻松增加工作量
• 隐蔽性高 — 看起来像许多独立的访客

❌ 轮换代理的缺点：

• 无法授权 — 持续更换IP会破坏会话
• Cookies无效 — 每个请求都像来自新用户
• 多步骤操作失败 — 购物车、结账等流程无法完成
• 可能导致更多验证码 — 频繁更换IP可能引起怀疑
• 成本更高 — 需要大量的代理IP

标准	粘性会话	轮换代理
授权	✅ 是	❌ 否
Cookies	✅ 有效	❌ 无效
速率限制规避	⚠️ 有限	✅ 优秀
抓取速度	⭐⭐⭐	⭐⭐⭐⭐⭐
封禁风险	⚠️ 中等	✅ 低
成本	💰 较低	💰💰 较高
复杂度	⭐ 简单	⭐⭐ 中等

1. 社交媒体管理

在处理多个Instagram、Facebook、Twitter账户时，必须在会话期间保持同一个IP。频繁更换IP在授权的账户上很容易导致封禁。

建议： 粘性会话 1-2 小时，每个账户对应一个唯一的IP。

2. 电子商务和购物车

将商品添加到购物车、完成订单、结账流程都需要保持会话。IP更换会导致购物车丢失，需要重新开始。

建议： 粘性会话 30-60 分钟，以完成整个购买周期。

3. 表单填写和注册

多步骤表单、网站注册、邮件验证等过程需要IP保持一致。步骤间的IP切换可能引起怀疑或验证失败。

建议： 粘性会话 10-30 分钟，以完成整个流程。

4. Web应用测试

端到端测试、使用Selenium/Puppeteer的自动化、用户场景验证——所有这些都需要保持IP以模仿真实用户体验。

建议： 粘性会话持续整个测试过程（5-60分钟）。

5. 需要身份验证的API

许多API会根据IP地址验证访问令牌的有效性。更换IP将导致令牌失效，需要重新认证。

建议： 粘性会话持续令牌的生命周期（通常1-24小时）。

import requests
from itertools import cycle

# 代理列表
proxies_list = [
    'http://user:pass@185.45.12.34:8000',
    'http://user:pass@92.118.45.78:8000',
    'http://user:pass@178.62.91.22:8000',
    'http://user:pass@45.89.234.56:8000'
]

# 创建无限迭代器
proxy_pool = cycle(proxies_list)

# 发送请求的函数
def make_request(url):
    proxy = next(proxy_pool)
    proxies = {
        'http': proxy,
        'https': proxy
    }

    try:
        response = requests.get(url, proxies=proxies, timeout=10)
        print(f"成功使用 {proxy}: {response.status_code}")
        return response
    except Exception as e:
        print(f"使用 {proxy} 时出错: {e}")
        return None

# 使用
urls = ['https://example.com/page1', 'https://example.com/page2']
for url in urls:
    make_request(url)
    # 每个请求都使用列表中的下一个代理

import requests
import random
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

class ProxyRotator:
    def __init__(self, proxies_list):
        self.proxies = proxies_list
        self.failed_proxies = set()

    def get_random_proxy(self):
        """获取一个随机的可用代理"""
        available = [p for p in self.proxies if p not in self.failed_proxies]
        if not available:
            # 如果所有代理都失败了，则重置列表
            self.failed_proxies.clear()
            available = self.proxies
        return random.choice(available)

    def make_request(self, url, max_retries=3):
        """发送请求，出错时自动轮换代理"""
        session = requests.Session()

        # 设置重试策略
        retry = Retry(
            total=max_retries,
            backoff_factor=0.5,
            status_forcelist=[500, 502, 503, 504]
        )
        adapter = HTTPAdapter(max_retries=retry)
        session.mount('http://', adapter)
        session.mount('https://', adapter)

        for attempt in range(max_retries):
            proxy = self.get_random_proxy()
            proxies = {'http': proxy, 'https': proxy}

            try:
                response = session.get(url, proxies=proxies, timeout=15)

                # 检查速率限制
                if response.status_code == 429:
                    print(f"在 {proxy} 上被限速，正在轮换...")
                    self.failed_proxies.add(proxy)
                    continue

                print(f"✓ 成功使用 {proxy}")
                return response

            except Exception as e:
                print(f"✗ 使用 {proxy} 失败: {e}")
                self.failed_proxies.add(proxy)

        raise Exception(f"所有重试都失败了 {url}")

# 使用
proxies = [
    'http://user:pass@proxy1.com:8000',
    'http://user:pass@proxy2.com:8000',
    'http://user:pass@proxy3.com:8000'
]

rotator = ProxyRotator(proxies)
response = rotator.make_request('https://example.com')

import requests
import time
from datetime import datetime, timedelta

class TimeBasedRotator:
    def __init__(self, proxies_list, rotation_interval=600):
        """
        rotation_interval: 间隔时间（秒，600 = 10分钟）
        """
        self.proxies = proxies_list
        self.rotation_interval = rotation_interval
        self.current_proxy = None
        self.last_rotation = None
        self.current_index = 0

    def get_proxy(self):
        """获取当前代理或在时间到期时轮换"""
        now = datetime.now()

        # 首次运行或时间到期
        if (self.last_rotation is None or
            (now - self.last_rotation).seconds >= self.rotation_interval):

            self.current_proxy = self.proxies[self.current_index]
            self.current_index = (self.current_index + 1) % len(self.proxies)
            self.last_rotation = now
            print(f"🔄 轮换至: {self.current_proxy}")

        return self.current_proxy

    def make_request(self, url):
        proxy = self.get_proxy()
        proxies = {'http': proxy, 'https': proxy}

        response = requests.get(url, proxies=proxies, timeout=10)
        return response

# 使用：每10分钟更换一次IP
rotator = TimeBasedRotator(proxies_list, rotation_interval=600)

for i in range(100):
    response = rotator.make_request('https://example.com')
    print(f"请求 {i}: {response.status_code}")
    time.sleep(2)  # 请求间暂停2秒

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

class ProxyRotator {
  constructor(proxies) {
    this.proxies = proxies;
    this.currentIndex = 0;
  }

  getNextProxy() {
    const proxy = this.proxies[this.currentIndex];
    this.currentIndex = (this.currentIndex + 1) % this.proxies.length;
    return proxy;
  }

  async makeRequest(url, options = {}) {
    const proxy = this.getNextProxy();
    const agent = new HttpsProxyAgent(proxy);

    try {
      const response = await axios.get(url, {
        ...options,
        httpAgent: agent,
        httpsAgent: agent,
        timeout: 10000
      });

      console.log(`✓ 成功使用 ${proxy}: ${response.status}`);
      return response.data;

    } catch (error) {
      console.error(`✗ 使用 ${proxy} 失败: ${error.message}`);
      throw error;
    }
  }
}

// 使用
const proxies = [
  'http://user:pass@proxy1.com:8000',
  'http://user:pass@proxy2.com:8000',
  'http://user:pass@proxy3.com:8000'
];

const rotator = new ProxyRotator(proxies);

async function scrape() {
  const urls = [
    'https://example.com/page1',
    'https://example.com/page2',
    'https://example.com/page3'
  ];

  for (const url of urls) {
    try {
      await rotator.makeRequest(url);
    } catch (error) {
      console.error(`抓取 ${url} 失败`);
    }
  }
}

scrape();

const puppeteer = require('puppeteer');

class PuppeteerProxyRotator {
  constructor(proxies) {
    this.proxies = proxies;
    this.currentIndex = 0;
  }

  getNextProxy() {
    const proxy = this.proxies[this.currentIndex];
    this.currentIndex = (this.currentIndex + 1) % this.proxies.length;
    return proxy;
  }

  async scrapeWithRotation(url) {
    const proxy = this.getNextProxy();

    // 解析 proxy URL
    const proxyUrl = new URL(proxy);

    const browser = await puppeteer.launch({
      headless: true,
      args: [
        `--proxy-server=${proxyUrl.protocol}//${proxyUrl.host}`,
        '--no-sandbox',
        '--disable-setuid-sandbox'
      ]
    });

    try {
      const page = await browser.newPage();

      // 如果代理需要认证
      if (proxyUrl.username && proxyUrl.password) {
        await page.authenticate({
          username: proxyUrl.username,
          password: proxyUrl.password
        });
      }

      await page.goto(url, { waitUntil: 'networkidle2', timeout: 30000 });

      const content = await page.content();
      console.log(`✓ 使用 ${proxy} 抓取 ${url}`);

      await browser.close();
      return content;

    } catch (error) {
      console.error(`✗ ${proxy} 错误: ${error.message}`);
      await browser.close();
      throw error;
    }
  }
}

// 使用
const proxies = [
  'http://user:pass@185.45.12.34:8000',
  'http://user:pass@92.118.45.78:8000'
];

const rotator = new PuppeteerProxyRotator(proxies);

async function scrapeMultiplePages() {
  const urls = ['https://example.com/1', 'https://example.com/2'];

  for (const url of urls) {
    await rotator.scrapeWithRotation(url);
    // 每个页面都使用新的代理打开
  }
}

scrapeMultiplePages();

工作原理：

1. 您连接到一个统一的端点（例如，rotate.proxycove.com:8000）
2. 在每次请求时，网关会自动选择一个随机IP池中的IP
3. 您无需管理代理列表或编写轮换逻辑
4. 可以通过在用户名中添加会话ID来配置粘性会话

# Python 示例，使用轮换网关
import requests

# 用于轮换：每个请求 = 新IP
proxies = {
    'http': 'http://username:password@rotate.proxycove.com:8000',
    'https': 'http://username:password@rotate.proxycove.com:8000'
}

# 用于粘性会话：添加 session_id
sticky_proxies = {
    'http': 'http://username-session-abc123:password@rotate.proxycove.com:8000',
    'https': 'http://username-session-abc123:password@rotate.proxycove.com:8000'
}

# 轮换：每个请求都是新IP
for i in range(10):
    r = requests.get('https://api.ipify.org', proxies=proxies)
    print(f"请求 {i}: IP = {r.text}")  # 每次都不同

# 粘性：所有请求都是同一个IP
for i in range(10):
    r = requests.get('https://api.ipify.org', proxies=sticky_proxies)
    print(f"请求 {i}: IP = {r.text}")  # 始终相同

1. ProxyBroker

用于查找、验证和使用代理的库，支持自动轮换。

2. rotating-proxies (Scrapy 中间件)

Scrapy 的中间件，支持自动轮换和黑名单管理。

3. requests-ip-rotator

requests 库的扩展，支持通过 AWS API Gateway 进行IP轮换。

1. proxy-chain

用于创建具有轮换和隧道功能的HTTP代理服务器的库。

2. puppeteer-extra-plugin-proxy-rotation

Puppeteer 的插件，为每个页面自动轮换代理。

1. 加权轮换 (Weighted Rotation)

信誉和速度更好的代理使用频率更高。例如，住宅IP权重为0.6，数据中心IP权重为0.4。

2. 地理位置定向轮换

根据目标URL自动选择特定国家/城市的代理。例如，访问.de域名时使用德国代理。

3. 健康检查与自动移除

定期检查代理健康状况，并自动将故障IP移出池外。在“冷却”期后恢复使用。

4. 请求率自适应轮换

轮换频率根据收到的HTTP状态码自动调整。收到429错误时，请求速度加快。

1. 目标网站类型

• 高防护网站（银行、社交媒体）： 每3-5个请求轮换一次，或每10-15分钟轮换一次
• 电子商务（亚马逊、沃尔玛）： 每5-10个请求轮换一次，或每5-10分钟轮换一次
• 新闻门户： 每10-20个请求轮换一次，或每15-30分钟轮换一次
• 公共 API： 根据API文档（通常是每小时100-1000个请求/IP）
• 静态网站： 每20-50个请求轮换一次，或每30-60分钟轮换一次

2. 收集数据量

• 小批量（少于1,000页）： 基于时间，每15-30分钟轮换一次
• 中等批量（1,000-10,000页）： 基于请求，每10-15个请求轮换一次
• 大批量（10,000+页）： 每个请求轮换（Per-request rotation），需要大IP池

3. 代理池大小

• 小池（10-50个IP）： 基于时间，每30-60分钟轮换一次，以确保每个IP有足够“冷却”时间
• 中等池（50-200个IP）： 基于请求，每10-20个请求轮换一次
• 大池（200+个IP）： 每个请求轮换，最大速度

4. 会话要求

• 无需授权： 激进轮换，每1-5个请求
• 需要授权： 粘性会话（1-24小时）
• 混合模式： 授权时粘性，数据收集时轮换

使用场景	轮换频率	所需池大小	请求间隔延迟
Google 搜索抓取	每 3-5 个请求	200-500 IP	5-10 秒
Amazon 价格监控	每 5-10 个请求	100-300 IP	3-7 秒
Instagram 自动化	粘性会话 1-2 小时	1 IP/账户	30-60 秒
新闻聚合器	每 15-30 分钟	50-100 IP	1-3 秒
房产信息抓取	每 10-20 个请求	50-200 IP	2-5 秒
API 监控	根据 API 限制	根据限制	根据文档
SEO 排名跟踪	每 20-30 个请求	100-300 IP	3-8 秒
Avito/Craigslist 抓取	每 7-15 个请求	100-200 IP	3-6 秒

1. 轮询 (Round-Robin)

代理按顺序从列表中选择。用完列表后从头开始。最简单的方法，确保均匀分布。

2. 随机 (Random)

每次都从池中随机选择一个代理。模式更难被检测到。

3. 最少连接 (Least Connections)

选择当前活动连接数最少的代理。并行请求的理想选择。

4. 加权轮询 (Weighted Round-Robin)

根据代理性能分配权重。信誉好的IP使用更频繁。数据中心IP权重为1，住宅IP权重为3。

5. IP 哈希 (IP Hash)

根据URL或域名的哈希值选择代理。确保对特定域名的请求始终通过同一IP。

推荐冷却时间：

• 小池（10-50个IP）： 30-60分钟内不得重复使用同一IP
• 中等池（50-200个IP）： 15-30分钟内不得重复使用同一IP
• 大池（200+个IP）： 5-15分钟，或在每个请求轮换时无需冷却

1. 激进模式 (Speed-First)

• 每个请求轮换 (Per-request rotation)
• 最小延迟 (0.5-1 秒请求间隔)
• 大IP池 (500+ IP)
• 高并行度 (5-15 个线程)

⚠️ 风险： 封锁风险高，IP生命周期短，即使轮换也可能触发速率限制。

📊 适用场景： 一次性任务、抓取对反爬要求不高的网站。

2. 平衡模式 (Balanced)

• 基于请求轮换 — 每10-20个请求
• 中等延迟 (2-5 秒请求间隔)
• 中等IP池 (100-300 IP)
• 适度并行度 (5-15 线程)

✅ 优点： 速度和稳定性的良好平衡，适用于大多数任务。

📊 适用场景： 电子商务价格监控、定期抓取、长期项目。

3. 隐蔽模式 (Stealth-First)

• 基于时间轮换 — 每15-30分钟
• 大延迟 (5-15 秒请求间隔)
• 高质量代理池 (50-100 住宅IP)
• 低并行度 (1-3 线程)
• 模拟人类行为（随机延迟、用户操作）

✅ 优点： 封锁风险最小，长期稳定性高，看起来像真实用户。

📊 适用场景： 社交媒体、高防护网站、账户操作、竞争情报分析。

❌ 问题 1: 即使轮换了IP，仍收到 429 (请求过多) 错误

可能原因：

• 对同一域名轮换过于频繁
• 所有代理都来自同一子网（通过ASN检测）
• User-Agent 和其他请求头未轮换
• 冷却期设置过短

✅ 解决方案：

• 将请求间隔增加到 5-10 秒
• 使用住宅代理而非数据中心代理
• 添加 User-Agent、请求头和 TLS 指纹的轮换
• 将代理池大小增加 2-3 倍
• 在延迟中加入抖动（随机偏差）

❌ 问题 2: 每次更换IP都会弹出验证码 (CAPTCHA)

可能原因：

• 使用了信誉差的数据中心代理
• 过于激进的轮换策略引起怀疑
• 使用了公共（免费）代理
• 更换IP时浏览器指纹未同步更新

✅ 解决方案：

• 切换到住宅或移动代理
• 使用粘性会话代替持续轮换
• 集成验证码解决服务（如2Captcha, AntiCaptcha）
• 使用反指纹浏览器（如Playwright, puppeteer-extra-plugin-stealth）
• 在主要工作前对代理进行“预热”（发送少量简单请求）

❌ 问题 3: 授权后会话丢失

可能原因：

• IP轮换中断了会话
• Cookies未在请求间保存
• 粘性会话已过期

✅ 解决方案：

• 使用带有会话ID的粘性会话进行授权请求
• 延长粘性会话持续时间（1-24小时）
• 在会话间保存和重用Cookies/Token
• 混合方法：授权时粘性，数据收集时轮换

❌ 问题 4: 代理池消耗过快

可能原因：

• 过于激进的每个请求轮换
• 池子太小，无法应对工作量
• 未考虑冷却期

✅ 解决方案：

• 切换到突发轮换（每N个请求更换一次，而非每个请求）
• 根据工作量增加代理池大小
• 实现基于跟踪的冷却队列
• 使用提供商的轮换代理网关

❌ 问题 5: 抓取速度慢

可能原因：

• 代理延迟高 (High Ping)
• 使用串行处理而非并行处理
• 请求间隔设置过大
• 频繁轮换带来的连接建立开销

✅ 解决方案：

• 使用连接池和Keep-Alive
• 并行处理请求（使用线程/异步IO）
• 减少请求间隔
• 使用延迟较低的代理（按延迟过滤）
• 使用ISP代理而非住宅代理（如果对隐蔽性要求不高）

指标	正常值	出现问题（如果）
成功率	> 95%	< 85%
429 错误率	< 2%	> 10%
403/503 错误率	< 3%	> 15%
验证码率	< 1%	> 5%
平均响应时间	< 3 秒	> 10 秒
超时率	< 1%	> 5%
使用过的唯一IP	> 80% 池	< 50% 池

✅ 1. IP轮换应始终与其他技术结合使用

IP轮换只是一个组件。同时轮换 User-Agent、使用真实浏览器（Puppeteer/Playwright）、模拟人类行为、轮换请求头和 Cookies。

✅ 2. 关键任务使用住宅/移动代理

数据中心代理更便宜，但信誉度低。对于社交媒体、电子商务和高防护网站，请使用住宅或移动IP。

✅ 3. 实现优雅降级 (Graceful Degradation)

当错误率上升时，自动减慢请求速度、增加延迟、切换到更高质量的代理。适应性比固定配置更重要。

✅ 4. 在大规模部署前进行小规模测试

在运行大型抓取任务前，先在100-1000个请求上测试策略。确保成功率 > 95%，没有大规模封锁，速度可接受。

✅ 5. 遵守 robots.txt 和服务条款

道德抓取是长期成功的关键。遵守 robots.txt，不要过度请求服务器，不要在未经同意的情况下收集个人数据。

✅ 6. 投资高质量代理

免费或廉价代理从长远来看成本更高：速度慢、封锁频繁、数据丢失、安全风险。使用像ProxyCove这样信誉良好的供应商，并保证正常运行时间。

核心要点总结：

1. 没有万能方案 — 策略选择取决于网站类型、数据量、预算和速度要求
2. 三种主要类型： 基于时间的（稳定性）、基于请求的（速度）、随机的（隐蔽性）——应结合使用
3. 粘性会话至关重要 — 用于授权、购物车和多步骤流程。轮换代理用于大规模数据收集
4. 质量胜于数量 — 对于大多数任务，50个住宅IP优于500个数据中心IP
5. 监控是必须的 — 跟踪成功率、错误代码、响应时间，以便及时优化
6. 平衡速度与隐蔽性 — 激进的抓取带来短期收益，谨慎的抓取带来长期稳定
7. 通过供应商实现自动化 — 使用轮换网关而不是自己编写轮换代码，节省时间和精力
8. 保持适应性 — 安全系统在不断变化，您的策略也必须随之调整

🚀 启动抓取任务前的清单：

💼 商业用途：

如果您的业务依赖于Web抓取（价格监控、竞争情报、潜在客户开发），请不要在代理基础设施上节省开支。停机或封锁的成本将远远超过高质量代理和正确轮换策略的费用。

🎓 开发者用途：

一次性投入时间构建可靠的轮换系统，而不是不断解决封锁问题。使用现成库，记录指标，测试不同策略。自动化将带来巨大的回报。