返回博客
指南
Raluca PenciucLast updated on May 8, 20262 min read

如何在 Python 中旋转代理

如何在 Python 中旋转代理
简而言之:本指南将详细介绍如何在 Python 中实现端到端的代理轮换:选择合适的代理类型,构建并验证代理池,然后使用 itertools.cycle,或使用 random.choice,或通过 aiohttp。我们还将 IP 轮换与 User-Agent 轮换相结合,并添加了状态感知重试机制,确保单个故障代理不会导致整个抓取任务失败。

如果您的 Python 爬虫昨天还运行正常,今天却开始返回 403、429 错误或空白页面,那几乎可以肯定您的 IP 已被限流或封禁。大多数团队会选择的解决方案就是代理轮换,而掌握如何在 Python 中轮换代理,对于任何希望将项目从业余脚本扩展到生产环境的人来说,都是一项必经的考验。

在 Python 中进行代理轮换,意味着根据预定时间表或随机方式,在每次请求时更改外发 IP,使每个请求看起来都像是来自不同的机器。如果操作得当,它能将负载分散到多个 IP 上,规避基于 IP 的速率限制,并让反机器人系统更难识别爬虫流量。如果操作不当,使用过期的可用 IP 列表并采用一刀切的 try/except,只会将一个被封的 IP 变成一堆被封的 IP。

本文将以实操指南的形式讲解如何在 Python 中实现代理轮换。我们将选择代理类型、构建经过验证的代理池、通过 Requests 发送请求,并详细演示三种轮换策略(顺序、随机、异步)。我们将结合 IP 轮换与请求头轮换,添加真实的错误处理机制,并以“购买 vs 自建”的客观对比作为结尾。

什么是代理轮换,以及为何您的 Python 爬虫需要它

代理通过中间IP隐藏您的真实IP,但单个静态代理仍是一个IP,目标服务器仍可能对其进行速率限制并封禁。代理轮换会在每次请求或每次会话中切换外发IP,从而使同一个爬虫看起来像是来自多个来源。

这一点至关重要,因为反机器人系统主要依赖速率限制机制,会在特定时间段内限制每个 IP 的请求次数,超过阈值后返回 429 状态码。通过在健康的代理池中轮换,每个 IP 的请求量都能保持在阈值之下,从而避免因单个 IP 被封禁而导致整个任务中断。

轮换前请选择合适的代理类型

代理轮换的效果取决于所选IP的质量。若选择错误的代理类型,团队可能会耗费数周时间调试逻辑,却始终无法突破目标服务对流量的封锁。

代理类型

速度

封禁风险

成本

最适合

数据中心

最快

受保护站点数量多

最低

公共API,防御较弱

住宅

中等

中高等

电子商务、搜索结果页面、基于地理位置的页面

移动端(4G/5G)

最慢

最低

最高

社交媒体、应用API、硬性目标

ISP(静态住宅)

中上

长时会话、账户抓取

在 Python 中如何轮换代理的首要决定并非算法,而是将代理池与防御机制相匹配。

为代理轮换配置 Python 环境

在虚拟环境中使用 Python 3.8 及以上版本。安装 Requests 和 aiohttp,并将代理保存在纯文本文件中,以便轮换器能实时加载。

mkdir proxy_rotator && cd proxy_rotator
python -m venv .venv && source .venv/bin/activate
pip install requests aiohttp
touch app.py proxies.txt

构建并验证有效的代理列表

您可以整合 proxies.txt (免费代理聚合器、GitHub镜像),或从付费池加载凭据。无论哪种方式,都应预料到在首次请求前会有相当一部分代理失效,尤其是在免费列表中,其中大多数条目可能已被热门目标网站屏蔽。

请采用每行一个条目的格式 http://host:porthttp://user:pass@host:port,然后通过 IP 回显接口进行验证:

import requests

def validate(proxy, timeout=5):
    try:
        r = requests.get("https://httpbin.io/ip",
                         proxies={"http": proxy, "https": proxy},
                         timeout=timeout)
        return r.ok and proxy.split("@")[-1].split(":")[0] in r.text
    except requests.RequestException:
        return False

with open("proxies.txt") as f:
    pool = [p.strip() for p in f if p.strip() and validate(p.strip())]

IP 匹配检查可识别会直接透传您真实地址的透明代理。若需更深入的检测,请对真实目标页面进行 ping 测试,而非仅 httpbin.io/ip.

使用 Requests 通过代理发送单次请求

在轮换任何代理之前,请确保有一个代理能实现端到端通信。Requests 支持 proxies 字典 get()Session;通常同一 URL 适用于两者 httphttps 键。

import requests

proxy = "http://user:pass@host:port"   # auth is embedded in the URL
proxies = {"http": proxy, "https": proxy}

with requests.Session() as s:
    s.proxies.update(proxies)
    r = s.get("https://httpbin.io/ip", timeout=10)
    print(r.status_code, r.json())

若您希望将代理配置从代码中分离出来,请设置 HTTP_PROXYHTTPS_PROXY 环境变量;Requests 会自动读取它们。免费代理通常会引发 SSLError: CERTIFICATE_VERIFY_FAILED ,因为它们会拦截 TLS 连接。作为临时解决方法,你可以传入 verify=False,但这仅应视为调试工具,而非生产环境配置,因为它会完全禁用证书验证。

如何在 Python 中轮换代理:三种策略(顺序、随机和异步)

一旦单次请求成功,如何在 Python 中轮换代理的问题就变成了可预测性、隐蔽性和吞吐量之间的权衡。以下三种模式几乎涵盖了所有实际的爬虫场景。

使用 itertools.cycle 进行顺序轮换

顺序轮换按顺序遍历代理池并循环回到起始位置,从而均匀分配流量。这是最容易理解的模式,因为下一个 IP 地址总是可知的。

import itertools, requests

with open("proxies.txt") as f:
    proxies = [p.strip() for p in f if p.strip()]

pool = itertools.cycle(proxies)

for _ in range(8):
    proxy = next(pool)
    r = requests.get("https://httpbin.io/ip",
                     proxies={"http": proxy, "https": proxy},
                     timeout=10)
    print(proxy, r.status_code)

其缺点在于,这种确定性的顺序本身就是一种指纹。如果防御方在几秒内从同一个浏览器指纹中看到 IP A、B、C、D、A、B、C、D,他们就可以将整个池标记为可疑。顺序轮换在 IP 池较大且每个 IP 的延迟较长时效果最佳。

使用 random.choice 的随机轮换

随机轮换通过每次请求随机选择一个代理来打破模式,这使得流量更难建立关联。

import random, requests

with open("proxies.txt") as f:
    proxies = [p.strip() for p in f if p.strip()]

for _ in range(8):
    proxy = random.choice(proxies)
    r = requests.get("https://httpbin.io/ip",
                     proxies={"http": proxy, "https": proxy},
                     timeout=10)
    print(proxy, r.status_code)

其缺点是使用不均衡:小型池会过度使用某些 IP,而让其他 IP 闲置。为了获得更好的平衡,可以使用 random.sample(proxies, len(proxies)) ,然后重新洗牌。这既能保持请求不可预测性,又能分散负载。

使用 aiohttp 和 asyncio 实现异步轮询

当代理池规模超过几十个 IP 时,串行验证会成为瓶颈。异步轮询可在单个线程中并发处理多个请求,这能大幅缩短验证时间,并让工作池能够快速处理任务列表,而不会因慢速代理而阻塞。

import asyncio, aiohttp

CONCURRENCY = 20
TIMEOUT = aiohttp.ClientTimeout(total=10)

async def check_proxy(session, proxy, sem):
    async with sem:
        try:
            async with session.get("https://httpbin.io/ip",
                                   proxy=proxy, timeout=TIMEOUT) as r:
                return proxy, r.status, await r.text()
        except (aiohttp.ClientError, asyncio.TimeoutError) as e:
            return proxy, None, str(e)

async def main(proxies):
    sem = asyncio.Semaphore(CONCURRENCY)
    async with aiohttp.ClientSession() as session:
        tasks = [check_proxy(session, p, sem) for p in proxies]
        return await asyncio.gather(*tasks)

with open("proxies.txt") as f:
    proxies = [p.strip() for p in f if p.strip()]
results = asyncio.run(main(proxies))

信号量限制了同时处理的请求数量,从而避免耗尽文件描述符或触发目标服务器的突发流量限制。 aiohttp 暴露了一个按请求 proxy= 参数,aiohttp 高级客户端文档对此进行了详细说明,并涵盖了认证和信任环境的行为。

将代理轮换与 User-Agent 和标头轮换结合使用

仅轮换 IP 地址仍会泄露指纹。如果 200 个不同的 IP 地址发送相同的默认 python-requests/2.31.0 User-Agent,反机器人系统可以立即将它们关联起来。

在轮换代理的同时轮换头部,并确保 Cookie 始终与设置它们的身份相关联:

import random

UAS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ...",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 13_5) AppleWebKit/605.1.15 ...",
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 ...",
]
LANGS = ["en-US,en;q=0.9", "en-GB,en;q=0.8", "de-DE,de;q=0.9,en;q=0.8"]

def rotated_headers():
    return {"User-Agent": random.choice(UAS),
            "Accept-Language": random.choice(LANGS),
            "Referer": "https://www.google.com/"}

在逻辑会话的生命周期内,将一个 User-Agent 和 Cookie 集合绑定到一个代理,并在切换身份时将它们一同轮换。

生产级错误处理与代理健康检查

大多数初学者在遇到任何错误时都会丢弃代理。这不仅会浪费刚被限流的IP,还会将故障代理与要求减速的目标站点混为一谈。

将响应代码视为信号。根据 RFC 6585,429 表示请求过多,而非代理已死:应暂缓请求并使用同一 IP 重试。遇到 407 或重复连接错误时应弃用,并将故障代理隔离,以便在冷却期后重新检查。

import time, random
from collections import defaultdict

class ProxyManager:
    def __init__(self, proxies, max_fail=3, cooldown=300):
        self.live, self.dead = list(proxies), {}
        self.fails = defaultdict(int)
        self.max_fail, self.cooldown = max_fail, cooldown

    def get(self):
        self._revive()
        return random.choice(self.live) if self.live else None

    def report(self, proxy, status=None, error=None):
        if status == 429 or (status and 500 <= status < 600):
            time.sleep(min(2 ** self.fails[proxy], 30))   # keep, back off
        elif status == 407 or error:
            self.fails[proxy] += 1
            if self.fails[proxy] >= self.max_fail:
                self.live.remove(proxy)
                self.dead[proxy] = time.time() + self.cooldown

    def _revive(self):
        now = time.time()
        for p, t in list(self.dead.items()):
            if now >= t:
                self.live.append(p); self.dead.pop(p); self.fails[p] = 0

根据 max_fail 并根据您的流量特征调整基础延迟,而非盲目采用默认值。

手动轮换与托管轮换:选择适合您的路径

自行开发轮换器适用于学习和小型任务。但当规模扩大时,它将演变为一个独立的产品:代理更新、有效性验证、重试机制,以及当目标服务器更新其架构时需要值班维护。

托管式轮换代理或爬虫 API 通过单一接口隐藏了这些复杂机制,并按成功请求计费。

Signal

精简 DIY

精简托管

池大小

< 100 个 IP

数千+

目标难度

防御较弱

电商平台、搜索结果页面、社交媒体

SLA要求

尽最大努力

可预测的成功率

学习如何在 Python 中轮换代理的目的,并非为了永久维护一个轮换器;而是为了知道何时手动轮换就足够了,何时需要委托给系统处理。

关键要点

  • 在调整轮换逻辑前,需确保代理类型与目标网站匹配;若针对高安全性的网站轮换廉价的数据中心 IP,注定是一场徒劳的斗争。
  • 可靠的 Python 代理轮换配置应基于经过验证的代理池(而非随机列表),并在冷却期后重新检查失效的代理,因为免费代理池会不断在可用和失效状态之间循环。
  • 使用 itertools.cycle 实现可预测的分布, random.choice 用于隐蔽访问, aiohttp 配合 asyncio 实现高吞吐量验证和并发抓取。
  • 将 User-Agent 和标头值与 IP 地址一同轮换,以免在 200 个不同代理背后泄露固定的指纹。
  • 构建一个具备状态感知能力的轮换器:在遇到 429 和 5xx 状态码时自动回退,遇到 407 或重复连接错误时断开连接,并将不良代理隔离,而不是对每个异常都采用相同的方式处理。

常见问题

自己轮换代理与使用轮换代理网关或爬取 API 有什么区别?

自行构建轮换器意味着您完全掌控代理列表、验证、重试逻辑和地理路由。轮换代理网关仅提供一个端点,在每次请求时为您自动选择 IP;而爬虫 API 还会处理浏览器渲染、验证码识别和解锁操作。DIY 模式赋予您最大控制权;网关和 API 则以牺牲部分控制权为代价,大幅减少基础设施代码量。

对于典型的爬取任务,我的轮换池中需要多少个代理?

一个实用的起始经验法则是:每个并发任务分配一个有效的 IP,并额外预留 5 到 10 倍的 IP 作为备用,以应对 IP 流失、封禁和代理失效的情况。几千次请求的小型任务可以使用 20 到 50 个经过验证的住宅 IP 运行;而针对高防护目标或数百万页面的抓取任务,通常需要数千个轮换 IP,以保持单个 IP 的请求频率较低。

为什么即使我每次请求都轮换IP,免费代理仍然会被封禁?

免费代理由许多运行各自爬虫的陌生人共享,因此这些 IP 通常在你使用之前就已经被热门目标列入黑名单。它们还会以明显的方式泄露信息,发送诸如 ViaX-Forwarded-For,与声称的IP不符,或破坏TLS。轮换无法修复已列入拒绝列表的IP。

我应该在每次请求时轮换代理,还是针对每个目标网站保持粘性会话?

当目标网站将状态与 IP 绑定时(例如登录页面、多步骤结账流程,或会发出大量子请求的 JavaScript 密集型流程),请使用粘性会话。在抓取无状态列表、搜索页面或产品数据源时,请按每次请求轮换。一种常见的模式是:每个逻辑会话使用一个 IP,然后为下一个会话使用新的 IP。

能否在 Selenium 或 Playwright 中复用此轮换模式,而非使用 Requests?

可以,但需要进行调整。这两款浏览器自动化工具均支持代理设置,但通常需要为每个代理启动一个浏览器会话,因为大多数驱动程序不允许在会话中途更改代理。建议创建一个浏览器工作池,每个浏览器绑定一个 IP 和 User-Agent,通过轮换工作池中的浏览器实例(而非单个浏览器内的代理变量)来实现轮换。

总结:从手动轮换到可靠的爬取

掌握 Python 中的代理轮换技巧,是任何希望突破单 IP 爬虫局限的开发者的基础技能。现在您已掌握了构建模块:为目标选择合适的代理类型,在信任代理池前进行验证,根据权衡选择顺序、随机或异步轮换,叠加 User-Agent 轮换,并使用状态感知管理器,以免单次 429 错误就烧毁一个健康的 IP。

更难的是在目标网站毫无预警地改变防御策略时,仍能保持系统平稳运行。免费代理列表会过期,住宅代理池需要重新平衡,而 429 状态码的判定规则也会不断变化。如果您希望将工程时间投入到数据处理而非基础架构维护上,WebScrapingAPI 的 Scraper API 可在单一接口后处理代理轮换、反机器人规避及重试机制,让您无需修改 Requests 或 aiohttp 代码,只需替换请求层即可。若用于学习或小型任务,可自行轮换 IP;当维护成本超过节省的开支时,便可借助托管服务层。

关于作者
Raluca Penciuc, 全栈开发工程师 @ WebScrapingAPI
Raluca Penciuc全栈开发工程师

Raluca Penciuc 是 WebScrapingAPI 的全栈开发工程师,主要负责开发爬虫、优化规避机制,并探索可靠的方法以降低在目标网站上的被检测概率。

开始构建

准备好扩展您的数据收集规模了吗?

加入2,000多家企业,使用WebScrapingAPI在无需任何基础设施开销的情况下,以企业级规模提取网络数据。