先说个扎心的现实：你的爬虫为什么总被封
做数据采集这行，谁还没被目标站点"教育"过几次？你辛辛苦苦写好的请求逻辑，跑了两三百次，IP就被拉黑了。页面返回403、429，或者直接给你弹一个验证码，再往后连响应都收不到。这时候你盯着终端日志发呆，心里就一个念头：得换IP，而且得换得聪明点。
代理IP这东西，说白了就是给爬虫套了个"马甲"。你每次请求出去，对方看到的不是你的真实出口，而是代理池里某个IP。但光有代理IP不够，你要是傻乎乎地用一个IP死磕，那跟不用没区别。真正能跑稳的爬虫，核心就三件事：重试、限速、自动换源。下面我把自己踩坑攒下来的写法摊开讲，都是能直接往项目里塞的代码。
重试机制：别一失败就放弃，但也别死磕
网络请求失败太正常了。代理IP可能刚好过期，目标服务器可能那一秒在重启，DNS解析可能抖了一下。你要是第一次失败就抛异常退出，那整个任务基本废了。但反过来，你要是无脑重试十次、二十次，那等于在告诉对方"你封我啊"。
我的做法是分层重试：先判断失败原因，再决定怎么重试。超时和连接拒绝，大概率是代理IP的问题，直接换IP重试；HTTP 429（请求过多）说明你打太快了，该退避；HTTP 403可能是IP被临时标记，换个IP再试一次；如果是5xx，那是对方服务器的事，等几秒再试就行。
import time
import random
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
class ProxyRetryHandler:
"""
分层重试：根据失败类型决定重试策略
"""
def __init__(self, proxy_pool, max_retries=3):
self.proxy_pool = proxy_pool 你的代理IP池对象
self.max_retries = max_retries
def request_with_retry(self, url, method="GET", kwargs):
last_exception = None
for attempt in range(self.max_retries):
每次重试都从池子里取一个新IP
proxy = self.proxy_pool.get_next()
proxies = {
"http": f"http://{proxy}",
"https": f"http://{proxy}"
}
try:
resp = requests.request(
method, url,
proxies=proxies,
timeout=(5, 15), 连接超时5s，读取超时15s
kwargs
)
429：对方限流了，退避后重试
if resp.status_code == 429:
wait = random.uniform(3, 8)
print(f"[429] 被限流，等待 {wait:.1f}s 后重试")
time.sleep(wait)
continue
403：IP可能被临时标记，换IP再试
if resp.status_code == 403:
print(f"[403] IP {proxy} 被标记，换源重试")
self.proxy_pool.mark_bad(proxy)
continue
5xx：对方服务器问题，短等待
if resp.status_code >= 500:
wait = random.uniform(2, 5)
print(f"[{resp.status_code}] 服务端异常，等待 {wait:.1f}s")
time.sleep(wait)
continue
return resp
except (requests.exceptions.ConnectTimeout,
requests.exceptions.ConnectionError) as e:
连接层失败，大概率代理IP本身有问题
print(f"[连接失败] {proxy} -> {e.__class__.__name__}")
self.proxy_pool.mark_bad(proxy)
last_exception = e
continue
except requests.exceptions.ReadTimeout:
读超时，可能是网络抖动，短等待
time.sleep(random.uniform(1, 3))
last_exception = e
continue
重试次数用完了
raise Exception(f"重试 {self.max_retries} 次后仍失败: {last_exception}")这里有个细节很多人忽略：每次重试一定要换IP。你同一个IP连续打三次403，第四次大概率还是403。换一个新IP，对方那边就是一个全新的"访客"，之前那些临时标记跟它没关系。
限速：不是越慢越好，是"像人一样"地慢
新手最容易犯的错误就是：要么全速跑，要么把间隔设成固定5秒。前者秒封，后者效率低到怀疑人生。真正好用的限速策略是随机化 + 自适应。
所谓随机化，就是每次请求之间的间隔不是一个固定值，而是一个区间内的随机数。比如你设的基准间隔是2秒，那实际间隔在1.2秒到3.5秒之间浮动。这样从流量特征上看，你的请求节奏就不像机器了。
自适应更关键。你一开始可以跑快一点，比如平均2秒一个请求。一旦连续出现两次429或者响应时间明显变长（比如从300ms飙到2000ms），说明对方开始"注意"你了，这时候自动把间隔拉长到4秒、6秒。等过了一两分钟，再慢慢缩回来。这个逻辑我一般用一个简单的滑动窗口来统计：
import time
import random
from collections import deque
class AdaptiveRateLimiter:
"""
自适应限速器：根据近期响应情况动态调整请求间隔
"""
def __init__(self, base_interval=2.0, min_interval=1.0, max_interval=15.0):
self.base_interval = base_interval
self.min_interval = min_interval
self.max_interval = max_interval
self.current_interval = base_interval
记录最近10次请求的"健康度"（1=正常，0=异常）
self.history = deque(maxlen=10)
self._last_request_time = 0
def wait(self):
"""在每次请求前调用，阻塞到允许发请求的时间点"""
elapsed = time.time() - self._last_request_time
加入随机抖动：±30%
jitter = self.current_interval random.uniform(0.7, 1.3)
if elapsed < jitter:
time.sleep(jitter - elapsed)
self._last_request_time = time.time()
def report(self, status_code, latency_ms):
"""
每次请求完成后调用，反馈结果
status_code: HTTP状态码
latency_ms: 本次请求耗时（毫秒）
"""
is_healthy = (status_code < 500) and (status_code != 429) and (latency_ms < 3000)
self.history.append(1 if is_healthy else 0)
计算近期健康率
if len(self.history) >= 5:
health_rate = sum(self.history) / len(self.history)
if health_rate < 0.6:
健康率低于60%，拉大间隔
self.current_interval = min(
self.current_interval 1.5, self.max_interval
)
elif health_rate > 0.9:
健康率高于90%，慢慢缩回来
self.current_interval = max(
self.current_interval 0.85, self.base_interval
)
使用示例
limiter = AdaptiveRateLimiter(base_interval=2.0)
for url in target_urls:
limiter.wait() 请求前：等
start = time.time()
resp = session.get(url, proxies=proxies, timeout=10)
latency = (time.time() - start) 1000
limiter.report(resp.status_code, latency) 请求后：反馈这套东西跑起来之后，你会发现一个现象：你的爬虫在"顺畅"的时候跑得挺快，一旦碰到对方开始限流，它会自动慢下来，等风头过了再加速。整个过程不需要你手动干预，也不需要写死一堆 if-else 去判断"现在该快还是该慢"。
自动换源：IP池不是摆设，得让它"活"起来
很多人拿到代理IP池之后，用法就是列表里轮着取，取完了从头再来。这其实是最粗糙的用法。真正好用的IP池管理，至少得做到三件事：标记坏IP、按权重分配、定期健康检查。
我一般把IP池设计成这样一个结构：
import threading
import time
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class ProxyEntry:
ip: str
port: int
region: str = "" 城市/省份
last_used: float = 0.0 上次使用时间戳
fail_count: int = 0 连续失败次数
is_bad: bool = False 是否被标记为坏IP
expire_at: float = 0.0 过期时间戳（短效IP用）
class ProxyPool:
"""
代理IP池：支持标记、轮换、健康检查
"""
def __init__(self, proxies: list[str]):
self._pool: list[ProxyEntry] = []
self._lock = threading.Lock()
self._cursor = 0
for p in proxies:
假设格式为 "ip:port"
ip, port = p.rsplit(":", 1)
self._pool.append(ProxyEntry(ip=ip, port=int(port)))
def get_next(self) -> str:
"""
取下一个可用IP，跳过坏IP和过期IP
"""
with self._lock:
now = time.time()
checked = 0
total = len(self._pool)
while checked < total:
entry = self._pool[self._cursor % total]
self._cursor += 1
checked += 1
跳过坏IP
if entry.is_bad:
continue
跳过已过期的短效IP
if entry.expire_at and now > entry.expire_at:
entry.is_bad = True
continue
entry.last_used = now
return f"{entry.ip}:{entry.port}"
全部不可用，重置坏标记（给它们一次机会）
for e in self._pool:
e.is_bad = False
e.fail_count = 0
再取一次
entry = self._pool[self._cursor % total]
self._cursor += 1
entry.last_used = now
return f"{entry.ip}:{entry.port}"
def mark_bad(self, proxy_str: str):
"""标记某个IP为坏IP"""
ip, port = proxy_str.rsplit(":", 1)
with self._lock:
for entry in self._pool:
if entry.ip == ip and entry.port == int(port):
entry.fail_count += 1
if entry.fail_count >= 3:
entry.is_bad = True
break
def health_check(self, check_url="http://httpbin.org/ip"):
"""
定期健康检查：主动探测IP是否还活着
建议每5分钟跑一次
"""
import requests
with self._lock:
for entry in self._pool:
if entry.is_bad:
continue
try:
r = requests.get(
check_url,
proxies={"http": f"http://{entry.ip}:{entry.port}"},
timeout=5
)
if r.status_code == 200:
entry.fail_count = 0
else:
entry.fail_count += 1
if entry.fail_count >= 3:
entry.is_bad = True
except Exception:
entry.fail_count += 1
if entry.fail_count >= 3:
entry.is_bad = True这里有个实操建议：短效IP和长效IP的管理策略不一样。短效IP（比如3分钟、5分钟就过期的那种）你不需要做太复杂的健康检查，因为它本身存活时间就短，过期了直接丢弃就行，重点放在"取的时候判断有没有过期"。长效IP或者固定IP则不同，它们存活时间长，但可能会因为运营商侧的变动突然不可用，这时候定期健康检查就很有必要了，我一般设成每5分钟跑一轮，把连续三次探测失败的IP标记掉。
说到IP池的选型，我目前项目里用的是神龙HTTP的短效动态IP池。它的IP资源是三大运营商正规授权的，3000万+的池子每天更新去重，延迟确实低，我实测在华东节点平均延迟在30ms左右，跑高并发采集的时候不容易卡。而且它支持按城市级定位取IP，比如你采集的是某个本地生活平台的数据，你指定IP来源是目标城市，这样请求出去的时候地理位置信息是对得上的，不容易触发风控。计费方式是包量或者包时，个人开发者用包量比较划算，企业跑大任务用包时更省心。
把三样东西串起来：一个能跑的最小框架
上面三块东西——重试、限速、换源——单独看都不难，但真正要跑起来，得把它们捏在一起。下面是一个精简版的采集循环，你看着改就行：
import time
import random
import requests
def run_crawler(urls, proxy_pool, limiter, retry_handler):
"""
urls: 待采集的URL列表
proxy_pool: ProxyPool 实例
limiter: AdaptiveRateLimiter 实例
retry_handler: ProxyRetryHandler 实例
"""
success_count = 0
fail_count = 0
for i, url in enumerate(urls):
1. 限速：请求前等
limiter.wait()
2. 带重试的请求（内部会自动换IP）
try:
resp = retry_handler.request_with_retry(url)
3. 反馈给限速器
limiter.report(resp.status_code, resp.elapsed.total_seconds() 1000)
if resp.status_code == 200:
解析数据...
parse_data(resp.text)
success_count += 1
else:
fail_count += 1
except Exception as e:
fail_count += 1
print(f"[失败] {url} -> {e}")
每处理100条，打印一次进度
if (i + 1) % 100 == 0:
print(f"进度: {i+1}/{len(urls)} | 成功: {success_count} | 失败: {fail_count}")
每500条，触发一次IP池健康检查（异步跑，不阻塞主流程）
if (i + 1) % 500 == 0:
import threading
t = threading.Thread(target=proxy_pool.health_check, daemon=True)
t.start()
print(f"采集完成。成功: {success_count}, 失败: {fail_count}")这个框架看起来不复杂，但跑起来之后你会发现，它比你之前"裸跑"的爬虫稳定太多了。以前跑200条就崩，现在跑几千条、几万条，失败率能压到5%以内，而且失败的也大多是目标页面本身的问题（比如页面结构变了），不是IP被封导致的。
几个容易踩的坑，提前说
第一，别在请求头里暴露太多信息。你用了代理IP，但User-Agent还是默认的Python-requests/2.28，Referer是空的，Accept-Encoding也不对。对方一看就知道是脚本。至少把这几个头伪装成正常浏览器的样子，这个成本很低，收益很大。
第二，短效IP的过期时间要留余量。比如你用的是5分钟有效期的IP，你别在第4分50秒还在用它。我一般设一个"安全窗口"，到期前30秒就视为过期，主动换下一个。上面代码里 expire_at 的判断就是干这个的。
第三，别所有请求都走同一个代理出口。如果你同时跑多个采集任务，每个任务用独立的IP池或者至少是池子里不同的IP段，避免多个任务共享同一个IP导致"交叉污染"——A任务把IP打废了，B任务跟着遭殃。
第四，日志一定要记。每次请求用了哪个IP、状态码多少、耗时多少、是否重试了，这些全记下来。不是给你自己看的，是出了问题之后你才知道该调哪个参数。没有日志的爬虫，调参全靠猜。
常见问题
Q1：我用的短效IP只有3分钟有效期，但一个页面请求加上解析要40秒，3分钟够不够用？
完全够。3分钟是IP的存活时间，不是单次请求的超时时间。你在这3分钟里可以发几十次请求都没问题。真正需要注意的是：如果你一个任务要跑超过3分钟，那中途IP过期了，你的重试逻辑会自动从池子里取一个新IP接着跑，业务层是无感的。只要你的 ProxyPool.get_next() 里做了过期判断就行。
Q2：代理IP的可用率标称99.9%，但我实际跑下来感觉只有95%左右，正常吗？
正常。标称可用率是在"健康检查"维度上说的，也就是IP能不能连通、能不能返回正常响应。但你实际采集的时候，还叠加了目标站点的因素——对方可能对你的请求模式敏感，可能那个IP刚好被对方临时标记了。所以实际"有效可用率"会比标称值低几个点，这是正常的。你真正要关注的指标不是"IP能不能通"，而是"用这个IP发请求，目标站点返回200的比例"。如果这个比例持续低于90%，那说明你的限速策略或者请求头伪装需要调一调，不一定是IP本身的问题。
Q3：我同时跑5个采集任务，IP池怎么分配比较合理？
最简单的做法是按任务隔离：每个任务从池子里取IP的时候，用不同的起始游标，或者干脆把池子按IP段切成5份，每个任务用一份。这样任务之间不会互相抢IP，也不会出现A任务把某个IP打废了、B任务又取到同一个IP的情况。如果你的IP池够大（比如神龙HTTP那种千万级资源池），每个任务分到几百万个IP，完全不用担心不够用。如果池子比较小，那就用共享池 + 标记机制，坏IP全局标记，所有任务都跳过。
Q4：固定IP和短效动态IP，我到底该用哪个？
看你的场景。如果你的采集频率不高（比如一天跑一两次，每次几百条），而且目标站点风控不严格，固定IP就够了，省心，不用管过期、不用频繁换。神龙HTTP的固定IP池是ISP正式分配的，纯净度99.83%以上，存活时间长，按个数买、包时计费，适合这种"低频但求稳"的场景。但如果你跑的是高频采集（比如每分钟几十次请求），或者目标站点风控比较敏感，那必须用短效动态IP，频繁换IP才能把"同一个IP反复访问"的特征打散。神龙HTTP的短效池支持3/5/10/15/30分钟多种时效，你可以根据目标站点的封禁策略选一个合适的档位，延迟低、并发提取快，跑起来不卡。
最后说两句
代理IP在爬虫里的角色，不是"多一个工具"，而是整个采集架构的地基。你重试写得再漂亮、限速调得再精细，如果底层IP质量不行、可用率低、延迟高，那上面那层逻辑就是在沙子上盖楼。所以选IP服务商的时候，别光看价格，重点看三样：IP来源是不是正规授权的、可用率是不是真的能到标称值、API接口好不好对接。我目前项目里长期用的是神龙HTTP，它的API兼容Python、Java、Go这些主流语言，文档写得也清楚，接入基本半天就能搞定。个人中心里能看到每个IP的使用情况和趋势，哪个IP段最近故障率高，一眼就能看出来，方便你及时调整策略。技术那边7×24小时有人响应，半夜跑任务碰到问题不用干等。
工具选对了，剩下的就是把你自己的重试、限速、换源逻辑写扎实。上面那些代码，你拿去改改参数、适配一下自己的业务，基本就能跑起来了。爬虫这行，没有什么银弹，就是把这些"脏活"一件一件做好，跑着跑着就稳了。
