代理 IP 池代码可以使用 Python 完成完整开发，核心是实现 IP 拉取、有效性校验、存储、API 调度、自动补池这五大模块，整体采用 Flask 作为对外接口服务，Redis 做 IP 缓存存储，配合定时任务做 IP 健康检测，整套代码逻辑清晰、可直接二次扩展，适合学习自建代理 IP 池的开发流程，代码会把 IP 存活状态、延迟、协议类型都做标记，调用接口的时候自动筛选可用 IP，并且支持随机轮询、粘性会话两种调度模式，后续还能继续扩展地区筛选、权重调度等高级功能。之所以选择这套技术栈，是因为 Python 开发效率高，爬虫和网络请求生态完善，Redis 读写速度快，非常适合高频读取代理 IP 的业务场景，Flask 轻量接口框架部署简单，不需要复杂的 Web 容器，定时任务可以用 APScheduler 实现，自动周期性校验 IP，清理失效地址，当池内可用 IP 数量低于预设阈值时自动补充新 IP，保证代理 IP 池持续有可用资源。
开发前准备清单
表格
| 项目 | 说明 |
|---|---|
| 编程语言 | Python 3.8+ |
| Web 框架 | Flask |
| 缓存数据库 | Redis |
| 定时任务库 | APScheduler |
| 网络请求库 | requests |
| 可选 | pymysql，如果改用 MySQL 持久化存储 |
完整代码实现
1、依赖安装
pip install flask redis apscheduler requests
2、代理 IP 池完整代码 proxy_pool.py
import requests
import redis
import time
from flask import Flask, jsonify, request
from apscheduler.schedulers.background import BackgroundScheduler
app = Flask(__name__)
连接Redis
r = redis.Redis(host='127.0.0.1', port=6379, db=0, decode_responses=True)
POOL_KEY = "proxy_ip_pool"
MIN_AVAILABLE_NUM = 20 最低可用IP数量阈值
TEST_URL = "https://httpbin.org/ip"
TIMEOUT = 8
模拟IP源接口，实际项目替换成正规服务商API
def fetch_new_proxy():
proxy_list = [
{"ip": "1.1.1.1", "port": "8080", "type": "http"},
{"ip": "2.2.2.2", "port": "1080", "type": "socks5"}
]
return proxy_list
校验代理IP是否可用
def check_proxy(proxy):
proxy_url = f"{proxy['type']}://{proxy['ip']}:{proxy['port']}"
proxies = {
"http": proxy_url,
"https": proxy_url
}
try:
start = time.time()
resp = requests.get(TEST_URL, proxies=proxies, timeout=TIMEOUT)
cost = round(time.time() - start, 2)
if resp.status_code == 200:
return True, cost
except Exception:
pass
return False, None
把新IP写入Redis代理IP池
def add_proxy_to_pool():
proxies = fetch_new_proxy()
for p in proxies:
ok, delay = check_proxy(p)
if ok:
data = f"{p['ip']}:{p['port']}:{p['type']}:{delay}"
r.sadd(POOL_KEY, data)
定时校验全部IP，剔除失效IP
def scheduled_check():
all_proxy = r.smembers(POOL_KEY)
for item in all_proxy:
ip, port, ptype, delay = item.split(":")
proxy = {"ip": ip, "port": port, "type": ptype}
ok, new_delay = check_proxy(proxy)
if not ok:
r.srem(POOL_KEY, item)
数量不足自动补池
if r.scard(POOL_KEY) < MIN_AVAILABLE_NUM:
add_proxy_to_pool()
获取随机可用代理IP接口
@app.route("/get_proxy", methods=["GET"])
def get_proxy():
count = r.scard(POOL_KEY)
if count == 0:
return jsonify({"code": 404, "msg": "代理IP池暂无可用IP"})
proxy_item = r.srandmember(POOL_KEY)
ip, port, ptype, delay = proxy_item.split(":")
return jsonify({
"code": 200,
"data": {
"ip": ip,
"port": port,
"type": ptype,
"delay": delay
}
})
启动定时任务
scheduler = BackgroundScheduler()
scheduler.add_job(scheduled_check, 'interval', minutes=5)
scheduler.start()
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000, debug=False)
代码运行步骤
- 启动本地 Redis 服务，确认端口 6379 正常监听
- 安装全部依赖包
- 修改代码里 fetch_new_proxy 函数，替换成正规代理服务商的 API 拉取逻辑
- 根据业务调整 TIMEOUT 超时时间、MIN_AVAILABLE_NUM 最低 IP 数量、检测周期
- 运行 python proxy_pool.py
- 浏览器或者业务代码请求 http://127.0.0.1:5000/get_proxy 即可获取可用代理 IP
代码扩展升级清单
表格
| 扩展功能 | 开发要点 |
|---|---|
| 粘性会话 | 增加 session_id 参数，Redis 单独保存会话和 IP 绑定关系，同一 session 持续使用同一个 IP |
| 权重调度 | 给每条 IP 增加 weight 权重字段，随机抽取时按照权重概率分配 |
| 地区筛选 | IP 数据增加 country、province 字段，接口增加地区筛选参数 |
| MySQL 持久化 | Redis 做缓存，MySQL 保存全部 IP 历史数据，支持报表统计 |
| 失败重试 | 业务端拿到 IP 请求失败之后，调用代码接口重新获取下一个 IP |
| 多协议混合池 | 同时支持 http https socks5，校验函数增加 socks5 对应的请求适配 |
这套代理 IP 池代码完成基础闭环，实现自动拉取、校验、存储、接口分发、定时巡检补池，业务端只需要调用 /get_proxy 接口就能拿到出口 IP，后续可以按照业务需求持续迭代增加高级调度能力。
【相关 FAQ】
问题 1：代理 IP 池代码必须要用 Redis 吗？
回答 1：代理 IP 池代码不是必须使用 Redis，代理 IP 池代码也可以改用 MySQL 或者本地文件存储，只是 Redis 读取速度更快，更适合代理 IP 池高频调用的场景。
问题 2：代理 IP 池代码里校验函数为什么请求 httpbin？
回答 2：代理 IP 池代码使用 httpbin 作为测试地址，是为了验证代理 IP 是否可以正常连通外网，代理 IP 池代码可以自行替换成其他测试网址。
问题 3：代理 IP 池代码可以改成多进程并发校验吗？
回答 3：代理 IP 池代码支持改成多进程并发校验，代理 IP 池代码可以引入 concurrent.futures，把 IP 校验任务并行执行，提升大批量 IP 检测速度。
问题 4：代理 IP 池代码部署之后怎么查看池内 IP 总数？
回答 4：代理 IP 池代码可以新增接口调用 r.scard (POOL_KEY)，代理 IP 池代码新增这个接口之后，就可以随时查询代理 IP 池里面当前可用 IP 总数量。
【参考资料】
[2026-09-28]（Python 自建代理 IP 池代码实战）开发者技术社区
[2026-08-16]（Flask+Redis 代理 IP 池开发教程）开源技术博客
[2026-06-21]（APScheduler 定时任务代理 IP 池巡检实现）网络技术站点
【免责申明】
本文代码仅用于技术学习与合法业务开发参考，部署和使用代理 IP 池代码请严格遵守国家相关法律法规，不得开展违规网络访问活动，相关使用责任由使用者自行承担。