Python脚本爬虫代理IP如何轮换

本文目录导读：

1. 使用代理IP池
2. 带验证的代理池
3. 使用requests.Session自动轮换
4. 使用第三方库
5. 高级代理轮换策略
6. 从代理提供商获取代理
7. 最佳实践建议
8. 完整的代理轮换示例

在Python爬虫中轮换代理IP是防止IP被封禁的重要策略,以下是几种常用的代理IP轮换方法：

使用代理IP池

基础代理池实现

import random
import requests
from itertools import cycle
class ProxyPool:
    def __init__(self):
        self.proxies = [
            'http://proxy1.example.com:8080',
            'http://proxy2.example.com:8080',
            'http://proxy3.example.com:8080',
            # 添加更多代理
        ]
        self.proxy_cycle = cycle(self.proxies)
    def get_random_proxy(self):
        """随机返回一个代理"""
        return random.choice(self.proxies)
    def get_next_proxy(self):
        """轮流返回下一个代理"""
        return next(self.proxy_cycle)
# 使用示例
proxy_pool = ProxyPool()
proxies = {
    'http': proxy_pool.get_random_proxy(),
    'https': proxy_pool.get_random_proxy()
}
response = requests.get('http://example.com', proxies=proxies)

带验证的代理池

import requests
import time
from typing import List, Dict
import threading
class ValidatedProxyPool:
    def __init__(self):
        self.proxies = []  # 存储有效的代理
        self.lock = threading.Lock()
        self.test_url = 'http://httpbin.org/ip'
        self.timeout = 5
    def add_proxy(self, proxy: Dict[str, str]):
        """添加代理并验证"""
        if self.validate_proxy(proxy):
            with self.lock:
                self.proxies.append(proxy)
            return True
        return False
    def validate_proxy(self, proxy: Dict[str, str]) -> bool:
        """验证代理是否可用"""
        try:
            response = requests.get(
                self.test_url, 
                proxies=proxy, 
                timeout=self.timeout
            )
            return response.status_code == 200
        except:
            return False
    def get_valid_proxy(self) -> Dict[str, str]:
        """获取有效的代理"""
        with self.lock:
            if not self.proxies:
                raise Exception("没有可用的代理")
            # 获取代理并验证
            proxy = random.choice(self.proxies)
            if self.validate_proxy(proxy):
                return proxy
            else:
                # 移除无效代理
                self.proxies.remove(proxy)
                return self.get_valid_proxy()
    def refresh_pool(self, proxy_list: List[Dict[str, str]]):
        """批量刷新代理池"""
        valid_proxies = []
        for proxy in proxy_list:
            if self.validate_proxy(proxy):
                valid_proxies.append(proxy)
        with self.lock:
            self.proxies = valid_proxies
# 使用示例
pool = ValidatedProxyPool()
pool.add_proxy({
    'http': 'http://proxy1.example.com:8080',
    'https': 'http://proxy1.example.com:8080'
})
# 获取有效代理
proxy = pool.get_valid_proxy()
response = requests.get('http://example.com', proxies=proxy)

使用requests.Session自动轮换

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import random
class RotatingProxySession:
    def __init__(self, proxy_list):
        self.proxy_list = proxy_list
        self.session = requests.Session()
        # 配置重试策略
        retry_strategy = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        self.session.mount("http://", adapter)
        self.session.mount("https://", adapter)
    def get_proxy(self):
        """获取随机代理"""
        return random.choice(self.proxy_list)
    def request(self, url, **kwargs):
        """发送请求时自动轮换代理"""
        max_retries = 3
        for attempt in range(max_retries):
            try:
                proxy = self.get_proxy()
                kwargs['proxies'] = proxy
                response = self.session.get(url, timeout=10, **kwargs)
                return response
            except Exception as e:
                if attempt == max_retries - 1:
                    raise e
                print(f"代理 {proxy} 失败，重试...")
# 使用示例
proxy_list = [
    {'http': 'http://proxy1:8080', 'https': 'http://proxy1:8080'},
    {'http': 'http://proxy2:8080', 'https': 'http://proxy2:8080'},
]
session = RotatingProxySession(proxy_list)
response = session.request('http://example.com')

使用第三方库

使用 requests-ip-rotator

# 安装：pip install requests-ip-rotator
from requests_ip_rotator import ApiGateway
# AWS API Gateway IP轮换
gateway = ApiGateway("https://example.com")
gateway.start()
session = requests.Session()
session.mount("https://example.com", gateway)
response = session.get("https://example.com/api/data")
gateway.shutdown()

使用 scrapy-rotating-proxies (Scrapy框架)

# settings.py
ROTATING_PROXY_LIST = [
    'proxy1.com:8000',
    'proxy2.com:8031',
    # ...
]
ROTATING_PROXY_PAGE_SIZE = 20  # 同时使用多少个代理
DOWNLOADER_MIDDLEWARES = {
    'rotating_proxies.middlewares.RotatingProxyMiddleware': 610,
    'rotating_proxies.middlewares.BanDetectionMiddleware': 620,
}

高级代理轮换策略

import random
import time
from collections import deque
import requests
class AdvancedRotatingProxy:
    def __init__(self, proxy_list):
        self.proxy_list = proxy_list
        self.proxy_queue = deque()
        self.used_proxies = set()
        self.proxy_stats = {}  # 记录每个代理的使用情况
    def init_queue(self):
        """初始化代理队列"""
        random.shuffle(self.proxy_list)
        self.proxy_queue = deque(self.proxy_list)
    def get_next_proxy(self):
        """获取下一个代理"""
        if not self.proxy_queue:
            self.init_queue()
        proxy = self.proxy_queue.popleft()
        # 避免短时间内重复使用
        if proxy in self.used_proxies:
            self.proxy_queue.append(proxy)
            time.sleep(1)
            return self.get_next_proxy()
        self.used_proxies.add(proxy)
        return proxy
    def release_proxy(self, proxy):
        """释放代理回队列"""
        if proxy in self.used_proxies:
            self.used_proxies.remove(proxy)
            self.proxy_queue.append(proxy)
        # 更新统计
        if proxy not in self.proxy_stats:
            self.proxy_stats[proxy] = {'success': 0, 'fail': 0}
        self.proxy_stats[proxy]['success'] += 1
    def mark_failed_proxy(self, proxy):
        """标记失败的代理"""
        if proxy not in self.proxy_stats:
            self.proxy_stats[proxy] = {'success': 0, 'fail': 0}
        self.proxy_stats[proxy]['fail'] += 1
        # 如果失败率过高，移除该代理
        stats = self.proxy_stats[proxy]
        if stats['fail'] > 3 and stats['success'] / (stats['fail'] + stats['success']) < 0.3:
            self.proxy_list.remove(proxy)
    def fetch_with_rotation(self, url, max_retries=3):
        """带轮换的请求"""
        for _ in range(max_retries):
            proxy = self.get_next_proxy()
            try:
                response = requests.get(url, proxies={'http': proxy, 'https': proxy}, timeout=10)
                self.release_proxy(proxy)
                return response
            except Exception as e:
                self.mark_failed_proxy(proxy)
                print(f"代理 {proxy} 失败: {e}")
                time.sleep(1)
        raise Exception("所有代理都失败")
# 使用示例
proxy_list = [
    'http://proxy1:8080',
    'http://proxy2:8080',
    'http://proxy3:8080',
]
rotator = AdvancedRotatingProxy(proxy_list)
response = rotator.fetch_with_rotation('http://example.com')

从代理提供商获取代理

import requests
import json
def get_proxies_from_provider(api_url, api_key):
    """从代理服务商获取代理列表"""
    headers = {'Authorization': f'Bearer {api_key}'}
    response = requests.get(api_url, headers=headers)
    if response.status_code == 200:
        data = response.json()
        proxies = []
        for item in data.get('data', []):
            proxy = {
                'http': f"http://{item['ip']}:{item['port']}",
                'https': f"http://{item['ip']}:{item['port']}"
            }
            proxies.append(proxy)
        return proxies
    return []

最佳实践建议

1. 代理池大小：至少准备20-50个代理
2. 轮换频率：不要在每次请求都换代理，可以每5-10次请求换一次
3. 错误处理：实现失败重试机制
4. 代理验证：定期验证代理可用性
5. 随机延迟：在请求间添加随机延迟
6. User-Agent轮换：同时轮换User-Agent

完整的代理轮换示例

import random
import time
import requests
from typing import List, Dict
class CompleteProxyRotator:
    def __init__(self, proxies: List[Dict[str, str]]):
        self.proxies = proxies
        self.current_index = 0
        self.request_count = 0
        self.failed_proxies = set()
    def get_user_agent(self) -> str:
        """随机User-Agent"""
        user_agents = [
            'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36',
            # 更多UA...
        ]
        return random.choice(user_agents)
    def get_proxy(self) -> Dict[str, str]:
        """获取代理"""
        available = [p for p in self.proxies if str(p) not in self.failed_proxies]
        if not available:
            self.failed_proxies.clear()
            available = self.proxies
        # 每5次请求换一次代理
        if self.request_count % 5 == 0:
            self.current_index = random.randint(0, len(available) - 1)
        self.request_count += 1
        return available[self.current_index]
    def request(self, url: str, **kwargs) -> requests.Response:
        """发送请求"""
        headers = kwargs.pop('headers', {})
        headers['User-Agent'] = self.get_user_agent()
        max_retries = 3
        for _ in range(max_retries):
            proxy = self.get_proxy()
            try:
                response = requests.get(
                    url,
                    headers=headers,
                    proxies=proxy,
                    timeout=10,
                    **kwargs
                )
                # 随机延迟
                time.sleep(random.uniform(1, 3))
                return response
            except Exception as e:
                self.failed_proxies.add(str(proxy))
                print(f"代理 {proxy} 失败: {e}")
                time.sleep(2)
        raise Exception("所有代理尝试均失败")
# 使用
proxies = [
    {'http': 'http://proxy1:8080', 'https': 'http://proxy1:8080'},
    {'http': 'http://proxy2:8080', 'https': 'http://proxy2:8080'},
]
rotator = CompleteProxyRotator(proxies)
response = rotator.request('http://example.com')

选择哪种方法取决于你的具体需求：

• 小规模爬虫：使用基础代理池即可
• 中等规模：使用带验证的代理池
• 大规模爬虫：建议使用专业的代理服务+轮换策略