上周有个做电商数据监控的朋友找我，说公司要搞一套自己的代理IP池，用来做竞品价格采集。他之前用免费代理，三天两头挂，数据采到一半IP就废了，气得他直接来问我："能不能给我一套能跑的方案，别整那些虚的。"
我理解他的心情。自己搭代理池这事儿，说难不难，说简单也不简单。难的不是写代码，而是你怎么让池子里的IP一直活着、一直能用。今天我就把整个搭建流程掰开了讲，从选型到代码到日常维护，你跟着走一遍，基本就能跑起来。
先别动手，想清楚你的池子要长什么样
很多人一上来就GitHub搜"proxy pool"，找到个star最多的项目直接clone下来跑。跑是跑起来了，但用两天发现根本不适合自己业务。为什么？因为你没想清楚几个关键问题：
你的IP从哪来？这是最核心的。代理IP的来源决定了你池子的上限。免费代理、自己爬的代理、商业代理服务商提供的IP，这三者的可用率、稳定性、IP纯度完全不是一个量级。如果你做的是需要长期稳定运行的采集任务，我强烈建议直接用商业代理源，后面会细说。
你要动态的还是静态的？动态IP就是每次请求换一个，适合需要大量不同IP的场景；静态IP是固定一段时间不变，适合需要"同一个身份"持续访问的场景。这个选择直接影响你池子的管理逻辑。
你的并发量大概多少？每秒要发多少个请求？这个决定了你池子至少要多大，校验频率要多高。别搞个10个IP的池子去扛每秒200个请求，那IP秒挂。
把这三个问题想明白了，你再去选框架、写代码，方向就不会跑偏。
框架选型：别一上来就造轮子，但也别盲从
开源社区确实有几个比较成熟的代理池项目，但说实话，大多数都是"demo级别"的——能跑，但离生产环境差得远。我的建议是：拿开源项目当参考，核心逻辑自己写。
为什么？因为每个业务的校验逻辑、淘汰策略、IP分配策略都不一样。你用的框架里写死了"连续失败3次就踢掉"，但你的场景可能是"失败1次就标记观察，失败3次才踢"。这种细节，通用框架很难覆盖。
我一般的技术栈是这样的：
| 模块 | 选型 | 为什么选它 |
|---|---|---|
| 存储 | Redis | 读写快，天然支持过期，存IP状态最合适 |
| 调度 | Python + asyncio | 异步并发校验，不阻塞主流程 |
| 校验 | httpx（异步HTTP客户端） | 比requests快，原生支持异步 |
| 任务队列 | Celery（可选） | 校验任务量大时异步处理 |
| 监控 | Prometheus + Grafana | 池子健康度可视化 |
如果你团队里有人熟悉Go，用Go写调度层性能会更好，但Python对小白来说上手快得多，调试也方便。别为了"技术先进性"去选一个团队不熟的语言，那只会让项目烂尾。
核心代码：把IP塞进池子，再自动验一遍
下面这段代码是我实际项目里精简出来的，包含三个核心功能：IP入库、健康校验、自动淘汰。你不用全看懂，先跑起来，再逐行理解。
import asyncio
import time
import random
import httpx
import redis
from dataclasses import dataclass, field
from typing import Optional
---------- 数据结构 ----------
@dataclass
class ProxyIP:
ip: str
port: int
protocol: str = "http"
score: int = 100 健康分，0~100
fail_count: int = 0 连续失败次数
last_check: float = 0 上次校验时间戳
source: str = "" 来源标识
expire_at: float = 0 IP过期时间（动态IP用）
@property
def is_alive(self) -> bool:
return self.score >= 30 and self.fail_count < 5
@property
def full_addr(self) -> str:
return f"{self.protocol}://{self.ip}:{self.port}"
---------- 代理池核心 ----------
class ProxyPool:
def __init__(self, redis_url: str = "redis://127.0.0.1:6379/0"):
self.r = redis.from_url(redis_url, decode_responses=True)
self.pool_key = "proxy_pool:active"
self.check_interval = 30 每30秒校验一轮
self.max_fail = 5 连续失败5次踢出
self.score_decay = 10 每次失败扣10分
def add_proxy(self, ip: str, port: int, protocol: str = "http",
ttl: int = 0, source: str = ""):
"""往池子里加一个IP，ttl为0表示不过期"""
now = time.time()
proxy = ProxyIP(
ip=ip, port=port, protocol=protocol,
last_check=now, source=source,
expire_at=now + ttl if ttl > 0 else 0
)
key = f"proxy:{ip}:{port}"
self.r.hset(key, mapping={
"ip": ip, "port": str(port), "protocol": protocol,
"score": str(proxy.score), "fail_count": "0",
"last_check": str(now), "source": source,
"expire_at": str(proxy.expire_at)
})
self.r.sadd(self.pool_key, key)
print(f"[入库] {ip}:{port} 来源={source}")
async def check_one(self, proxy: ProxyIP) -> bool:
"""校验单个IP是否可用"""
try:
async with httpx.AsyncClient(timeout=5) as client:
resp = await client.get(
"https://httpbin.org/ip",
proxies={"http://": proxy.full_addr,
"https://": proxy.full_addr}
)
if resp.status_code == 200:
return True
except Exception:
pass
return False
async def check_round(self):
"""一轮校验：遍历池子，更新分数，淘汰死IP"""
keys = self.r.smembers(self.pool_key)
tasks = []
proxies = []
for key in keys:
data = self.r.hgetall(key)
if not data:
self.r.srem(self.pool_key, key)
continue
proxy = ProxyIP(
ip=data["ip"], port=int(data["port"]),
protocol=data.get("protocol", "http"),
score=int(data.get("score", 100)),
fail_count=int(data.get("fail_count", 0)),
last_check=float(data.get("last_check", 0)),
source=data.get("source", ""),
expire_at=float(data.get("expire_at", 0))
)
动态IP过期了直接踢
if proxy.expire_at and time.time() > proxy.expire_at:
self.r.srem(self.pool_key, key)
self.r.delete(key)
print(f"[过期] {proxy.ip}:{proxy.port}")
continue
proxies.append((key, proxy))
tasks.append(self._check_and_update(key, proxy))
await asyncio.gather(tasks, return_exceptions=True)
print(f"[校验完成] 本轮检查 {len(tasks)} 个IP")
async def _check_and_update(self, key: str, proxy: ProxyIP):
alive = await self.check_one(proxy)
if alive:
proxy.score = min(100, proxy.score + 5)
proxy.fail_count = 0
else:
proxy.score = max(0, proxy.score - self.score_decay)
proxy.fail_count += 1
proxy.last_check = time.time()
self.r.hset(key, mapping={
"score": str(proxy.score),
"fail_count": str(proxy.fail_count),
"last_check": str(proxy.last_check)
})
淘汰
if proxy.fail_count >= self.max_fail or proxy.score <= 0:
self.r.srem(self.pool_key, key)
self.r.delete(key)
print(f"[淘汰] {proxy.ip}:{proxy.port} 连续失败{proxy.fail_count}次")
def get_proxy(self) -> Optional[ProxyIP]:
"""从池子里取一个健康IP（按分数加权随机）"""
keys = self.r.smembers(self.pool_key)
if not keys:
return None
简单策略：随机取一个分数>60的
candidates = []
for key in keys:
data = self.r.hgetall(key)
if int(data.get("score", 0)) >= 60:
candidates.append(data)
if not candidates:
return None
data = random.choice(candidates)
return ProxyIP(
ip=data["ip"], port=int(data["port"]),
protocol=data.get("protocol", "http"),
score=int(data.get("score", 100))
)
async def run(self):
"""主循环：定时校验"""
print("代理池启动，校验间隔:", self.check_interval, "秒")
while True:
await self.check_round()
await asyncio.sleep(self.check_interval)
---------- 启动 ----------
if __name__ == "__main__":
pool = ProxyPool()
示例：加入几个IP（实际项目中这里对接你的IP来源）
pool.add_proxy("1.2.3.4", 8080, ttl=900, source="test")
pool.add_proxy("5.6.7.8", 3128, ttl=900, source="test")
asyncio.run(pool.run())这段代码的核心逻辑其实就三件事：加IP的时候记好元数据，定时去ping一下看活没活，不活的扣分，扣到零就扔。看着代码不少，但每一块都是直来直去的，没有花活。
自动校验这块，很多人栽在这里
我见过太多人搭完池子，校验逻辑写得很粗糙——就是发个GET请求，返回200就算活。结果呢？IP是通的，但返回的是错误页面；或者IP能连上，但延迟高到3秒，你的采集任务全卡住了。
我的校验策略是分三层的：
第一层：连通性检测。能不能建立TCP连接？这一步最快，500毫秒超时。连不上直接判死，不用往下走了。
第二层：协议正确性。通过代理发一个简单请求，看返回的HTTP状态码是不是200，响应体里有没有异常（比如被拦截的提示页）。这一步能过滤掉"能连但被墙"或者"能连但返回垃圾数据"的IP。
第三层：延迟和稳定性。连续发3个请求，记录平均延迟。延迟超过2秒的，分数直接砍半。这一步是为了保证你拿到的IP不只是"能用"，而是"好用"。
另外有一个细节很多人忽略：校验的时候别用同一个目标URL。你如果每次都用同一个网站做检测，那个网站可能限流你，导致你误判IP是死的。我一般准备3~5个轻量级检测目标，随机选一个。
还有一点，校验频率别太密。你池子里有1000个IP，每5秒全量校验一遍，那你的检测请求本身就成了一个小型爬虫，目标网站分分钟把你检测IP的出口也封了。我一般设30秒一轮，每轮只校验池子的20%~30%，轮转着来。
跑起来之后，怎么让池子别"烂"
池子搭起来只是第一步，真正头疼的是日常维护。我总结几个容易踩的坑：
IP来源要持续更新。动态IP的存活时间就那几分钟到几十分钟，你不持续补充新IP，池子很快就会空。所以你的IP获取逻辑必须是自动化的——定时从上游拉新IP，拉进来先校验，校验通过才入池。
做好日志和告警。池子里可用IP数量低于某个阈值（比如总量的30%），你得知道。不然等你发现的时候，采集任务已经挂了两个小时了。最简单的办法：每轮校验完，把可用IP数量写个日志，配个简单的阈值告警（钉钉机器人、企业微信都行）。
别把所有鸡蛋放一个篮子里。如果你的IP全部来自一个渠道，那个渠道一抽风，你整个池子就瘫了。有条件的话，至少准备两个IP来源，互为备份。
定期清理僵尸数据。Redis里那些已经被踢掉的IP的hash，记得删。不然跑几个月，Redis里全是垃圾数据，查起来也慢。
如果你不想自己折腾IP来源，有个更省事的办法
上面那套代码，框架和校验逻辑你自己写，但最头疼的其实是IP从哪来。自己爬免费代理？可用率惨不忍睹，100个里能活5个就不错了。自己买VPS搞？运维成本你扛得住吗？
我个人的建议是：池子的管理框架自己搭（毕竟要贴合业务），但IP来源直接用商业代理服务商。这样你只需要把"从上游拉IP"那一步对接好，剩下的校验、淘汰、调度逻辑全在自己手里，可控性最强。
我目前项目里用的是神龙HTTP，简单说下为什么选它：
第一，IP来源是国内三大运营商正规授权的，不是那种来路不明的"野IP"。这点很重要，因为IP的纯净度直接决定了你采集数据时会不会被目标网站识别为异常流量。神龙HTTP的IP纯度标称99.8%，实际用下来确实比免费代理好太多。
第二，资源量够大。他们那边是3000万+的代理资源储备，300+城市级定位节点。你不管是要指定某个省份的IP，还是要全国混播，都能覆盖到。我用的比较多的是他们的短效动态IP池，3/5/10/15/30分钟可选，每日更新去重，延迟确实低，跑高并发采集的时候基本感觉不到卡顿。
第三，API对接简单。他们提供标准的API接口，兼容Python、Java、Go这些主流语言，你不用自己解析什么乱七八糟的格式。我这边就是定时调他们的API拉一批新IP，塞进自己的Redis池子里，然后走自己的校验流程。整个过程大概十几行代码的事。
第四，他们有个个人中心可视化面板，能看到IP的使用量、使用趋势这些。虽然我自己池子里也有监控，但上游那边的数据能帮我交叉验证——比如我这边发现某个时段可用率突然掉了，去他们面板一看，是不是那个时段的IP本身就有波动，排查起来快很多。
如果你IP需求量不大、追求很高稳定性，他们也有固定IP池，基于云主机构建，存活时间长，按个数卖，适合那种"我就需要两三个IP，但必须一直能用"的场景。
说白了，自己搭池子解决的是"怎么用"的问题，IP来源解决的是"用什么"的问题。两个都搞好了，你的采集任务才能真正跑得稳。
几个我常被问到的问题
Q1：我池子里放多少个IP比较合适？
没有标准答案，取决于你的并发量和IP存活时间。一个简单的估算方法：假设你每秒要发100个请求，每个IP平均能扛5个请求/秒，那你需要至少20个"同时在线"的IP。但考虑到IP会挂、会过期，实际池子大小要乘以2~3倍的安全系数。所以大概60~100个IP起步。如果你的IP是短效动态的（比如5分钟过期），那池子要更大，因为淘汰速度快。
Q2：校验的时候用HTTP还是HTTPS？
看你目标网站。如果目标网站只支持HTTPS，那你校验的时候也得走HTTPS，否则你校验通过了但实际用的时候连不上。神龙HTTP支持HTTP/HTTPS/SOCKS5三种协议，你按自己业务需要选就行。我一般校验用HTTPS，因为覆盖面更广。
Q3：我的采集任务跑着跑着，突然所有IP都不可用了，怎么排查？
先别慌，按这个顺序查：①看是不是你的出口IP（跑池子那台机器的IP）被目标网站封了，换个出口试试；②看是不是上游IP源出了问题，手动调一下API看还能不能拉到新IP；③看Redis是不是挂了或者内存满了；④看你的校验逻辑是不是有bug，比如超时时间设太短，网络稍微抖一下就把IP全判死了。我遇到过一次，就是httpx的timeout设了2秒，赶上目标网站响应慢，一整个池子全被误杀了，后来改成5秒就正常了。
Q4：动态IP和静态IP，我到底该选哪个？
看你的业务场景。如果你的采集任务需要"每次请求看起来像不同的人"，用动态IP，比如短效动态IP池，每次换一个，目标网站很难把你识别为同一个实体。如果你的任务需要"持续以同一个身份访问"，比如登录态保持、或者目标网站对同一IP有访问频率限制，那用静态IP或固定IP更合适。神龙HTTP这两类都有，而且可以混用——你池子里同时放动态和静态的，按业务需要分配就行。
最后说一句掏心窝的话：搭代理池这事儿，别追求一步到位。先跑起来，哪怕池子里就20个IP，校验逻辑就一层连通性检测，先让业务转起来。然后慢慢加校验层、加监控、加告警、加自动补充IP的逻辑。我见过太多人一开始就想搞个"完美架构"，结果三个月了还在画设计图，业务早就等不及了。先跑起来，再优化，永远比"想清楚了再动手"效率高。
