requests使用ip代理池：Python爬虫IP轮换的代码实现

为什么需要IP代理池

在日常网络数据采集中，频繁使用同一IP地址发送请求容易触发目标网站的访问限制。IP代理池通过轮流使用多个代理IP，将请求分散到不同的IP地址上，有效降低被识别为异常访问的概率。这种方式不仅能提高数据采集的成功率，还能保持网络请求的稳定性。

神龙IP代理提供的动态高级套餐特别适合构建IP代理池，其日更200万+IP资源库能充分满足轮换需求，6Mbps带宽保证请求速度，2-360小时可调的IP存活时间让用户能根据业务特点灵活配置。

搭建基础IP代理池

一个基本的IP代理池需要包含IP获取、验证、存储和轮换四个核心功能。我们先从最简单的本地代理池开始实现：

import requests
import time
from threading import Lock

class SimpleProxyPool:
    def __init__(self):
        self.proxies = []
        self.current_index = 0
        self.lock = Lock()
    
    def add_proxy(self, proxy_dict):
        """添加代理到池中"""
        if self.validate_proxy(proxy_dict):
            self.proxies.append(proxy_dict)
            return True
        return False
    
    def get_proxy(self):
        """轮询获取代理"""
        with self.lock:
            if not self.proxies:
                return None
            proxy = self.proxies[self.current_index]
            self.current_index = (self.current_index + 1) % len(self.proxies)
            return proxy
    
    def validate_proxy(self, proxy_dict):
        """验证代理是否可用"""
        try:
            response = requests.get('http://httpbin.org/ip', 
                                  proxies=proxy_dict, 
                                  timeout=5)
            return response.status_code == 200
        except:
            return False

集成神龙IP代理服务

将神龙IP代理API集成到代理池中，可以自动获取新鲜可用的代理IP。神龙IP代理支持多种协议，这里我们以HTTP/HTTPS为例：

class ShenlongProxyPool(SimpleProxyPool):
    def __init__(self, api_url, api_key):
        super().__init__()
        self.api_url = api_url
        self.api_key = api_key
        self.refresh_interval = 300   5分钟刷新一次
        self.last_refresh = 0
    
    def refresh_proxies(self):
        """从神龙IP代理API获取新IP"""
        current_time = time.time()
        if current_time - self.last_refresh < self.refresh_interval:
            return
        
        try:
            headers = {'Authorization': f'Bearer {self.api_key}'}
            response = requests.get(self.api_url, headers=headers, timeout=10)
            if response.status_code == 200:
                new_proxies = response.json()['proxies']
                self.proxies.clear()
                
                for proxy_info in new_proxies:
                    proxy_dict = {
                        'http': f"http://{proxy_info['ip']}:{proxy_info['port']}",
                        'https': f"http://{proxy_info['ip']}:{proxy_info['port']}"
                    }
                    self.add_proxy(proxy_dict)
                
                self.last_refresh = current_time
                print(f"成功刷新 {len(new_proxies)} 个代理IP")
                
        except Exception as e:
            print(f"刷新代理失败: {e}")

在requests中实现智能轮换

结合神龙IP代理的轮换特性，我们可以创建一个智能的请求会话类：

import random

class SmartRequestSession:
    def __init__(self, proxy_pool):
        self.proxy_pool = proxy_pool
        self.session = requests.Session()
        self.max_retries = 3
    
    def request_with_retry(self, method, url, kwargs):
        """带重试机制的请求方法"""
        for attempt in range(self.max_retries):
            try:
                 每次请求前刷新代理池
                self.proxy_pool.refresh_proxies()
                proxy = self.proxy_pool.get_proxy()
                
                if proxy:
                    kwargs['proxies'] = proxy
                
                response = self.session.request(method, url, kwargs)
                
                if response.status_code == 200:
                    return response
                else:
                    print(f"请求失败，状态码: {response.status_code}")
                    
            except requests.exceptions.RequestException as e:
                print(f"第{attempt+1}次请求失败: {e}")
                if attempt == self.max_retries - 1:
                    raise e
        
        return None

 使用示例
proxy_pool = ShenlongProxyPool('您的API地址', '您的API密钥')
smart_session = SmartRequestSession(proxy_pool)

 发送请求
response = smart_session.request_with_retry('GET', 'https://httpbin.org/ip')
if response:
    print("请求成功:", response.json())

错误处理与性能优化

在实际使用中，需要充分考虑网络异常和代理失效的情况。以下是几个关键的优化点：

代理质量监控：定期检测池中代理的响应速度和可用性，自动剔除失效代理。

def monitor_proxy_quality(self):
    """监控代理质量"""
    quality_threshold = 2   秒
    for proxy in self.proxies[:]:   使用切片创建副本避免修改问题
        try:
            start_time = time.time()
            response = requests.get('https://httpbin.org/ip', 
                                  proxies=proxy, 
                                  timeout=5)
            response_time = time.time() - start_time
            
            if response_time > quality_threshold or response.status_code != 200:
                self.proxies.remove(proxy)
                print(f"移除慢速代理: {proxy}")
                
        except:
            self.proxies.remove(proxy)
            print(f"移除失效代理: {proxy}")

连接池优化：使用requests的Session对象保持连接复用，减少TCP连接建立的开销。

超时设置：合理设置连接超时和读取超时，避免因单个代理问题导致整个程序阻塞。

神龙IP代理套餐选择指南

根据不同的使用场景，选择合适的代理套餐至关重要：

套餐类型	适用场景	核心优势
动态高级套餐	高频请求、需要大量IP轮换	日更200万+IP，灵活控制存活时间
动态独享套餐	对带宽要求高的业务	10Mbps峰值带宽，性能更优
静态高级套餐	需要长期稳定IP的场景	IP长期固定，高纯净度