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

wen 实用脚本 2026-07-19 1

本文目录导读：

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

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

在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 []

最佳实践建议

代理池大小：至少准备20-50个代理
轮换频率：不要在每次请求都换代理，可以每5-10次请求换一次
错误处理：实现失败重试机制
代理验证：定期验证代理可用性
随机延迟：在请求间添加随机延迟
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')

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