爬虫怎样使用代理?代理池对接、请求轮换与失败重试完整教程

爬虫跑着跑着IP就废了?先搞清楚代理到底怎么接
做数据采集这行,谁没被目标站点封过IP?你辛辛苦苦写了一整套解析逻辑,结果跑了不到两百个请求,对方直接给你返回403或者弹验证码。这时候你盯着终端日志发呆,心里就一个念头——得搞代理。
但代理这东西,真不是”找个IP塞进requests的proxies参数里”就完事了。你面对的是一个代理池,里面可能有几百上千个IP,有的能用有的不能用,有的延迟低有的卡得要命。怎么从池子里取、怎么轮换着用、用坏了怎么重试、重试几次该放弃——这一整套流程没理顺,你的爬虫就是”跑十分钟歇半小时”,效率低得让人想砸键盘。
这篇文章我就把这套流程从头到尾捋一遍,从代理池的对接方式,到请求轮换的策略设计,再到失败重试的兜底逻辑,全部用代码和实际场景讲清楚。你看完之后,基本就能把自己项目里的代理模块搭起来了。
代理池对接:三种常见模式,选对省一半事
先说一个很多人容易搞混的点:代理池的”对接”,本质上就是解决“我怎么拿到一个能用的代理地址”这个问题。市面上代理服务商提供的接入方式,归纳起来就三种:
第一种:API提取式。你调一个HTTP接口,传参数(比如要哪个省、存活多久),它返回一个或一批IP:Port。你拿到之后自己存、自己管、自己判断什么时候该换。这种方式灵活度最高,适合你对代理生命周期有精细控制需求的场景。网帆代理的短效动态代理和长效动态代理走的就是这个路子,你调接口提取,IP按你设定的存活时长在线,到期自动回收。
第二种:隧道代理式。你只需要配一个固定的隧道入口地址(一个IP:Port),后面所有请求都走这个入口。代理服务商在隧道内部帮你自动调度、轮换IP,你完全不用管”当前用的是哪个IP”。对开发者来说这是最省心的方式,代码里就写死一个proxy地址,剩下的事服务商的调度层处理。网帆代理的隧道代理产品就是干这个的,一次接入,后面IP怎么轮、什么时候换,你不用操心,后台还有可视化面板能实时看IP消耗和运行状态。
第三种:静态列表式。服务商直接给你一份IP列表(txt或者csv),你自己维护。这种方式最原始,适合代理数量不多、更新频率低的场景。但说实话,2024年还在用这种方式做高频采集的,基本都会被目标站点很快识别。
下面这张表帮你快速对比:
| 对接模式 | 你的工作量 | 适合场景 | 典型延迟 |
|---|---|---|---|
| API提取式 | 需自己管理IP生命周期、健康检测 | 需要精确控制地域、存活时长 | 毫秒级提取 |
| 隧道代理式 | 最小,配一个入口地址即可 | 高频采集、不想维护IP池 | 取决于隧道调度 |
| 静态列表式 | 最大,需定期更新列表 | 低频、小量级任务 | 无额外开销 |
我的建议是:如果你刚起步、代理用量不大,直接用隧道代理,省得自己写一堆管理逻辑。等你的采集规模上来了,对IP的地域分布、存活时长有精细化要求了,再切到API提取式,自己搭一套轻量级的代理池管理。
代理池对接的代码实操
下面用Python演示API提取式的对接流程。假设你用的是网帆代理的短效动态代理,通过API提取IP:
import requests
import time
import random
from collections import deque
class ProxyPool:
"""轻量级代理池管理器"""
def __init__(self, api_url, max_size=50, min_size=10):
self.api_url = api_url 服务商提供的提取接口
self.pool = deque() 可用代理队列
self.max_size = max_size
self.min_size = min_size
self.failed = {} 记录失败次数 {proxy: count}
def fetch_from_api(self, count=10, province=None, duration=10):
"""从服务商API提取代理"""
params = {
"count": count,
"duration": duration 存活时长,单位分钟
}
if province:
params["province"] = province
resp = requests.get(self.api_url, params=params, timeout=5)
if resp.status_code == 200:
ips = resp.json().get("data", [])
for ip in ips:
proxy = f"http://{ip['ip']}:{ip['port']}"
if proxy not in self.pool:
self.pool.append(proxy)
return len(ips)
return 0
def get_proxy(self):
"""从池中取一个代理,池子不够就补货"""
if len(self.pool) = max_fail:
if proxy in self.pool:
self.pool.remove(proxy)
del self.failed[proxy]
return True 已被踢出
return False
这段代码的核心思路就三个:取、用、反馈。从池子里随机取一个代理去发请求,成功了就放回池子,失败了就记一笔,连续失败三次直接踢出池子不再用。池子快空了就去API补货。逻辑不复杂,但能覆盖90%的日常场景。
如果你用的是隧道代理模式,代码就简单多了,根本不需要上面这一坨:
# 隧道代理模式:一个入口搞定所有
TUNNEL_PROXY = "http://tunnel.fanproxy.com:8888"
session = requests.Session()
session.proxies = {
"http": TUNNEL_PROXY,
"https": TUNNEL_PROXY
}
# 每次请求自动走隧道,IP由服务商后台调度
resp = session.get("https://target-site.com/data", timeout=10)
print(resp.status_code) 你不用关心当前用的是哪个IP
看到区别了吧?隧道模式你连”代理池”这个概念都不需要,代码干净很多。代价就是你没法精确控制”这次我要用杭州移动的IP”这种细粒度需求。所以选型的时候想清楚你到底需不需要这种控制力。
请求轮换:不是”换个IP”就完事,节奏很重要
很多人对”轮换”的理解停留在”每次请求换一个IP”。这么干确实能降低单IP被标记的概率,但如果你一秒钟换了50个IP,目标站点的WAF(Web应用防火墙)反而会觉得异常——正常用户不会这么跳。
轮换策略的核心是模拟真实用户的访问节奏。我总结了几种常用策略,你根据目标站点的严格程度选:
| 策略 | 具体做法 | 适用场景 | 风险点 |
|---|---|---|---|
| 固定周期轮换 | 每N个请求或每M秒换一次IP | 目标站点风控较松 | 周期太固定容易被识别规律 |
| 随机间隔轮换 | 每次请求后随机等待2~8秒再换IP | 中等风控站点 | 效率略低 |
| 失败触发轮换 | 正常请求不换,遇到403/429才换 | 风控较松、追求效率 | 连续失败时恢复慢 |
| 混合策略 | 基础随机间隔 + 失败立即换 + 每100请求强制换 | 高风控站点、长期运行 | 逻辑稍复杂 |
实际项目中我比较推荐混合策略,代码大概长这样:
import random
import time
class RotatingSession:
def __init__(self, proxy_pool, base_delay=(2, 6), force_rotate_every=100):
self.pool = proxy_pool
self.base_delay = base_delay
self.force_rotate_every = force_rotate_every
self.request_count = 0
self.current_proxy = None
def _should_rotate(self, force=False):
"""判断是否需要轮换代理"""
if force:
return True
if self.request_count >= self.force_rotate_every:
self.request_count = 0
return True
return False
def _rotate(self):
"""执行轮换:随机等待 + 取新代理"""
wait = random.uniform(self.base_delay)
time.sleep(wait)
self.current_proxy = self.pool.get_proxy()
self.request_count = 0
def get(self, url, kwargs):
"""带轮换逻辑的GET请求"""
if self.current_proxy is None or self._should_rotate():
self._rotate()
proxies = None
if self.current_proxy:
proxies = {"http": self.current_proxy, "https": self.current_proxy}
resp = requests.get(url, proxies=proxies, timeout=10, kwargs)
self.request_count += 1
if resp.status_code in (200, 301, 302):
self.pool.report_success(self.current_proxy)
else:
kicked = self.pool.report_failure(self.current_proxy)
if kicked or resp.status_code in (403, 429):
被踢了或者被限流,立即轮换
self._rotate(force=True)
return resp
注意几个细节:随机等待的时间区间不要设得太窄,2到6秒比1到2秒安全得多;force_rotate_every这个强制轮换阈值,你根据目标站点的容忍度调,一般50到200之间比较合理;遇到429(Too Many Requests)的时候不要硬刚,先等一会儿再换IP重试,否则你换十个IP也全是429。
失败重试:别傻等,要”聪明地等”
重试是爬虫里最容易被忽视但最影响稳定性的环节。你不可能假设每个请求都一次成功,网络抖动、代理IP刚好过期、目标站点偶尔抽风,这些都会导致请求失败。但重试不是”失败了就立刻再发一次”,那样只会把对方服务器打得更惨,也更容易触发风控。
我一般用指数退避 + 随机抖动的策略:
import time
import random
def smart_retry(func, max_retries=3, base_wait=2, max_wait=30):
"""
带指数退避的智能重试
func: 要执行的请求函数(无参数,内部闭包处理)
"""
last_exception = None
for attempt in range(max_retries):
try:
result = func()
if result.status_code == 200:
return result
elif result.status_code == 429:
被限流,退避时间加倍
wait = min(base_wait (2 attempt) + random.uniform(0, 1), max_wait)
print(f"[429] 被限流,等待 {wait:.1f}s 后重试 ({attempt+1}/{max_retries})")
time.sleep(wait)
elif result.status_code in (403, 503):
被拦截或服务不可用,换代理后重试
wait = min(base_wait (2 attempt) + random.uniform(0, 2), max_wait)
print(f"[{result.status_code}] 请求被拒,等待 {wait:.1f}s 后重试")
time.sleep(wait)
else:
return result 其他状态码不重试,直接返回
except (requests.exceptions.ConnectionError,
requests.exceptions.Timeout) as e:
last_exception = e
wait = min(base_wait (2 attempt) + random.uniform(0, 1), max_wait)
print(f"[网络异常] {e},等待 {wait:.1f}s 后重试 ({attempt+1}/{max_retries})")
time.sleep(wait)
所有重试都失败了
print(f"[放弃] 重试 {max_retries} 次后仍失败")
if last_exception:
raise last_exception
return None
这里有个关键点:429和403的重试策略不一样。429是”你太频繁了,歇会儿”,你退避等待就行,同一个代理还能用。403是”你这个IP我不认了”,这时候光等没用,必须换代理再试。所以你在重试逻辑里要区分对待,不能一刀切。
重试次数别设太多。三次是上限,超过三次说明这个代理或者这个目标站点当前状态就是不行,继续重试只是浪费时间和代理额度。三次都失败的话,把这条任务丢到延迟队列里,过几分钟再处理,比死磕强。
几个实际踩过的坑,省你几天调试时间
坑一:代理IP的存活时长比你以为的短。你提取的时候设了10分钟存活,但实际从提取到真正用上可能已经过了30秒,中间还有网络传输、DNS解析的时间。如果你的采集任务单轮跑下来超过8分钟,那后几个请求大概率用的是已经过期的IP。解决办法:要么把存活时长设得比你的任务周期长一些,要么在每轮任务开始前重新提取一批。
坑二:HTTPS请求的代理配置。很多人只配了http代理,结果访问https站点的时候代理根本没生效,请求直接走了本机IP。Python的requests里,proxies字典里http和https要分别写,而且https代理地址本身也要用http://开头(不是https://),这个坑我见过太多人踩了。
坑三:User-Agent和代理IP不匹配。你用了个杭州的代理IP,但User-Agent写的是Windows系统,Referer又是Mac的浏览器指纹。目标站点一比对,”这IP是杭州移动,怎么浏览器指纹是深圳联通的?”直接标记。所以你的请求头要和代理IP的地域、运营商尽量保持一致,至少别太离谱。
坑四:并发太高把代理池打穿。你开了50个线程,每个线程都在疯狂取代理、发请求,代理池里的IP瞬间被消耗完,后面的线程全在等补货。如果你的代理服务商支持无并发上限(网帆代理的短效动态代理就是单秒无并发上限、平均延迟0.03秒),那问题不大。但如果你用的是有并发限制的代理,线程数一定要控制在代理池容量之内,不然就是自己给自己制造瓶颈。
选代理服务商的时候看什么
代理这东西,”能用”和”好用”之间差着十万八千里。你选服务商的时候,别光看价格,下面这几个点比价格重要得多:
IP纯净度。一个IP如果之前被很多人用来干过各种事,目标站点早就把它标记了。你拿到手还没发请求就已经是”黑IP”。所以看服务商的IP来源是不是运营商直供,纯净度有没有具体数据。网帆代理这块标的是99.8%以上,三大运营商合规线路,3000万+动态IP储备,覆盖全国300多个省市,这个量级和来源在业内算是比较扎实的。
存活时长能不能自定义。你的采集任务可能跑3分钟,也可能跑30分钟,如果服务商只给你固定5分钟或固定30分钟,你要么浪费要么不够用。能按1到30分钟自由定制存活时长的,用起来才顺手。
延迟和并发。代理本身如果延迟高、并发有限制,你爬虫的整体效率就被卡住了。毫秒级的提取速度、无并发上限、平均延迟控制在0.03秒左右,这些指标直接决定你的采集吞吐量。
计费模式是否透明。有些服务商看着单价低,但提取有冷却间隔、并发有上限、超时了还计费,算下来实际成本比标价的贵不少。包量计费和按时长包月两种模式都有的,你可以根据用量灵活选,长期跑的话包月折算下来能省不少。网帆代理这两种模式都有,包量最低到0.0023元一个IP,包月长期用最低4.5折,没有隐形收费,这点比较实在。
如果你是刚接触代理、想先试试水,网帆代理注册后能领最高2000个免费测试IP,够你把整套代理池对接、轮换、重试的流程跑通验证一遍了。隧道代理那边也有免费体验,配个1V1客户经理帮你调通,不用自己对着文档猜半天。
常见问题
Q1:我的爬虫用代理后速度反而变慢了,怎么回事?
大概率是三个原因之一:第一,你选的代理延迟太高,每个请求多等了几百毫秒,累积起来就很明显;第二,你的并发数超过了代理池的承载能力,线程在排队等代理;第三,你的重试逻辑太激进,一个请求失败后立刻重试,重试又失败,反复循环把时间都耗在无效请求上了。排查方法:先单独测一下代理的延迟(用curl -o /dev/null -s -w “%{time_total}” 通过代理访问一个轻量页面),再检查你的线程数和代理池容量是否匹配,最后把重试的退避时间适当拉长。
Q2:代理IP用着用着突然全部失效了,是服务商的问题还是我的问题?
先别急着找客服。你先确认一下:你的代理存活时长是不是刚好到了?如果你设的是5分钟存活,而你的任务跑了6分钟,那IP过期是正常行为,不是故障。再检查一下你的提取接口调用是否正常返回(看HTTP状态码和响应体)。如果确认IP没过期、接口也正常,但所有代理都连不上,那可能是服务商侧的网络波动,这时候联系服务商的运维确认一下。网帆代理那边是7×24小时运维值守的,遇到这种情况直接找你的客户经理,响应比较快。
Q3:我需要精确到某个城市的代理IP,怎么实现?
在调用提取API的时候传地域参数就行。比如你要杭州的IP,参数里指定province=浙江、city=杭州。网帆代理支持精确到省、市、区县三级,你可以只提一个城市的,也可以设多个城市混播(比如同时提杭州、南京、上海的IP,池子里随机分配)。如果你用的是隧道代理模式,隧道入口一般也支持地域参数配置,具体看你接入时服务商给你的参数说明。
Q4:短效动态代理和隧道代理,我到底该选哪个?
一句话:你需不需要自己控制”当前用的是哪个IP”。需要,就选短效动态代理(API提取式),你自己管池子、自己决定什么时候换、换哪个地域的。不需要,就选隧道代理,你只管发请求,IP的事服务商后台帮你调度。从开发成本看,隧道代理省心得多,代码里就一个proxy地址,不用写池子管理、不用写健康检测。从灵活度看,短效动态代理上限更高,适合你的业务对IP地域、存活时长、使用节奏有精细要求的场景。两者不冲突,你甚至可以核心采集用短效动态代理,辅助任务用隧道代理,混搭着用。
