先说个扎心的事实:你的IP撑不过三轮请求
做数据采集这行干久了,你肯定遇到过这种场景:脚本刚跑起来,前几个请求还正常,到第十几个请求开始返回403,再往后干脆连页面都加载不出来。你盯着终端看了半天,最后发现——不是你的代码写错了,是你的IP被目标站点"标记"了。
说白了,同一个出口IP在短时间内反复请求同一个站点,对方的风控系统会把你判定为异常流量,轻则限流,重则直接封掉你当前这个IP。这时候你手动改IP重新跑?那效率也太低了。
所以代理IP池这个东西,不是"锦上添花",而是你采集链路能不能稳定跑下去的基础设施。这篇文章我就从最基础的环境搭建讲起,一步步用Python把一套可用的代理IP池管理模块搭出来,最后接进你的采集脚本里,整条链路跑通。不整那些虚的,跟着敲代码就行。
动手之前,先把环境捋清楚
别急着写代码,先把该装的东西装好,省得到时候跑到一半报个ImportError,心态直接崩。
你需要准备的东西不多:
Python 3.10及以上版本(3.8也能跑,但有些库的新特性用不上),requests 库(发HTTP请求用的),httpx(如果你后续想上异步采集,这个比requests方便),还有一个线程池相关的标准库 concurrent.futures,这个Python自带不用额外装。
你需要提前在神龙HTTP的后台开通一个API调用权限。神龙HTTP是国内三大运营商正规授权的代理IP服务商,资源池里有3000万+的IP储备,每个IP都经过筛选验证,可用率标称99.9%。你开通之后会拿到一个API Key,后面拉取IP全靠它。支持HTTP、HTTPS、SOCKS5三种协议,覆盖300多个城市级节点,延迟和并发表现都比较稳。
装依赖就一行命令的事:
pip install requests httpx
装完验证一下,终端里敲 python -c "import requests; print(requests.__version__)",能打印出版本号就说明没问题。
核心步骤:通过API从神龙HTTP拉取代理IP
这是整篇文章最关键的一步。你不需要自己去"找"IP,神龙HTTP提供了标准的API接口,你传参数、它返回IP列表,就这么简单。
先搞清楚几个概念,别被术语绕晕:
| 套餐类型 | IP存活时长 | 适合场景 |
|---|---|---|
| 短效动态IP池 | 3/5/10/15/30分钟(可定制) | 高频采集、需要频繁更换出口IP的场景 |
| 长效静态IP池 | 1/4/8/12/24小时(可定制) | 需要一段时间内保持同一IP的场景 |
| 固定IP池 | 长期有效 | 对稳定性要求很高、IP需求量不大的场景 |
如果你做的是常规的数据采集,比如抓取公开的市场信息、行业报告页面,短效动态IP池是最常用的选择。3000万+资源每日更新去重,延迟低,按量或者按时计费,个人和企业都能用得上。
下面这段代码就是调用神龙HTTP API获取代理IP的完整逻辑,我写得比较"土",但每一步都注释了,你直接能跑:
import requests
import time
import random
class ShenlongProxyPool:
"""神龙HTTP代理IP池管理器"""
def __init__(self, api_key: str, protocol: str = "http"):
self.api_key = api_key
self.protocol = protocol
self._proxy_cache = [] 本地缓存的可用IP列表
self._cache_expire = 0 缓存过期时间戳
self._cache_ttl = 180 缓存有效期(秒),短效IP建议设短一点
def _fetch_from_api(self, count: int = 10, city: str = None) -> list:
"""
从神龙HTTP API拉取代理IP
count: 需要获取的IP数量
city: 指定城市,不传则随机分配
"""
url = "https://h.shenlongip.com/api/get"
params = {
"key": self.api_key,
"num": count,
"protocol": self.protocol,
"format": "text"
}
if city:
params["city"] = city
try:
resp = requests.get(url, params=params, timeout=10)
resp.raise_for_status()
返回格式:每行一个 ip:port
proxies = [line.strip() for line in resp.text.strip().splitlines() if line.strip()]
return proxies
except requests.RequestException as e:
print(f"[警告] 拉取代理IP失败: {e}")
return []
def get_proxy(self) -> str:
"""
获取一个可用的代理IP(带本地缓存)
返回格式: "http://ip:port"
"""
now = time.time()
缓存过期或为空,重新拉取
if now > self._cache_expire or len(self._proxy_cache) == 0:
self._proxy_cache = self._fetch_from_api(count=20)
self._cache_expire = now + self._cache_ttl
if not self._proxy_cache:
return None 没拿到IP,调用方需要处理
随机取一个,避免总用同一个
idx = random.randint(0, len(self._proxy_cache) - 1)
ip_port = self._proxy_cache.pop(idx)
return f"{self.protocol}://{ip_port}"
def remove_proxy(self, proxy: str):
"""
把用坏的IP从缓存里踢掉
"""
if proxy and proxy in self._proxy_cache:
self._proxy_cache.remove(proxy)
print(f"[清理] 移除失效IP: {proxy}")
这段代码的设计思路其实就三件事:拉取、缓存、淘汰。你不可能每次请求都去调API拿IP,那太浪费配额了,所以本地缓存一批;但短效IP存活时间就几分钟,缓存不能设太长;某个IP用着突然不通了,得赶紧从池子里踢掉,别让后面的请求继续用它。
把代理池真正接进你的采集脚本
上面那个类写好了,但光有IP没用,你得让它真正参与到HTTP请求里去。下面是一个完整的采集示例,我拿一个公开的天气数据页面举例(你换成自己实际要抓的站点就行):
import requests
from ShenlongProxyPool import ShenlongProxyPool 你保存上面的类到同目录
def fetch_with_proxy(pool: ShenlongProxyPool, target_url: str, max_retries: int = 3) -> str:
"""
带代理的页面抓取,失败自动换IP重试
"""
for attempt in range(max_retries):
proxy = pool.get_proxy()
if not proxy:
print("[错误] 没有可用代理IP,等待5秒后重试...")
time.sleep(5)
continue
proxies = {
"http": proxy,
"https": proxy
}
try:
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/125.0.0.0 Safari/537.36"
}
resp = requests.get(target_url, proxies=proxies, headers=headers, timeout=15)
if resp.status_code == 200:
return resp.text
elif resp.status_code in (403, 429):
被风控了,这个IP废了,踢掉换下一个
print(f"[重试 {attempt+1}] IP被限流({resp.status_code}),更换IP...")
pool.remove_proxy(proxy)
time.sleep(random.uniform(1, 3))
else:
print(f"[重试 {attempt+1}] 异常状态码: {resp.status_code}")
pool.remove_proxy(proxy)
time.sleep(2)
except (requests.ConnectionError, requests.Timeout) as e:
print(f"[重试 {attempt+1}] 连接异常: {e}")
pool.remove_proxy(proxy)
time.sleep(2)
raise Exception(f"连续{max_retries}次请求失败,目标: {target_url}")
===== 实际使用 =====
if __name__ == "__main__":
替换成你自己的API Key
pool = ShenlongProxyPool(api_key="你的API_KEY", protocol="http")
target = "https://www.example-weather.com/data/city/110000"
html = fetch_with_proxy(pool, target)
print(f"抓取成功,页面长度: {len(html)} 字符")
注意几个细节:
第一,重试机制。你不可能指望一个IP每次都好用,尤其是短效IP,可能你刚拿到手它就已经过期了。所以max_retries设个3次,每次失败就换IP重来,这个容错逻辑必须有。
第二,403和429要区别对待。403是明确告诉你"不让你访问",429是"你太频繁了,歇会儿"。两种情况都说明当前IP不可用了,得踢掉。但如果是502、503这种服务端错误,不一定是IP的问题,可能是目标站点本身在维护,这时候换IP也没用,应该等一会儿再试。
第三,请求间隔。就算你换了IP,也别像机关枪一样连续打。代码里那个 time.sleep(random.uniform(1, 3)) 不是随便写的,加个随机数比固定间隔自然得多,固定间隔反而容易被识别为机器行为。
并发场景下怎么管理代理池
如果你不是单线程慢慢跑,而是用多线程同时抓几十个页面,代理池的管理就要注意线程安全了。上面那个类在单线程下没问题,但多线程同时调 get_proxy() 可能会出现竞态条件——两个线程同时pop同一个IP。
最简单的解决办法是加个锁:
import threading
class ThreadSafeProxyPool(ShenlongProxyPool):
"""线程安全版本"""
def __init__(self, args, kwargs):
super().__init__(args, kwargs)
self._lock = threading.Lock()
def get_proxy(self) -> str:
with self._lock:
return super().get_proxy()
def remove_proxy(self, proxy: str):
with self._lock:
super().remove_proxy(proxy)
然后采集的时候用线程池:
from concurrent.futures import ThreadPoolExecutor, as_completed
def worker(pool, url):
try:
return fetch_with_proxy(pool, url)
except Exception as e:
return f"FAILED: {e}"
假设你有一组要抓的URL
url_list = [f"https://www.example-weather.com/data/city/{i}" for i in range(110000, 110050)]
pool = ThreadSafeProxyPool(api_key="你的API_KEY")
with ThreadPoolExecutor(max_workers=5) as executor:
futures = {executor.submit(worker, pool, url): url for url in url_list}
for future in as_completed(futures):
url = futures[future]
result = future.result()
status = "OK" if not result.startswith("FAILED") else "FAIL"
print(f"[{status}] {url}")
这里 max_workers=5 我设的是5个并发线程。别一上来就开50个、100个,你的代理IP池再大也扛不住那么高的瞬时并发,而且目标站点的压力也会骤增,反而更容易触发风控。5到10个并发线程对大多数场景来说够用了,具体数字你根据自己拿到的IP数量和目标站点的承受能力来调。
几个我踩过的坑,你大概率也会遇到
坑一:IP拿到了但连不上。 这种情况在短效IP里比较常见,可能IP刚分配给你,但对应的网络节点还没完全就绪。解决办法就是在 fetch_with_proxy 里把连接超时设短一点(比如5秒),连不上就快速失败、换下一个IP,别傻等30秒。
坑二:HTTPS站点用HTTP代理报错。 如果你的目标站点是HTTPS的,代理协议也要对应。神龙HTTP支持HTTP/HTTPS/SOCKS5三种协议,你在初始化 ShenlongProxyPool 的时候把 protocol 参数设成 "https" 就行,别混着用。
坑三:缓存时间设得太长。 我一开始图省事把 _cache_ttl 设成了3600秒(1小时),结果跑着跑着一大半IP都过期了,成功率从95%掉到60%。短效IP的存活时间就3到30分钟,你的缓存有效期不能超过IP最短存活时间,设120到180秒比较稳妥。
坑四:只盯着成功率,不看延迟。 有些IP虽然能连通,但延迟高达2-3秒,你5个并发线程全卡在这种慢IP上,整体吞吐量反而比单线程还低。神龙HTTP的IP资源整体延迟控制得不错,但如果你发现某些城市节点的响应明显偏慢,可以在API调用时指定其他城市,或者在代码里加一个延迟阈值,超过800ms的IP直接标记为低优先级。
常见问题
Q1:我一天大概要跑2000个请求,用哪种套餐比较合适?
2000个请求的量不算大,短效动态IP池完全够用。按量计费的话成本很低,而且你不需要一直挂着连接,跑完就停。如果你希望同一批请求(比如抓同一个站点的50个页面)尽量走同一个IP,那可以看看长效静态IP池,1小时或4小时的存活时间能覆盖你一轮采集。具体选哪个,你根据自己采集的"轮次"来定——一轮采集大概多长时间,就选对应时长的IP。
Q2:我的脚本跑着跑着突然全部403了,是IP池的问题还是目标站点的问题?
先别急着怀疑IP池。你拿一个刚拉出来的IP,手动用浏览器或者curl访问一下目标站点,如果手动也403,那大概率是目标站点那边做了IP段级别的封禁,或者你之前的请求频率触发了更高级别的风控。这时候换IP也没用,得等一段时间(通常几小时到一天)再试。如果手动访问正常、只有脚本403,那检查你的请求头、请求频率、以及是不是同一个IP短时间内请求次数太多了。
Q3:神龙HTTP的API接口兼容我用的Python框架吗?
兼容。神龙HTTP的API是标准的HTTP GET接口,返回纯文本格式的IP列表,你用requests、httpx、aiohttp、Scrapy、或者任何能发HTTP请求的库都能调。他们官方也提供了Python的示例代码和接口文档,你拿到API Key之后对着文档调就行,不需要装什么特殊的SDK。技术团队是7×24小时在线的,集成过程中遇到任何接口层面的问题都可以直接问。
Q4:我同时跑两个采集任务,共用一个代理池会不会互相影响?
会。两个任务共用同一个 ShenlongProxyPool 实例的话,A任务刚拿走的IP可能B任务也想用,缓存里的IP会被快速消耗完,导致频繁调API重新拉取。建议每个采集任务独立创建一个池实例,各自维护自己的缓存。如果两个任务的目标站点不同,还可以给不同任务指定不同的城市节点,减少IP重叠的概率。神龙HTTP的API支持按城市筛选,你在初始化时传不同的 city 参数就行。
最后说两句
代理IP池这个东西,代码层面其实不复杂,核心就是"拉取-缓存-淘汰-重试"四步循环。真正影响你采集稳定性的,是IP本身的质量。你代码写得再漂亮,IP池里全是三天没更新过的僵尸IP,那成功率照样拉胯。
这也是为什么我推荐用神龙HTTP这种正规渠道的IP资源——国内三大运营商授权、3000万+储备每日去重更新、99.8%以上的IP纯度,这些不是你自己在家用脚本"攒"IP能攒出来的。你省下来的调试时间,够你多跑好几轮采集了。
把上面那些代码拷下来,API Key换成你自己的,改一下目标URL,跑起来。跑通了,你的采集链路就算真正立住了。


