周一至周五 09:00 - 18:00(UTC+08:00) ©2026 NST LABS TECH LTD. 保留所有权利。
如何在Python中轮换代理:2026完整指南
Lena Zhou Growth & Integration Specialist
如何在 Python 中轮换代理:2026 年完整指南
TL;DR
在Python中,代理轮换意味着在每个请求中从列表中选择一个不同的代理URL,而不是requests库的特殊模式。 requests只知道如何使用您为该调用提供的代理字典;您的代码决定了使用哪个代理。
一个以协议为键的proxies字典({"http": ..., "https": ...})是requests接受的唯一格式,可以在每个请求中传递或一次性设置在Session上。 每个请求的方式更安全,因为在每个循环迭代中更新session.proxies容易出错。
环境变量默默覆盖session.proxies。 requests文档警告说,环境中的HTTP_PROXY/HTTPS_PROXY优先于您在会话中设置的任何内容,本文重现了这种覆盖情况,以便您可以看到它的发生。
itertools.cycle能够以三行代码实现循环轮换,但它不知道何时代理失效。 生产环境中的轮换需要围绕它构建一个重试和跳过的循环,本文构建并运行了一个真实的代理失败示例。
urllib3的Retry类重试相同的连接;它不会在不同的代理之间轮换。 混淆这两者是轮换代码“似乎没有轮换”的常见原因 — Retry和代理轮换解决不同的问题,通常您需要两者。
轮换代理URL与这些URL的来源无关。 同样的循环适用于您为测试硬编码的三个IP地址或者来自付费提供商的大量池;本文中的代码经过验证,针对本地测试服务器,因此并不依赖于任何一个提供商。
在脚本中 "轮换代理" 的实际含义
轮换代理意味着通过不同的代理服务器发送不同的请求,以便没有单一的出站IP处理每个请求。Python或requests中没有自动轮换任何东西 — requests.get(url, proxies=proxy_dict)通过一个代理发送确切的一个请求,而决定在下一个调用中传递哪个proxy_dict的正是您的循环。这个区别很重要,因为很多轮换错误实际上是“循环从未真正更改字典”,而不是代理问题。本文的其他部分将从一个硬编码的代理开始,逐步构建一个线程安全的池,并具有自动故障转移,运行每个示例以真正的本地服务器进行测试,以便您可以看到实际的行为,而不是依赖代码片段。
安装所需的软件 您需要requests库,它会依赖urllib3并处理普通的HTTP代理,并且从requests 2.10+开始,如果您还安装了socks扩展,那么也支持SOCKS代理:
pip install requests
pip install "requests[socks]" # 仅在socks5://代理URL时需要
本文中的模式不需要其他软件包 — 轮换代码由标准库的几十行代码(itertools、random、threading、concurrent.futures)加上requests本身组成。
在轮换任何内容之前配置单个代理 直接回答标题:requests期待一个包含"http"和"https"键的字典,每个键指向一个代理URL,您可以将该字典传递给单个调用或附加到Session上。requests文档 将每个请求的形式显示为:
import requests
proxies = {
"http" : "http://10.10.1.10:3128" ,
"https" : "http://10.10.1.10:1080" ,
}
requests . get ( "http://example.org" , proxies = proxies )
每个Session的形式为session.proxies.update(proxies),然后是session.get(...)。凭据直接放在URL中,例如http://user:pass@host:port。同样的文档提醒了一个值得知道的陷阱,在您在此基础上构建任何内容之前需要了解:文中明确指出“提供的值将被环境代理覆盖”,这意味着HTTP_PROXY或HTTPS_PROXY环境变量优先于您在session.proxies上设置的任何内容。这是真的,而且很容易偶然触发 — 在一台机器(或CI运行器)上已经导出了HTTP_PROXY,运行以下代码将重现这一点:
import os
import requests
os . environ [ "HTTP_PROXY" ] = "http://127.0.0.1:8002"
session = requests . Session ( )
session . proxies . update ( { "http" : "http://127.0.0.1:8001" } )
resp = session . get ( "http://internal.test/" , timeout = 5 )
print ( resp . json ( ) [ "served_by" ] )
该会话被告知使用8001端口的代理;但环境变量依然占了上风,而会话中的所有请求都通过了8002端口。如果您的轮换代码是一个Session并且似乎忽略了您的代理列表,请在检查轮换逻辑之前检查您的环境变量。
快速浏览
在三个本地服务器上测试旋转逻辑证明了循环是有效的,但它无法告诉你在负载下真实代理的表现 — Nstproxy 的网关为你提供了大量真实 IP 池,一旦你准备好,就可以将相同的代码指向这些 IP。
最简单的旋转循环遍历固定的代理 URL 列表,并在每次调用时给 requests 一个新的代理。该示例是在为本文启动的三个本地 HTTP 服务器(端口 8001–8003)上运行的,每个服务器在响应中自我标识,因此旋转是可验证的,而不是假设的:
import itertools
import requests
PROXIES = [
"http://127.0.0.1:8001" ,
"http://127.0.0.1:8002" ,
"http://127.0.0.1:8003" ,
]
proxy_cycle = itertools . cycle ( PROXIES )
for i in range ( 6 ) :
proxy = next ( proxy_cycle )
resp = requests . get (
"http://internal.test/products" ,
proxies = { "http" : proxy , "https" : proxy } ,
timeout = 5 ,
)
print ( f"请求 { i + 1 } : 通过 { proxy } 发送 -> { resp . json ( ) [ 'served_by' ] } " )
请求 1: 通过 http://127.0.0.1:8001 发送 -> proxy-1
请求 2: 通过 http://127.0.0.1:8002 发送 -> proxy-2
请求 3: 通过 http://127.0.0.1:8003 发送 -> proxy-3
请求 4: 通过 http://127.0.0.1:8001 发送 -> proxy-1
请求 5: 通过 http://127.0.0.1:8002 发送 -> proxy-2
请求 6: 通过 http://127.0.0.1:8003 发送 -> proxy-3
itertools.cycle 是整个窍门:它是一个无限迭代器,永远循环回列表的开头,因此 next(proxy_cycle) 始终返回一个代理,而无需你跟踪索引。用真实的代理 URL 替换 PROXIES,循环的工作方式完全相同 — 它不知道这个列表来自三个本地测试服务器还是一个商业池,这正是先以这种方式进行测试的目的。
高级模式:故障切换和并发旋转 自旋转按轮询方式并不能回答每个旋转脚本最终都要面对的问题:当一个代理宕机时会发生什么?下面的模式从池中随机选择一个代理,捕获 requests 针对代理和连接失败引发的特定异常,并在不让整个请求失败的情况下移动到下一个候选:
import random
import requests
def get_with_rotation ( url , proxy_pool , max_attempts = 4 , timeout = 3 ) :
pool = list ( proxy_pool )
random . shuffle ( pool )
last_error = None
for attempt , proxy in enumerate ( pool [ : max_attempts ] , start = 1 ) :
try :
resp = requests . get (
url ,
proxies = { "http" : proxy , "https" : proxy } ,
timeout = timeout ,
)
resp . raise_for_status ( )
return proxy , resp
except ( requests . exceptions . ProxyError ,
requests . exceptions . ConnectionError ,
requests . exceptions . Timeout ) as exc :
print ( f"尝试 { attempt } : { proxy } 失败 ( { exc . __class__ . __name__ } ), 正在旋转" )
last_error = exc
continue
raise last_error
ProxyError 被文档记录为专门针对代理相关故障引发的 ConnectionError 子类,因此它与更一般的 ConnectionError 和 Timeout 一起捕获。对四个地址(三个真实的本地服务器和一个没有监听的端口)运行时,在一次运行中产生了以下结果:
尝试 1: http://127.0.0.1:8004 失败 (ProxyError), 正在旋转
通过 http://127.0.0.1:8003 成功 -> proxy-3
死地址立即失败,循环在没有崩溃脚本的情况下继续。由于 random.shuffle 在每次调用时重新排列池子,因此如果洗牌恰巧将一个有效的代理放在第一位,不同的运行可以在第一次尝试时成功 — 这两种结果都是该模式的正确行为。
来自多个线程的旋转添加了一个额外的要求:无论“下一个代理”是什么,必须能安全地从多个线程同时调用。一个在 threading.Lock 后面的共享 itertools.cycle,通过 ThreadPoolExecutor 驱动,保持了轮询顺序在线程并发下的不变:
import itertools
import threading
from concurrent . futures import ThreadPoolExecutor , as_completed
import requests
PROXIES = [ "http://127.0.0.1:8001" , "http://127.0.0.1:8002" , "http://127.0.0.1:8003" ]
_lock = threading . Lock ( )
_cycle = itertools . cycle ( PROXIES )
def next_proxy ( ) :
with _lock :
return next ( _cycle )
def fetch ( path ) :
proxy = next_proxy ( )
resp = requests . get ( f"http://internal.test { path } " , proxies = { "http" : proxy , "https" : proxy } , timeout = 5 )
return path , proxy , resp . json ( ) [ "served_by" ]
路径 = [f"/item/{i}" for i in range(9)]
使用 ThreadPoolExecutor(max_workers=4) 作为池:
futures = [pool.submit(fetch, p) for p in paths]
对于 future 在 as_completed(futures):
打印(future.result())
九个请求通过四个工作线程仍然准确地每个代理达到三个请求,顺序执行,因为锁序列化访问共享迭代器,即使请求本身是并发运行的。 ThreadPoolExecutor 是标准库 concurrent.futures 模块的一部分——这个模式不需要额外的依赖。
诚实的限制:这个轮换代码不能解决的问题 这段代码决定了 哪个代理 URL 去每个请求;它对这些 URL 的来源或质量没有任何意见。三个无效 IP 的列表将以与三个有效 IP 完全相同的方式进行轮换——上面的故障转移循环只证明它可以从 一些 坏代理中恢复,而并不能证明您的特定列表是可用的。
urllib3 中的 Retry 不能替代本文中的轮换逻辑。其自身参数——total、backoff_factor、status_forcelist——描述的是对 相同 请求在 相同 连接上进行重试,并在尝试之间有一个后退延迟;在 urllib3.util.Retry 中没有任何内容能切换到不同的代理。如果您在设置了一个代理的会话上安装一个配置了 Retry 的 HTTPAdapter,每次重试仍然会通过那个相同的代理。切换到不同的代理以应对失败就是上面的 get_with_rotation 函数所做的,而这两种技术是应该结合使用,而不是相互选择。
这里的例子没有处理代理身份验证速率限制、粘性会话语义或每个国家的路由——这些是 URL 背后任何代理服务的属性,而不是 itertools.cycle 或重试循环能添加的功能。
解决常见轮换错误 requests.exceptions.ProxyError 表示与代理的连接失败——代理已关闭、端口错误,或防火墙阻止了它。这是上面的故障转移示例中触发的无效端口。
**requests.exceptions.ConnectionError**没有更具体的 ProxyError 子类型,通常意味着代理接受了连接,但无法到达目标站点,这通常显示为一个接近其可用寿命终点的代理,如果您从一个轮换池中提取。
轮换看起来没有发生 ,即使您的循环看起来正确——首先检查 HTTP_PROXY 或 HTTPS_PROXY 环境变量,按照本文早些时候的实时复现;它默默覆盖了 session.proxies,使每个请求都通过相同的地址,而不管您循环选择了什么。
每个请求在相同的 timeout 值处超时 ,而不是在无效代理上快速失败——专门为检查新池的健康降低 timeout 参数,因为一个慷慨的超时时间是为了真实请求而设定的,会使每次轮换尝试中的无效代理挂起整个持续时间。
将轮换代码指向真实的代理池 本文中的循环通过您提供的任何列表进行轮换,切换到商业网关意味着更改列表,而不是逻辑。 Nstproxy 是一个代理基础设施提供商,提供住宅、数据中心、静态 ISP、IPv6 和移动 IP 池,通过单个网关主机而不是您自己管理的单个 IP 列表进行 HTTP、HTTPS 或 SOCKS5 的访问。其 Residential Lite Proxies 产品页面直接记录了连接模式:一个固定的网关主机和端口,实际的 IP 轮换发生在提供商一侧的那个地址后面,而不是在您的 Python 列表中——在将其连接到上面的示例之前,请查看 Nstproxy 的文档 获取当前主机、端口和身份验证参数。这适用于希望在不维护和检查自己个别代理 IP 列表的情况下获得本文中轮换行为的团队。权衡的是,您信任提供商在网关后面的轮换行为,而不是像上面的 itertools.cycle 和故障转移示例那样直接控制它。
一个网关主机而不是一个维护列表 ——本文基本示例中的代码简化为一个指向那个网关地址的单个代理字典,而不是多个列表;在使用之前检查上面的文档链接以获取确切的当前主机和端口。
超过 5000 万个住宅 IP,覆盖 200 多个国家和地区 ——根据 Residential Lite 产品页面,这意味着在网关后面的轮换从一个比大多数自行管理的列表更大的池中抽取。
HTTP、HTTPS 和 SOCKS5 支持 — 与本文中 requests 示例所使用的协议相同,因此请求代码本身无需更改。
预付费套餐计费 — Nstproxy 发布的定价 以计量套餐的形式出售,而不是订阅,这一点很重要,如果您的轮换脚本只偶尔运行,而不是持续运行。
结论 在 Python 中轮换代理是一个在请求之间更改一个字典的循环,而不是安装的功能 — 本文中的一切,从三行 itertools.cycle 版本到线程安全的故障转移池,都是基于同一个 {"http": ..., "https": ...} 字典,即 requests 一直以来接受的。首先使用基本循环来确认您的请求代码确实有效,然后在您指向实际可能失败的代理时添加故障转移循环,只有在单请求延迟成为您的实际瓶颈时,才考虑使用线程。
常见问题 问:我需要特殊库来轮换 Python 中的代理吗,还是 requests 自己处理?
您不需要特殊库 — requests 每次调用仅接受一个代理字典,轮换是您自己的代码中的循环,负责更改每次请求传递的字典,文中已经展示。
首先检查 HTTP_PROXY 或 HTTPS_PROXY 环境变量,因为 requests 自身的文档确认它会静默覆盖您在 session.proxies 中设置的任何代理,这在本文中有体现。
问:urllib3 的 Retry 类会为我轮换代理吗?
不会 — Retry 会对相同的连接重试相同的请求,并不会切换到不同的代理; 您需要类似于本文的故障转移示例的轮换循环,而不是以此替代它。
只要选择“下一个代理”的过程受到锁的保护,就安全,例如上面用 threading.Lock 保护的 itertools.cycle 示例; 从多个线程访问未保护的共享迭代器通常是并发下微妙轮换错误的根源。
问:轮换我自己的代理列表与使用提供商的轮换网关有什么区别?
轮换您自己的列表意味着您的代码追踪哪些代理处于活动状态并在它们之间进行选择,而轮换网关在提供商的一台固定主机和端口后面完成相同的工作,这是 Nstproxy 为其住宅轻量代理记录的模式。
立即访问住宅、数据中心、IPv6 与 ISP 高质量代理池。 创建免费账号并立即试用 ->