为什么要搭建自己的代理IP池
在做数据采集、接口测试或者自动化任务时，经常需要用到代理IP来避免被目标网站封禁。市面上的付费代理服务虽然稳定，但在项目初期或者预算有限的情况下，搭建一个免费代理IP池是一个不错的过渡方案。
免费代理IP池的核心思路是：从多个公开来源自动抓取免费代理，然后通过程序批量验证这些代理的可用性，去除重复和失效的代理，最终维护一个可用的代理列表供业务调用。整个过程完全自动化，只需要定期运行即可。
环境准备
搭建代理IP池需要以下基础环境：
# Python 3.8+ pip install requests pip install beautifulsoup4 pip install lxml pip install aiohttp # 用于异步验证
项目目录结构建议如下：
proxy_pool/ ├── crawler.py # 抓取模块 ├── verifier.py # 验证模块 ├── dedup.py # 去重模块 ├── storage.py # 存储模块 ├── scheduler.py # 调度器 └── proxy_pool.db # SQLite数据库
第一步：自动抓取公开代理
公开免费代理的来源有很多，常见的包括一些代理网站和技术社区。下面是一个抓取模块的示例代码：
import requests
from bs4 import BeautifulSoup
import re
class ProxyCrawler:
def __init__(self):
self.headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}
self.sources = [
'https://www.example-free-proxy-list.com',
'https://www.example-proxy-source.net',
]
def crawl(self):
"""从多个来源抓取代理"""
proxies = []
for url in self.sources:
try:
resp = requests.get(url, headers=self.headers, timeout=10)
proxies.extend(self._parse_page(resp.text))
except Exception as e:
print(f"抓取失败 {url}: {e}")
return proxies
def _parse_page(self, html):
"""解析页面提取IP和端口"""
proxies = []
soup = BeautifulSoup(html, 'lxml')
# 根据实际页面结构调整解析规则
rows = soup.select('table tr')
for row in rows[1:]:
cells = row.find_all('td')
if len(cells) >= 2:
ip = cells[0].text.strip()
port = cells[1].text.strip()
if self._is_valid_ip(ip) and port.isdigit():
proxies.append({'ip': ip, 'port': port})
return proxies
def _is_valid_ip(self, ip):
"""验证IP格式"""
pattern = r'^(\d{1,3}\.){3}\d{1,3}$'
return re.match(pattern, ip) is not None
# 使用示例
crawler = ProxyCrawler()
raw_proxies = crawler.crawl()
print(f"共抓取到 {len(raw_proxies)} 个代理")提示：公开代理来源不稳定，建议配置多个数据源，并定期更新解析规则以适应页面变化。
第二步：批量验证代理可用性
抓取到的免费代理大部分是失效的，需要通过实际请求来验证。使用异步并发可以大幅提升验证速度：
import aiohttp
import asyncio
import time
class ProxyVerifier:
def __init__(self, timeout=5, max_concurrency=50):
self.timeout = aiohttp.ClientTimeout(total=timeout)
self.semaphore = asyncio.Semaphore(max_concurrency)
self.test_url = 'https://httpbin.org/ip'
async def verify_single(self, session, proxy_info):
"""验证单个代理"""
ip = proxy_info['ip']
port = proxy_info['port']
proxy_url = f"http://{ip}:{port}"
async with self.semaphore:
try:
start = time.time()
async with session.get(
self.test_url,
proxy=proxy_url,
timeout=self.timeout
) as resp:
if resp.status == 200:
elapsed = round((time.time() - start) * 1000)
return {
'ip': ip,
'port': port,
'status': 'valid',
'response_time': elapsed
}
except Exception:
pass
return {
'ip': ip,
'port': port,
'status': 'invalid',
'response_time': -1
}
async def verify_batch(self, proxy_list):
"""批量验证代理"""
valid_proxies = []
async with aiohttp.ClientSession() as session:
tasks = [self.verify_single(session, p) for p in proxy_list]
results = await asyncio.gather(*tasks)
for r in results:
if r['status'] == 'valid':
valid_proxies.append(r)
print(f"✓ {r['ip']}:{r['port']} - {r['response_time']}ms")
return valid_proxies
# 使用示例
verifier = ProxyVerifier(timeout=5)
valid_proxies = asyncio.run(verifier.verify_batch(raw_proxies))
print(f"验证通过 {len(valid_proxies)} 个")第三步：去重处理
不同来源可能返回相同的代理，需要去重避免重复存储和验证：
class ProxyDeduplicator:
def __init__(self):
self.seen = set()
def deduplicate(self, proxy_list):
"""基于IP+端口去重"""
unique_proxies = []
for proxy in proxy_list:
key = f"{proxy['ip']}:{proxy['port']}"
if key not in self.seen:
self.seen.add(key)
unique_proxies.append(proxy)
return unique_proxies
def remove_duplicates_with_db(self, proxy_list, existing_keys):
"""与数据库已有记录去重"""
return [
p for p in proxy_list
if f"{p['ip']}:{p['port']}" not in existing_keys
]
# 使用示例
dedup = ProxyDeduplicator()
unique_proxies = dedup.deduplicate(raw_proxies)
print(f"去重后剩余 {len(unique_proxies)} 个")第四步：存储与管理
使用SQLite存储验证通过的代理，方便查询和管理：
import sqlite3
from datetime import datetime
class ProxyStorage:
def __init__(self, db_path='proxy_pool.db'):
self.conn = sqlite3.connect(db_path)
self._init_db()
def _init_db(self):
"""初始化数据库表"""
self.conn.execute('''
CREATE TABLE IF NOT EXISTS proxy_pool (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ip TEXT NOT NULL,
port TEXT NOT NULL,
response_time INTEGER DEFAULT 0,
status TEXT DEFAULT 'valid',
last_check TEXT,
check_count INTEGER DEFAULT 1,
UNIQUE(ip, port)
)
''')
self.conn.commit()
def save_proxies(self, proxies):
"""保存代理列表"""
now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
saved = 0
for p in proxies:
try:
self.conn.execute(
'''INSERT OR IGNORE INTO proxy_pool
(ip, port, response_time, status, last_check)
VALUES (?, ?, ?, 'valid', ?)''',
(p['ip'], p['port'], p.get('response_time', 0), now)
)
saved += self.conn.total_changes > 0
except Exception:
pass
self.conn.commit()
return saved
def get_valid_proxy(self):
"""获取一个可用代理"""
cursor = self.conn.execute(
'''SELECT ip, port FROM proxy_pool
WHERE status='valid'
ORDER BY response_time ASC LIMIT 1'''
)
row = cursor.fetchone()
if row:
return f"http://{row[0]}:{row[1]}"
return None
def remove_invalid(self, ip, port):
"""移除失效代理"""
self.conn.execute(
'UPDATE proxy_pool SET status=? WHERE ip=? AND port=?',
('invalid', ip, port)
)
self.conn.commit()
def get_stats(self):
"""获取统计信息"""
cursor = self.conn.execute(
'''SELECT status, COUNT(*) as count
FROM proxy_pool GROUP BY status'''
)
return dict(cursor.fetchall())
# 使用示例
storage = ProxyStorage()
storage.save_proxies(valid_proxies)
print(f"统计: {storage.get_stats()}")第五步：定时调度运行
将抓取、验证、去重、存储串联起来，定时执行保持代理池新鲜：
import schedule
import time
import threading
class ProxyScheduler:
def __init__(self):
self.crawler = ProxyCrawler()
self.verifier = ProxyVerifier(timeout=5)
self.dedup = ProxyDeduplicator()
self.storage = ProxyStorage()
def run_cycle(self):
"""执行一轮完整流程"""
print(f"[{time.strftime('%H:%M:%S')}] 开始抓取代理...")
raw = self.crawler.crawl()
print(f"抓取到 {len(raw)} 个原始代理")
unique = self.dedup.deduplicate(raw)
print(f"去重后 {len(unique)} 个")
valid = asyncio.run(self.verifier.verify_batch(unique))
print(f"验证通过 {len(valid)} 个")
saved = self.storage.save_proxies(valid)
print(f"新增存储 {saved} 个")
print(f"当前池状态: {self.storage.get_stats()}")
def start(self):
"""启动定时任务"""
# 立即执行一次
self.run_cycle()
# 每30分钟执行一次
schedule.every(30).minutes.do(self.run_cycle)
while True:
schedule.run_pending()
time.sleep(1)
# 启动调度器
if __name__ == '__main__':
scheduler = ProxyScheduler()
# 在后台线程运行
thread = threading.Thread(target=scheduler.start, daemon=True)
thread.start()
print("代理池调度器已启动")免费代理池的局限性
通过以上代码可以搭建一个基本可用的免费代理IP池，但在实际使用中会遇到以下问题：
| 问题 | 表现 | 影响 |
|---|---|---|
| 存活率低 | 免费代理平均可用率不足10% | 频繁验证消耗资源 |
| 稳定性差 | 代理随时失效 | 业务请求中断 |
| 速度慢 | 响应时间长，带宽小 | 采集效率低 |
| 匿名度低 | 多数为透明代理 | 真实IP暴露风险 |
| 来源不稳定 | 公开源经常关闭或改版 | 需要持续维护代码 |
更稳定的选择：天启HTTP代理
如果业务对代理的稳定性和速度有要求，建议使用专业的代理服务。天启HTTP提供国内高质量代理IP，具备以下特点：
高匿名性：所有IP均为高匿代理，不会暴露真实IP地址，目标网站无法识别你在使用代理。
稳定可靠：IP资源来自正规渠道，连接稳定，响应速度快，不会出现免费代理频繁掉线的问题。
API接口：提供标准API接口，可以直接对接到你的代理池系统中，按需获取IP，无需自己抓取和验证。
使用示例：
import requests
# 天启HTTP API获取代理
api_url = "https://api.tianqiip.com/getip"
params = {
"key": "你的API密钥",
"count": 10,
"format": "json"
}
resp = requests.get(api_url, params=params)
proxy_list = resp.json()
# 直接使用，无需验证
for proxy in proxy_list:
proxies = {
"http": f"http://{proxy['ip']}:{proxy['port']}",
"https": f"http://{proxy['ip']}:{proxy['port']}"
}
# 业务请求
r = requests.get("https://目标网站.com", proxies=proxies, timeout=10)
print(r.status_code)使用天启HTTP后，你可以省去抓取、验证、去重等繁琐步骤，将精力集中在核心业务上。
常见问题
Q: 免费代理IP池能用于正式业务吗？
不建议。免费代理稳定性差、匿名度低，适合学习和测试用途。正式业务建议使用天启HTTP等专业代理服务。
Q: 代理验证的并发数设多少合适？
一般建议50-100个并发。设置过高可能导致本地网络拥塞或被测试目标限制，过低则验证速度慢。
Q: 多久运行一次抓取和验证？
免费代理失效很快，建议每30分钟到1小时运行一次完整流程。如果使用频率高，可以缩短间隔。
Q: SQLite够用吗？需要换Redis吗？
代理池规模在几千个以内，SQLite完全够用。如果需要多进程共享或更高性能，可以迁移到Redis。
Q: 天启HTTP的API怎么对接到现有代理池？
将天启HTTP的API返回结果直接存入你的数据库即可，省去抓取和验证步骤。API返回的IP已经过验证，可以直接使用。
Q: 免费代理来源去哪里找？
可以搜索"免费代理IP"相关网站，但要注意这些来源随时可能失效。这也是免费代理池维护成本高的原因之一。
