Python爬虫实战：用requests搭配免费代理IP绕过反爬，附西刺/快代理实测代码

Python爬虫实战：高效构建免费代理IP池与智能切换策略

在数据采集领域，反爬机制如同横亘在开发者面前的隐形高墙。当你的爬虫频繁遭遇403 Forbidden或请求频率限制时，代理IP便成了突破封锁的利器。本文将带你深入实战，从零构建一个高可用的免费代理IP池，并实现智能切换机制，让爬虫在公开数据采集时游刃有余。

1. 代理IP基础与核心挑战

代理IP的本质是网络请求的中转站，它像变色龙一样让爬虫隐藏真实身份。但免费代理的江湖鱼龙混杂，约70%的公开代理IP在首次测试时就无法使用，剩下的30%中又有半数会在几小时内失效。这种不稳定性让许多开发者望而却步。

免费代理的主要痛点集中在三个方面：

• 存活率低：公开代理平均存活时间不足2小时
• 响应延迟：多数代理的响应时间超过3秒
• 协议限制：部分代理仅支持HTTP或HTTPS单一协议

# 典型代理IP格式示例 proxies = { 'http': 'http://58.218.214.138:33128', 'https': 'https://58.218.214.138:33128' }

提示：测试代理时建议使用httpbin.org/ip作为验证端点，该服务会返回请求使用的IP地址

2. 构建高可用代理IP池

2.1 代理源选择与采集策略

西刺代理和快代理是较为稳定的免费代理来源，但直接爬取它们的页面需要处理分页和反爬机制。更聪明的做法是：

1. 分析网站的分页规律（通常为/page/1形式）
2. 设置随机User-Agent和请求间隔（建议3-5秒）
3. 使用lxml或pyquery解析HTML表格数据

from pyquery import PyQuery as pq import requests def fetch_xici_proxies(page=1): url = f'https://www.xicidaili.com/nn/{page}' headers = {'User-Agent': 'Mozilla/5.0'} resp = requests.get(url, headers=headers) doc = pq(resp.text) proxies = [] for tr in doc('#ip_list tr').items(): if tr.find('td').length > 0: ip = tr.find('td:nth-child(2)').text() port = tr.find('td:nth-child(3)').text() protocol = tr.find('td:nth-child(6)').text().lower() proxies.append(f"{protocol}://{ip}:{port}") return proxies

2.2 代理验证与分级存储

采集到的代理需要经过严格验证才能入库。建议采用多级验证策略：

import concurrent.futures def validate_proxy(proxy): try: test_url = "http://httpbin.org/ip" protocols = ['http'] if 'http' in proxy else ['https'] for protocol in protocols: proxies = {protocol: proxy} resp = requests.get(test_url, proxies=proxies, timeout=5) if resp.json().get('origin'): return True except: return False return False # 使用线程池批量验证 with concurrent.futures.ThreadPoolExecutor() as executor: results = executor.map(validate_proxy, proxy_list) valid_proxies = [p for p, r in zip(proxy_list, results) if r]

3. 智能代理调度系统

3.1 基于响应时间的动态权重

简单的随机选择代理会导致性能不稳定。更优的方案是根据历史表现动态调整选择概率：

1. 记录每个代理的平均响应时间
2. 计算响应时间百分位（如P90）
3. 给响应快的代理更高选中概率

from collections import defaultdict import random import time class ProxyPool: def __init__(self): self.proxies = defaultdict(dict) self.history = defaultdict(list) def update_stats(self, proxy, response_time): self.history[proxy].append(response_time) if len(self.history[proxy]) > 10: self.history[proxy].pop(0) avg_time = sum(self.history[proxy])/len(self.history[proxy]) self.proxies[proxy]['weight'] = 1/(avg_time + 0.1) # 避免除零 def get_proxy(self): total_weight = sum(p['weight'] for p in self.proxies.values()) rand = random.uniform(0, total_weight) upto = 0 for proxy, data in self.proxies.items(): if upto + data['weight'] >= rand: return proxy upto += data['weight'] return random.choice(list(self.proxies.keys()))

3.2 失效代理的自动淘汰机制

维护代理池的关键在于及时清理失效节点。建议实现以下机制：

• 连续失败计数器：超过3次失败立即移出池子
• 定期全量验证：每小时重新验证全部代理
• 异常状态码处理：遇到407/502等代码时降级权重

def health_check(pool): while True: time.sleep(3600) # 每小时检查一次 dead_proxies = [] for proxy in pool.proxies: if not validate_proxy(proxy): dead_proxies.append(proxy) for proxy in dead_proxies: pool.remove_proxy(proxy)

4. 实战：集成代理的爬虫架构

4.1 请求重试与代理切换

将代理池与requests.Session结合，打造健壮的请求器：

class RobustRequestor: def __init__(self, proxy_pool): self.proxy_pool = proxy_pool self.session = requests.Session() self.max_retries = 3 def get(self, url, **kwargs): for attempt in range(self.max_retries): proxy = self.proxy_pool.get_proxy() proxies = {'http': proxy, 'https': proxy} try: start = time.time() resp = self.session.get(url, proxies=proxies, timeout=10, **kwargs) response_time = time.time() - start if resp.status_code == 200: self.proxy_pool.update_stats(proxy, response_time) return resp else: self.proxy_pool.mark_failure(proxy) except Exception as e: self.proxy_pool.mark_failure(proxy) raise Exception(f"Failed after {self.max_retries} attempts")

4.2 分布式代理池方案

当单机代理池不足以支撑大规模采集时，可以考虑：

1. 使用Redis存储代理状态
2. 实现代理节点的分布式锁
3. 通过消息队列同步代理更新

import redis import json class RedisProxyPool: def __init__(self): self.conn = redis.Redis() self.lock = self.conn.lock('proxy_pool_lock') def add_proxy(self, proxy): with self.lock: self.conn.hset('proxies', proxy, json.dumps({ 'weight': 1.0, 'last_used': time.time() })) def get_proxy(self): with self.lock: all_proxies = self.conn.hgetall('proxies') # 权重选择逻辑...

5. 性能优化与特殊场景处理

5.1 连接池调优

默认情况下requests会为每个代理创建独立连接池，这可能导致资源浪费。可以通过适配器调整：

from requests.adapters import HTTPAdapter session = requests.Session() adapter = HTTPAdapter( pool_connections=10, # 每个代理保留的连接数 pool_maxsize=50, # 连接池最大容量 max_retries=2 # 请求重试次数 ) session.mount('http://', adapter) session.mount('https://', adapter)

5.2 处理特殊反爬策略

某些网站会检测代理特征，此时需要：

• 随机化请求间隔（2-8秒之间）
• 轮换User-Agent
• 混合使用代理和直连
• 添加合理的Referer头

from fake_useragent import UserAgent ua = UserAgent() headers = { 'User-Agent': ua.random, 'Referer': 'https://www.google.com/', 'Accept-Language': 'en-US,en;q=0.9' }

在长期爬虫项目中，维护代理池就像打理一个花园——需要定期清除杂草（失效代理），培育优质品种（快速稳定的代理），并保持物种多样性（多来源代理）。我曾在一个电商价格监控项目中，用这套方法将请求成功率从最初的35%提升到了82%，关键就在于实现了代理的智能预热和实时淘汰。