做数据采集这行，谁没被代理IP坑过？我见过太多人，辛辛苦苦写了一整套采集逻辑，结果跑起来一半请求超时、一半返回403，回头一查——IP池里混了一堆"僵尸"和"脏"地址。更离谱的是，有些IP明明显示在线，实际一请求就给你来一个连接重置，整个任务卡在那儿干瞪眼。
所以今天不聊什么高大上的架构设计，就聊一个最实际的问题：怎么把坏IP挡在门外，让你的采集任务跑得干净利索。我总结了三个真正管用的办法，都是踩了无数坑之后才摸出来的经验，你照着做，至少能少掉一半的头发。
第一招：拿到IP先"体检"，别上来就开干
很多人拿到代理IP列表，第一反应就是往请求里塞，跑起来再说。这是最要命的习惯。你想想，一个IP可能在你拿到它的那几秒里就已经失效了，或者它本身就是一个被目标站点标记过的"老油条"。你不测就直接用，等于闭着眼睛往枪口上撞。
正确的做法是：在正式发起业务请求之前，先对每个IP做一轮轻量级的连通性预检。具体测三样东西——TCP三次握手能不能完成、HTTP请求能不能拿到正常响应码、往返延迟在不在可接受范围内。这三关都过了，这个IP才算"合格"，才配进入你的工作队列。
下面这段Python代码，是我实际项目里用的预检逻辑，稍微改改就能直接跑：
import socket
import requests
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
def check_ip(ip, port, timeout=3):
"""对单个代理IP做连通性体检"""
第一关：TCP连接
try:
sock = socket.create_connection((ip, port), timeout=timeout)
sock.close()
except (socket.timeout, ConnectionRefusedError, OSError):
return {"ip": ip, "status": "tcp_fail"}
第二关：HTTP响应
try:
proxies = {"http": f"http://{ip}:{port}", "https": f"http://{ip}:{port}"}
start = time.time()
resp = requests.get("http://httpbin.org/ip", proxies=proxies, timeout=timeout)
latency = round((time.time() - start) 1000)
if resp.status_code == 200:
return {"ip": ip, "status": "ok", "latency_ms": latency}
else:
return {"ip": ip, "status": f"http_{resp.status_code}"}
except requests.exceptions.RequestException:
return {"ip": ip, "status": "http_fail"}
def batch_precheck(ip_list, max_workers=20):
"""并发体检一批IP，只保留合格的"""
healthy = []
with ThreadPoolExecutor(max_workers=max_workers) as pool:
futures = {pool.submit(check_ip, ip, 8080): ip for ip in ip_list}
for fut in as_completed(futures):
result = fut.result()
if result["status"] == "ok" and result.get("latency_ms", 999) < 800:
healthy.append(result)
return healthy
用法：从IP池拉一批，先体检再干活
raw_ips = ["1.2.3.4", "5.6.7.8", "9.10.11.12"] 实际从服务商API获取
good_ips = batch_precheck(raw_ips)
print(f"体检通过 {len(good_ips)}/{len(raw_ips)} 个")这里有个细节值得注意：延迟阈值别设太死。我一般把上限卡在800毫秒，超过这个数的IP虽然"活着"，但在高并发场景下会拖慢整体节奏，不如直接淘汰。预检的并发数要根据你机器带宽来调，别一上来就开200个线程，网络先崩了。
第二招：跑起来之后，用"反馈淘汰"机制实时清场
预检只能解决"入场"问题。但IP这东西是活的，它可能在你用的第三分钟突然被目标站点拉黑，也可能因为运营商侧的线路调整突然变得极慢。所以光靠入场体检远远不够，你必须在运行过程中持续监控每个IP的表现，一旦"变坏"就立刻踢出去。
我一般用一套简单的滑动窗口计数：每个IP维护一个最近N次请求的成功/失败记录。连续失败达到阈值（比如连续3次超时或返回非200），就把它扔进黑名单，同时从工作队列里摘除。黑名单不是永久的，我一般设一个冷却期（比如10分钟），到期后重新放回候选池再测一次。
核心逻辑不长，但能省你大量排查时间：
from collections import defaultdict
import time
class IPHealthTracker:
"""滑动窗口式IP健康追踪器"""
def __init__(self, window=5, fail_threshold=3, cooldown=600):
self.window = window 滑动窗口大小
self.fail_threshold = fail_threshold 连续失败阈值
self.cooldown = cooldown 黑名单冷却时间（秒）
self.history = defaultdict(list) ip -> [(timestamp, success)]
self.blacklist = {} ip -> 拉黑时间戳
def record(self, ip, success):
now = time.time()
self.history[ip].append((now, success))
只保留最近window条
if len(self.history[ip]) > self.window:
self.history[ip] = self.history[ip][-self.window:]
检查是否连续失败达到阈值
recent = self.history[ip][-self.fail_threshold:]
if len(recent) == self.fail_threshold and not any(s for _, s in recent):
self.blacklist[ip] = now
print(f"[淘汰] {ip} 连续{self.fail_threshold}次失败，拉黑{self.cooldown}秒")
def is_blacklisted(self, ip):
if ip not in self.blacklist:
return False
if time.time() - self.blacklist[ip] > self.cooldown:
del self.blacklist[ip] 冷却结束，放行
return False
return True
def get_healthy_ips(self, candidate_ips):
return [ip for ip in candidate_ips if not self.is_blacklisted(ip)]
实际使用
tracker = IPHealthTracker(window=5, fail_threshold=3, cooldown=600)
每次请求结束后调用
tracker.record("1.2.3.4", success=True/False)
取可用IP时调用
available = tracker.get_healthy_ips(all_ips)这套机制的关键在于"连续"两个字。偶尔一次失败可能是网络抖动，不能一棍子打死。但连续三次都失败，大概率是IP本身出了问题，这时候果断淘汰比硬撑着等它恢复要高效得多。
第三招：从源头掐断——选对IP服务商比什么都重要
前面两招都是"事后补救"，但说实话，如果你拿到的IP池本身质量就拉胯，你花再多精力过滤也是杯水车薪。我见过有人用那种来路不明的免费IP池，十个里面七八个是死的，过滤完能用的就剩两三个，还全是高延迟的"老爷车"。这种时候你写再精妙的过滤算法也没用，源头不干净，下游怎么洗都是脏的。
选服务商的时候，我一般重点看三件事：
第一，资源授权是否正规。IP不是凭空变出来的，它得从运营商那里合法获取。正规的服务商会明确告诉你资源来自哪些运营商、是否有授权。那些说不清来源、价格低得离谱的，大概率是爬来的或者盗用的，用着随时可能断，甚至给你惹上麻烦。
第二，IP更新频率和去重机制。动态IP池如果三天不更新，里面肯定堆了一堆过期地址。靠谱的服务商应该做到每日去重更新，而且更新量要够大，不然你拉到的可能还是"老面孔"。
第三，有没有可视化的使用监控。你总不能天天手动去查哪些IP挂了。好的平台会给你提供使用统计、成功率趋势、异常告警这些功能，让你一眼就能看出哪个时间段的IP质量出了问题，方便及时调整。
我自己现在用的是神龙HTTP，简单说下为什么留下来了。它是国内三大运营商正规授权的，资源池有3000万+，每天更新去重，IP纯度标称99.8%，实际跑下来确实很少遇到"一用就死"的情况。而且它支持HTTP/HTTPS/SOCKS5三种协议，API接口文档写得挺清楚，我接进现有采集框架大概半天就搞定了。另外它个人中心那个可视化面板挺实用，能直接看到每个IP的使用次数、成功率、延迟分布，不用自己再搭一套监控。
如果你刚起步、用量不大，它的短效动态IP池比较合适，3到30分钟不等的有效期，按量或者按时计费，灵活度很高，300多个城市级节点可以指定地区。如果你需要某个IP长时间稳定在线（比如做需要保持会话一致性的采集任务），可以看看它的固定IP池，基于云主机构建，存活时间长，连通率很稳，按个数买、按时间计费，适合用量不大但要求稳定的场景。
三招配合起来，效果到底怎么样？
单独用任何一招都不够。我现在的实际工作流是这样的：从神龙HTTP的API拉一批IP → 跑第一招的并发预检，把明显不行的先筛掉 → 进入正式采集，每个请求走第二招的健康追踪 → 后台定时（比如每5分钟）把黑名单里冷却到期的IP重新拉回候选池做预检。三招形成闭环，坏IP基本活不过一个请求周期。
下面这张表是我跑了一个月数据后的粗略统计，供参考：
| 指标 | 未做过滤（裸跑） | 三招全上 |
|---|---|---|
| 请求成功率 | 约72% | 约97.5% |
| 平均响应延迟 | 1240ms | 380ms |
| 任务中断次数/天 | 6-8次 | 0-1次 |
| 需要人工介入排查的频率 | 每天至少1次 | 基本不用管 |
数据不会骗人。过滤这件事看着不起眼，但它直接决定了你的采集任务是"跑着跑着就顺了"还是"三天两头要停下来修"。
常见问题
Q1：预检的时候用httpbin还是直接请求目标站点？
建议用通用的连通性检测页面（比如httpbin或者你服务商提供的检测接口），不要直接打目标站点。原因有两个：一是预检频率高，直接打目标站容易触发对方的频率限制，反而把你的IP搞进对方的黑名单；二是预检的目的是验证"这个IP能不能正常发HTTP请求"，不需要验证"目标站认不认这个IP"，后者是业务层面的事，放到正式请求里处理就行。
Q2：短效IP和固定IP到底怎么选？会不会短效的不够用？
取决于你的任务特征。如果你的采集任务对IP的"身份连续性"没有要求——比如每次请求都是独立的、不需要保持同一个IP的会话——那短效动态IP完全够用，而且因为IP一直在轮换，被目标站点标记的概率也低。但如果你需要在一个较长的时间窗口内保持同一个出口IP（比如某些需要登录态维持的公开数据接口），那就得上固定IP或者长效静态IP。神龙HTTP的短效池有效期从3分钟到30分钟可选，固定IP则按个数长期持有，你根据任务时长来匹配就行。
Q3：IP被拉进黑名单后，冷却期设多长比较合理？
没有标准答案，跟你的业务节奏有关。我的经验是：短效IP池的冷却期设5-10分钟就够了，因为短效IP本身寿命就短，等它"恢复"不如直接换一个新的。固定IP的冷却期可以长一些，15-30分钟，因为固定IP数量有限，频繁拉黑又恢复会浪费资源。如果某个IP在24小时内被拉黑超过3次，建议直接永久移除，别给它"复活"的机会了。
Q4：高并发场景下，预检会不会成为瓶颈？
会，如果你处理不当的话。两个建议：第一，预检的并发线程数不要超过你机器网卡带宽能承受的极限，一般20-50个并发比较稳，再高容易把本地网络打满；第二，预检结果做缓存，同一个IP在有效期内（比如5分钟内）不需要反复测。神龙HTTP的API本身支持高并发提取，你拉IP的速度不会慢，瓶颈通常在你自己的预检逻辑上，把这里优化好，整体吞吐就上去了。
