为什么你的scrapy跑着跑着就"哑巴"了
做爬虫的人大概率都经历过这种场景：项目刚跑起来的时候风风火火，数据哗哗地进数据库，结果跑个十几分钟，日志里突然开始刷一堆 ConnectionError、Timeout，响应速度从2秒一条变成30秒一条，最后干脆卡死不动了。你打开代理IP列表一看，好家伙，一半的节点已经连不上了。
问题出在哪？说白了，代理IP是有寿命的。尤其是短效动态IP，可能3分钟、5分钟就过期了。你scrapy的代理池里塞了几百个IP，但其中有多少是"僵尸"，你根本不知道。等爬虫真正请求到那个IP的时候，才发现它已经废了，这时候再换下一个，时间就白白浪费了。
所以核心思路就一句话：在IP被scrapy真正使用之前，先把它"体检"一遍，不健康的直接踢掉。 下面我就把这套验证逻辑从头到尾讲清楚，代码可以直接拿去用。
验证代理IP到底验什么？别只盯着"通不通"
很多人写验证逻辑就一行代码：发个GET请求，看能不能收到200。这远远不够。一个"能用"的代理IP，至少得满足下面几个条件：
第一，连通性。 最基本的，TCP握手能不能完成，HTTP请求能不能发出去。这一步淘汰的是已经彻底失效的节点。
第二，响应速度。 有些IP虽然能通，但延迟高达8秒、10秒。你scrapy的timeout设的是5秒，那这个IP等于白给。一般超过3秒响应的节点，建议直接标记为慢节点，降低优先级或者剔除。
第三，出口IP是否一致。 有些代理服务商的节点存在"IP漂移"现象，你请求的时候显示的是A城市，实际出口跑到了B城市。如果你的业务对城市级定位有要求（比如做区域性的市场调研、本地化数据采集），这种IP就得过滤掉。
第四，协议兼容性。 你的scrapy走的是HTTP还是HTTPS？有些代理只支持HTTP，你拿它去请求HTTPS站点就会报SSL错误。验证的时候得用你实际要用的协议去测。
把这四步串起来，才算一次完整的"体检"。下面用表格对比一下不同验证粒度的效果：
| 验证层级 | 检测内容 | 耗时（单IP） | 能过滤的问题 |
|---|---|---|---|
| Level 1 - 连通性 | TCP连接 + HTTP状态码 | 约0.5~1s | 彻底失效的节点 |
| Level 2 - 速度 | 完整请求耗时 | 约1~3s | 高延迟慢节点 |
| Level 3 - 出口一致性 | 请求IP查询接口比对 | 约2~4s | IP漂移、城市不符 |
| Level 4 - 协议兼容 | HTTPS握手 + 证书校验 | 约1~2s | 协议不匹配 |
实际项目中，Level 1和Level 2是必做的，Level 3和Level 4根据你的业务需求选做。全部跑一遍大概每个IP需要3~5秒，如果你池子里有500个IP，用并发的方式跑完也就一两分钟的事。
完整代码：一个能直接跑的代理IP验证器
下面这段代码我拆成了几个部分，你可以根据需要取用。整体思路是：传入一组代理IP，并发地做连通性+速度检测，返回健康列表和失效列表。
import asyncio
import time
import aiohttp
from dataclasses import dataclass
from typing import List, Optional
@dataclass
class ProxyInfo:
"""单个代理IP的信息"""
host: str
port: int
protocol: str = "http" http / https
city: Optional[str] = None 期望的城市，用于Level 3校验
is_healthy: bool = False
latency: float = 0.0
fail_reason: str = ""
async def check_single_proxy(
session: aiohttp.ClientSession,
proxy: ProxyInfo,
timeout: float = 5.0,
speed_threshold: float = 3.0
) -> ProxyInfo:
"""
对单个代理做 Level 1 + Level 2 检测
"""
proxy_url = f"{proxy.protocol}://{proxy.host}:{proxy.port}"
构造一个轻量的测试请求
这里用一个公开的IP回显接口，同时完成连通性和出口IP检测
test_url = "http://httpbin.org/ip"
try:
start = time.time()
async with session.get(
test_url,
proxy=proxy_url,
timeout=aiohttp.ClientTimeout(total=timeout),
ssl=False
) as resp:
elapsed = time.time() - start
if resp.status != 200:
proxy.fail_reason = f"HTTP {resp.status}"
return proxy
Level 2: 速度检测
if elapsed > speed_threshold:
proxy.fail_reason = f"延迟过高: {elapsed:.2f}s"
return proxy
Level 3: 出口IP一致性（可选）
if proxy.city:
data = await resp.json()
real_ip = data.get("origin", "")
这里可以接入IP归属地查询，比对城市
简化处理：如果IP为空说明代理没生效
if not real_ip:
proxy.fail_reason = "代理未生效，出口IP为空"
return proxy
proxy.is_healthy = True
proxy.latency = elapsed
return proxy
except asyncio.TimeoutError:
proxy.fail_reason = "超时"
except aiohttp.ClientError as e:
proxy.fail_reason = f"连接异常: {type(e).__name__}"
except Exception as e:
proxy.fail_reason = f"未知错误: {str(e)[:80]}"
return proxy
async def validate_proxy_pool(
proxies: List[ProxyInfo],
concurrency: int = 50
) -> tuple:
"""
并发验证整个代理池
返回 (健康列表, 失效列表)
"""
semaphore = asyncio.Semaphore(concurrency)
healthy = []
failed = []
async def _check_with_limit(session, p):
async with semaphore:
result = await check_single_proxy(session, p)
if result.is_healthy:
healthy.append(result)
else:
failed.append(result)
async with aiohttp.ClientSession() as session:
tasks = [_check_with_limit(session, p) for p in proxies]
await asyncio.gather(tasks, return_exceptions=True)
return healthy, failed
---------- 使用示例 ----------
if __name__ == "__main__":
模拟一组代理IP（实际项目中从你的代理服务商API拉取）
raw_proxies = [
("120.234.56.78", 8080, "http", "北京"),
("223.104.12.34", 3128, "http", "上海"),
("116.25.89.101", 8888, "https", "广州"),
("183.232.45.67", 8080, "http", None),
... 更多IP
]
proxy_list = [
ProxyInfo(host=h, port=p, protocol=proto, city=c)
for h, p, proto, c in raw_proxies
]
healthy, failed = asyncio.run(validate_proxy_pool(proxy_list))
print(f"=== 验证完成 ===")
print(f"健康节点: {len(healthy)} 个")
print(f"失效节点: {len(failed)} 个")
print(f"--- 健康节点详情 ---")
for p in healthy:
print(f" {p.protocol}://{p.host}:{p.port} 延迟: {p.latency:.2f}s 城市: {p.city}")
print(f"--- 失效节点及原因 ---")
for p in failed:
print(f" {p.protocol}://{p.host}:{p.port} 原因: {p.fail_reason}")这段代码的几个关键点说一下：
并发数控制。 我用了 asyncio.Semaphore 把并发限制在50，避免你一下子对几百个IP同时发请求，把目标服务器打崩了，也避免你自己的网络带宽被占满。如果你的代理池特别大（比如上千个），可以把并发调到100~200，但别超过300，没意义。
超时设置。 ClientTimeout(total=timeout) 里的total是整体超时，包含了DNS解析、TCP连接、等待响应全过程。我默认设的5秒，如果你的代理线路质量一般，可以放宽到8秒，但别超过10秒，否则验证本身就成了瓶颈。
测试URL的选择。 代码里用的是 httpbin.org/ip，这个接口返回JSON，既轻量又能拿到出口IP。如果你在国内跑，访问海外接口可能本身就有延迟，可以换成国内的IP回显服务，或者干脆请求你自己业务目标站的一个轻量页面（比如首页的某个静态资源），这样验证结果更贴近真实场景。
把验证逻辑塞进scrapy中间件，让爬虫"自愈"
光有验证器还不够，你得让它跟scrapy跑起来的时候联动。思路是这样的：
启动前： 从代理服务商的API拉取一批IP，跑一遍上面的验证逻辑，把健康的IP灌进scrapy的代理池。
运行中： 每次请求失败（超时、连接拒绝），把对应的IP标记为"可疑"。连续失败3次就踢出池子，同时从服务商API补一个新的进来。
定时巡检： 每隔5分钟（对应你用的短效IP的有效期），对池子里的IP做一次快速连通性检测，把过期的提前清掉。
下面是scrapy中间件的完整实现：
import random
import time
import logging
from scrapy import signals
from scrapy.http import HttpError, TimeoutError as ScrapyTimeout
from collections import defaultdict
logger = logging.getLogger(__name__)
class Middleware:
"""
智能代理中间件：
- 自动剔除失效IP
- 失败重试时自动换IP
- 定时健康巡检
"""
def __init__(self, proxy_pool, max_retries=3, check_interval=300):
"""
proxy_pool: 你的代理池对象（下面会定义）
max_retries: 同一个IP连续失败几次后踢出
check_interval: 巡检间隔（秒），默认5分钟
"""
self.pool = proxy_pool
self.max_retries = max_retries
self.check_interval = check_interval
self.fail_count = defaultdict(int) ip -> 连续失败次数
self.last_check_time = 0
@classmethod
def from_crawler(cls, crawler):
从settings里读取配置
proxy_api_url = crawler.settings.get("PROXY_API_URL", "")
proxy_count = crawler.settings.get("PROXY_POOL_SIZE", 100)
max_retries = crawler.settings.get("PROXY_MAX_RETRIES", 3)
check_interval = crawler.settings.get("PROXY_CHECK_INTERVAL", 300)
pool = ProxyPoolManager(api_url=proxy_api_url, pool_size=proxy_count)
return cls(proxy_pool=pool, max_retries=max_retries, check_interval=check_interval)
def process_request(self, request, spider):
每次请求前，检查是否需要巡检
now = time.time()
if now - self.last_check_time > self.check_interval:
self.pool.run_health_check()
self.last_check_time = now
从健康池里随机取一个IP
proxy = self.pool.get_random_proxy()
if proxy:
request.meta["proxy"] = f"{proxy.protocol}://{proxy.host}:{proxy.port}"
request.meta["current_proxy"] = proxy
else:
logger.warning("代理池已空，等待补充...")
这里可以触发重新拉取
self.pool.refresh_from_api()
proxy = self.pool.get_random_proxy()
if proxy:
request.meta["proxy"] = f"{proxy.protocol}://{proxy.host}:{proxy.port}"
request.meta["current_proxy"] = proxy
def process_exception(self, request, exception, spider):
"""请求失败时，标记该IP并决定是否重试"""
proxy = request.meta.get("current_proxy")
if not proxy:
return None
proxy_key = f"{proxy.host}:{proxy.port}"
self.fail_count[proxy_key] += 1
if self.fail_count[proxy_key] >= self.max_retries:
连续失败达到阈值，踢出
logger.info(f"IP {proxy_key} 连续失败{self.max_retries}次，已剔除")
self.pool.remove_proxy(proxy)
self.fail_count.pop(proxy_key, None)
重试：换一个新IP
request.meta["proxy"] = None
request.meta["current_proxy"] = None
request.meta["retry_with_new_proxy"] = True
return request 返回request触发重试
return None
class ProxyPoolManager:
"""代理池管理器，对接神龙HTTP的API"""
def __init__(self, api_url: str, pool_size: int):
self.api_url = api_url
self.pool_size = pool_size
self.healthy_proxies: list = []
self._lock = None 多线程场景下加锁
def refresh_from_api(self):
"""
从神龙HTTP API拉取最新IP
实际调用时替换为你自己的API地址和鉴权参数
"""
import requests
try:
resp = requests.get(self.api_url, timeout=10)
resp.raise_for_status()
data = resp.json()
new_proxies = []
for item in data.get("data", []):
p = ProxyInfo(
host=item["ip"],
port=item["port"],
protocol=item.get("protocol", "http"),
city=item.get("city")
)
new_proxies.append(p)
跑一遍快速验证
import asyncio
healthy, _ = asyncio.run(
validate_proxy_pool(new_proxies, concurrency=50)
)
self.healthy_proxies = healthy[:self.pool_size]
logger.info(f"代理池已刷新，当前健康IP: {len(self.healthy_proxies)} 个")
except Exception as e:
logger.error(f"拉取代理IP失败: {e}")
def get_random_proxy(self):
if not self.healthy_proxies:
return None
return random.choice(self.healthy_proxies)
def remove_proxy(self, proxy: ProxyInfo):
self.healthy_proxies = [
p for p in self.healthy_proxies
if not (p.host == proxy.host and p.port == proxy.port)
]
def run_health_check(self):
"""定时巡检：快速检测当前池子里的IP是否还活着"""
if not self.healthy_proxies:
return
import asyncio
healthy, failed = asyncio.run(
validate_proxy_pool(self.healthy_proxies, concurrency=50)
)
removed = len(self.healthy_proxies) - len(healthy)
if removed > 0:
logger.info(f"巡检完成，剔除失效IP {removed} 个，剩余 {len(healthy)} 个")
self.healthy_proxies = healthy
如果健康IP低于池子的60%，触发补充
if len(self.healthy_proxies) < self.pool_size 0.6:
logger.info("健康IP不足，触发补充拉取")
self.refresh_from_api()
然后在你的 settings.py 里注册这个中间件：
settings.py
DOWNLOADER_MIDDLEWARES = {
"your_project.middlewares.Middleware": 543,
}
代理相关配置
PROXY_API_URL = "你的神龙HTTP API地址" 替换为实际地址
PROXY_POOL_SIZE = 100
PROXY_MAX_RETRIES = 3
PROXY_CHECK_INTERVAL = 300 5分钟巡检一次
配合scrapy自带的重试
RETRY_ENABLED = True
RETRY_TIMES = 2
RETRY_HTTP_CODES = [500, 502, 503, 504]这样整个流程就闭环了：启动时拉IP、验证、入池；运行中失败自动换IP；定时巡检清理过期节点。你不需要手动盯着日志看哪个IP挂了，系统自己会处理。
几个容易踩的坑，我替你踩过了
坑一：验证用的URL和业务目标URL不是同一个。 你验证的时候请求的是 httpbin.org，跑得好好的，结果scrapy实际去请求的目标站点把那个代理IP给ban了。这种情况建议验证的时候也请求一下你真正的目标站点（哪怕只是HEAD请求拿个状态码），确认代理在目标站是"清白"的。
坑二：短效IP的有效期比你验证时间还短。 你用的3分钟短效IP，结果验证一轮跑了4分钟，验证完IP已经过期了。解决办法：要么选5分钟或10分钟有效期的IP，要么把验证并发拉高，确保在有效期内跑完。神龙HTTP的短效动态IP池支持3/5/10/15/30分钟多种时长，你可以根据验证耗时来选，比如验证一轮需要4分钟，那就选10分钟的，留足余量。
坑三：HTTPS代理的证书问题。 有些代理IP走HTTPS的时候，中间人证书不是公共CA签发的，scrapy默认的 verify 会报错。如果你确认代理来源可靠，可以在scrapy里设置 DOWNLOAD_HANDLERS 关掉SSL验证，但生产环境不建议这么做，最好是让代理服务商提供正确的证书链。
坑四：代理池太小，一踢就空了。 你池子里就20个IP，踢掉3个就剩17个，再踢几个就快见底了。建议池子大小至少是你实际并发数的3倍以上。比如你scrapy同时跑30个请求，池子里至少放100个健康IP，这样即使有30%的IP在运行中失效，也不会影响整体吞吐。
关于代理IP资源，说两句实在话
上面代码写得再漂亮，底层IP质量不行也是白搭。我见过太多人用免费代理或者来路不明的IP列表，跑起来满屏报错，最后把时间全花在"换IP"上了，真正写业务逻辑的时间反而没多少。
如果你正在找稳定的代理IP资源，可以了解一下神龙HTTP。它是国内三大运营商正规授权的，IP资源池有3000万+，覆盖300多个城市级节点，IP纯度做到99.8%以上，可用率99.9%。支持HTTP/HTTPS/SOCKS5三种协议，API接口兼容主流爬虫语言，集成起来不费劲。
具体到套餐选择，如果你做的是日常数据采集、市场调研这类场景，对IP存活时间要求不算极端，短效动态IP池就够了，3到30分钟可选，每日更新去重，延迟低、连通率高，按量或按时计费，灵活度不错。如果你的业务需要同一IP持续在线（比如需要保持登录态、做长周期的数据跟踪），那长效静态IP池更合适，1到24小时可选，每日去重量10万+，可以指定省份或城市。对稳定性要求特别高、IP需求量又不大的，可以看看固定IP池，基于云主机构建，存活时间长，按个数售卖。
他们家还有个比较实用的功能：个人中心有可视化的数据统计面板，IP使用情况、使用趋势、异常告警都能直接看到，不用你自己再写一套监控。技术团队7×24小时在线，遇到集成问题随时能问到人。
常见问题
Q1：我的代理IP验证通过了，但scrapy实际请求还是超时，怎么回事？
大概率是验证的时候请求的目标和实际业务目标不一样。验证走的是轻量接口，响应快；但你的业务目标可能页面重、加载慢，或者对代理IP有额外的风控策略。建议把验证URL换成你实际要抓的站点，或者至少用同一个域名下的不同路径来测。另外检查一下scrapy的 DOWNLOAD_TIMEOUT 设置，默认18秒，如果你的代理延迟在3~5秒，加上页面本身的加载时间，可能刚好卡在边界上，适当放宽到25~30秒。
Q2：代理池里IP很多，每次启动都全量验证一遍太慢了，有没有更快的办法？
有。分两层处理：启动时做全量验证（一次性成本，可以接受），运行中只做增量巡检。巡检的时候不用重新走完整的Level 1~4，只做一个轻量的TCP连接检测（不实际发HTTP请求），100个IP并发跑完大概2~3秒。如果你用的是神龙HTTP这种有API的服务商，每次拉取的时候服务端已经做过一轮筛选了，你本地验证可以只关注速度和出口一致性，连通性那步可以省掉，整体耗时能砍掉一半。
Q3：scrapy的 process_exception 里返回request触发重试，会不会导致同一个URL被反复请求，把目标站打挂？
会的，如果你不加控制的话。两个建议：第一，在 process_exception 里加一个重试计数器（存在 request.meta 里），超过2次就不再重试，直接放弃这条请求；第二，scrapy自带的 RETRY_TIMES 和中间件的重试是叠加的，注意别把总重试次数设太高。如果你用的是神龙HTTP的代理，IP资源本身质量有保障，实际触发重试的频率会低很多，不像用杂牌代理那样动不动就失败。
Q4：我想让不同spider用不同地区的代理IP，怎么在中间件里实现？
在 process_request 里根据 request.meta 或者spider名称来筛选。比如你在spider里设置 self.custom_settings = {"PROXY_CITY": "杭州"}，中间件里读取这个字段，然后从池子里只取城市匹配的IP。神龙HTTP的IP资源支持指定省份、城市或混播，你在API拉取的时候就可以按城市过滤，中间件这边只需要做本地缓存和轮换就行，不用每次请求都去调API。
最后说点掏心窝的
代理IP验证这件事，说难不难，说简单也不简单。代码逻辑本身不复杂，真正花时间的是调参：超时设多少合适、并发开多大、巡检频率多高、失败阈值定几次——这些东西没有标准答案，得根据你用的IP质量、目标站点的响应速度、你自己的带宽条件去反复试。
我的建议是：先把上面那套代码跑通，用20~30个IP小范围试，观察日志里的失败分布，再逐步放大到几百个。别一上来就搞几千个IP的池子，出了问题你根本定位不了是哪个环节的问题。
IP质量是地基，代码逻辑是上层建筑。地基稳了，上层怎么搭都顺；地基不稳，你代码写得再花哨也白搭。选一个靠谱的IP来源，把验证和自愈机制做好，你的scrapy项目能少踩一大半的坑。
