在Scrapy框架中设置代理：完整的代码示例指南

Scrapy是最强大的Python网络抓取框架之一，但如果没有正确设置代理，您的爬虫在几分钟内就会被封锁。在本指南中，我将展示将代理集成到Scrapy中的所有方法：从最简单的设置到自动处理错误的高级IP轮换方法。

本材料基于抓取大型电子商务平台和受保护网站的实际经验。您将获得可以立即在项目中使用的代码示例。

为什么Scrapy在没有代理的情况下会被封锁

现代网站使用多层次的抓取保护。即使您设置了用户代理和请求之间的延迟，您的IP地址也会因几个迹象而显示出自动化：

• 请求频率：一个IP每分钟发出100个以上的请求——明显的机器人迹象
• 行为模式：顺序访问页面而没有随机跳转
• 缺乏JavaScript：Scrapy不执行JS，容易被检测到
• 地理位置：来自数据中心而不是家庭网络的访问

结果——IP被封锁几个小时或几天。尤其是市场（亚马逊、Wildberries、Ozon）、社交网络和使用Cloudflare的网站使用了特别激进的保护。代理解决了这个问题，通过多个IP地址分散请求。

在Scrapy中基本设置代理

最简单的方法是直接在爬虫设置中指定代理。此方法适用于测试或使用单个代理服务器抓取小量数据。

方法1：通过请求中的meta

import scrapy

class MySpider(scrapy.Spider):
    name = 'example'
    start_urls = ['https://example.com']
    
    def start_requests(self):
        proxy = 'http://username:password@proxy.example.com:8080'
        
        for url in self.start_urls:
            yield scrapy.Request(
                url=url,
                callback=self.parse,
                meta={'proxy': proxy}
            )
    
    def parse(self, response):
        # 您的解析逻辑
        self.log(f'Scraped {response.url} via {response.meta["proxy"]}')

代理的格式取决于协议和认证方法：

• http://proxy.example.com:8080 — 无需认证
• http://user:pass@proxy.example.com:8080 — 需要用户名/密码
• socks5://user:pass@proxy.example.com:1080 — SOCKS5代理

方法2：在settings.py中全局设置

# settings.py

# 所有请求的HTTP代理
HTTPPROXY_ENABLED = True
HTTPPROXY_AUTH_ENCODING = 'utf-8'

# 通过环境变量设置
HTTP_PROXY = 'http://username:password@proxy.example.com:8080'
HTTPS_PROXY = 'http://username:password@proxy.example.com:8080'

此方法便于快速测试，但不适合生产环境：没有IP轮换，代理崩溃时整个爬虫停止，无法为不同网站使用不同代理。

创建自定义代理中间件

对于生产抓取，需要一个自定义中间件来管理代理池、处理错误并自动轮换IP。以下是基本实现：

# middlewares.py

import random
from scrapy import signals
from scrapy.exceptions import NotConfigured

class RandomProxyMiddleware:
    def __init__(self, proxy_list):
        self.proxy_list = proxy_list
    
    @classmethod
    def from_crawler(cls, crawler):
        # 从设置中加载代理列表
        proxy_list = crawler.settings.getlist('PROXY_LIST')
        
        if not proxy_list:
            raise NotConfigured('PROXY_LIST未配置')
        
        return cls(proxy_list)
    
    def process_request(self, request, spider):
        # 从池中选择随机代理
        proxy = random.choice(self.proxy_list)
        request.meta['proxy'] = proxy
        
        spider.logger.info(f'使用代理: {proxy}')
    
    def process_exception(self, request, exception, spider):
        # 出现错误时尝试其他代理
        proxy = random.choice(self.proxy_list)
        request.meta['proxy'] = proxy
        
        spider.logger.warning(
            f'代理错误，切换到: {proxy}'
        )
        
        return request

现在在settings.py中配置使用中间件：

# settings.py

# 代理列表（可以从文件或API加载）
PROXY_LIST = [
    'http://user1:pass1@proxy1.example.com:8080',
    'http://user2:pass2@proxy2.example.com:8080',
    'http://user3:pass3@proxy3.example.com:8080',
    # ... 添加50个以上的代理以实现有效轮换
]

# 启用中间件
DOWNLOADER_MIDDLEWARES = {
    'myproject.middlewares.RandomProxyMiddleware': 350,
    'scrapy.downloadermiddlewares.httpproxy.HttpProxyMiddleware': 400,
}

# 错误重试次数
RETRY_TIMES = 3
RETRY_HTTP_CODES = [500, 502, 503, 504, 408, 429]

代理轮换：三种有效方法

随机选择代理（如上例所示）是最简单但不是最有效的方法。我们来看看三种适用于不同场景的轮换策略。

方法1：轮询（顺序轮换）

代理按顺序选择。适合均匀分配负载：

class RoundRobinProxyMiddleware:
    def __init__(self, proxy_list):
        self.proxy_list = proxy_list
        self.current_index = 0
    
    @classmethod
    def from_crawler(cls, crawler):
        proxy_list = crawler.settings.getlist('PROXY_LIST')
        return cls(proxy_list)
    
    def process_request(self, request, spider):
        # 轮流获取下一个代理
        proxy = self.proxy_list[self.current_index]
        self.current_index = (self.current_index + 1) % len(self.proxy_list)
        
        request.meta['proxy'] = proxy

方法2：智能轮换与黑名单

跟踪问题代理并暂时将其排除在轮换之外：

import time
from collections import defaultdict

class SmartProxyMiddleware:
    def __init__(self, proxy_list):
        self.proxy_list = proxy_list
        self.proxy_errors = defaultdict(int)
        self.blacklist = set()
        self.blacklist_timeout = 300  # 5分钟
        self.blacklist_time = {}
    
    @classmethod
    def from_crawler(cls, crawler):
        proxy_list = crawler.settings.getlist('PROXY_LIST')
        return cls(proxy_list)
    
    def get_working_proxies(self):
        # 从黑名单中移除超时的代理
        current_time = time.time()
        expired = [
            proxy for proxy, ban_time in self.blacklist_time.items()
            if current_time - ban_time > self.blacklist_timeout
        ]
        
        for proxy in expired:
            self.blacklist.discard(proxy)
            self.proxy_errors[proxy] = 0
        
        # 返回可用的代理
        return [p for p in self.proxy_list if p not in self.blacklist]
    
    def process_request(self, request, spider):
        working_proxies = self.get_working_proxies()
        
        if not working_proxies:
            spider.logger.error('所有代理都被列入黑名单！')
            return
        
        proxy = random.choice(working_proxies)
        request.meta['proxy'] = proxy
    
    def process_response(self, request, response, spider):
        # 如果收到封锁——添加到黑名单
        if response.status in [403, 429, 503]:
            proxy = request.meta.get('proxy')
            self.proxy_errors[proxy] += 1
            
            if self.proxy_errors[proxy] >= 3:
                self.blacklist.add(proxy)
                self.blacklist_time[proxy] = time.time()
                spider.logger.warning(
                    f'代理 {proxy} 被列入黑名单 {self.blacklist_timeout}s'
                )
        
        return response

方法3：通过提供商的API进行轮换

许多代理提供商（包括 住宅代理）提供旋转端点——一个URL，每次请求自动更改IP：

# settings.py

# 自动轮换的单一端点
ROTATING_PROXY = 'http://username:password@rotating.proxy.com:8080'

# 简单的中间件
class RotatingProxyMiddleware:
    def __init__(self, proxy):
        self.proxy = proxy
    
    @classmethod
    def from_crawler(cls, crawler):
        proxy = crawler.settings.get('ROTATING_PROXY')
        return cls(proxy)
    
    def process_request(self, request, spider):
        # 一个URL，但每个请求都使用新的IP
        request.meta['proxy'] = self.proxy

这是生产环境中最方便的方法：无需管理代理池，提供商会自行监控IP质量并替换有问题的。特别适合 住宅代理，其IP池可达到数百万个地址。

认证：用户名/密码与IP白名单

代理提供商提供两种认证方法。选择会影响连接速度和设置的便利性。

用户名：密码认证

用户名和密码通过代理的URL传递。Scrapy会自动将其转换为HTTP头 Proxy-Authorization：

proxy = 'http://username:password@proxy.example.com:8080'
request.meta['proxy'] = proxy

# Scrapy会自动添加头部：
# Proxy-Authorization: Basic dXNlcm5hbWU6cGFzc3dvcmQ=

优点：可以从任何IP工作，轻松在代码中更改代理。
缺点：每个请求有小的开销（约50-100毫秒），凭据以明文形式出现在代码中。

IP白名单认证

您将服务器的IP添加到提供商的白名单中，无需认证：

proxy = 'http://proxy.example.com:8080'  # 无需用户名/密码
request.meta['proxy'] = proxy

优点：比用户名/密码快50-100毫秒，更安全（代码中没有凭据）。
缺点：仅适用于特定IP，服务器更换时需要更新白名单。

错误处理和自动更换IP

即使使用高质量的代理也会出现错误：超时、连接拒绝、封锁。正确的错误处理对爬虫的稳定运行至关重要。

处理HTTP状态

class ProxyMiddleware:
    def process_response(self, request, response, spider):
        # 需要更换代理并重试的状态码
        ban_codes = [403, 407, 429, 503]
        
        if response.status in ban_codes:
            proxy = request.meta.get('proxy')
            spider.logger.warning(
                f'从 {proxy} 收到 {response.status}，正在重试...'
            )
            
            # 标记为重试并使用新代理
            request.meta['dont_retry'] = False
            request.meta['proxy'] = self.get_new_proxy()
            
            return request
        
        return response

处理网络异常

from twisted.internet.error import TimeoutError, ConnectionRefusedError
from scrapy.exceptions import IgnoreRequest

class ProxyMiddleware:
    def process_exception(self, request, exception, spider):
        # 代理连接错误
        proxy_errors = (
            TimeoutError,
            ConnectionRefusedError,
            ConnectionLost,
        )
        
        if isinstance(exception, proxy_errors):
            proxy = request.meta.get('proxy')
            spider.logger.error(
                f'代理 {proxy} 连接失败: {exception}'
            )
            
            # 更换代理并重试
            request.meta['proxy'] = self.get_new_proxy()
            return request
        
        # 对于其他错误使用标准处理
        return None

根据内容检测封锁

一些网站返回HTTP 200，但显示验证码或封锁页面：

class ProxyMiddleware:
    def process_response(self, request, response, spider):
        # 内容中的封锁迹象
        ban_indicators = [
            'captcha',
            'access denied',
            'blocked',
            'unusual traffic',
            'robot check',
        ]
        
        body_text = response.text.lower()
        
        if any(indicator in body_text for indicator in ban_indicators):
            spider.logger.warning(
                f'检测到来自 {request.meta.get("proxy")} 的封锁页面'
            )
            
            # 更换代理并重试
            request.meta['proxy'] = self.get_new_proxy()
            return request
        
        return response

选择哪种类型的代理用于Scrapy

选择代理类型取决于目标网站、预算和所需的抓取速度。以下是主要选项的比较：

选择建议

对于抓取市场（亚马逊、Wildberries、Ozon、AliExpress）——仅使用住宅代理。这些网站会激进地封锁数据中心。需要轮换和地理定位（例如，Wildberries需要俄罗斯IP）。

对于抓取新闻网站、博客、论坛——数据中心代理是合适的。保护措施最低，速度和低流量成本很重要。

对于抓取使用Cloudflare的网站——必须使用住宅代理。数据中心的Cloudflare几乎会立即检测到。为Scrapy添加cloudscraper库以绕过JS挑战。

对于抓取Google搜索、SEO工具——使用带有地理定位的住宅代理。Google会为不同国家和城市显示不同的结果。

高级技术：会话和粘性IP

在抓取某些网站时，需要在整个会话期间保持一个IP（认证、购物车、多步骤表单）。以下是在Scrapy中实现粘性会话的方法。

针对一个域的粘性IP

from urllib.parse import urlparse

class StickyProxyMiddleware:
    def __init__(self, proxy_list):
        self.proxy_list = proxy_list
        # 字典：域名 -> 代理
        self.domain_proxy_map = {}
    
    @classmethod
    def from_crawler(cls, crawler):
        proxy_list = crawler.settings.getlist('PROXY_LIST')
        return cls(proxy_list)
    
    def process_request(self, request, spider):
        # 从URL中提取域名
        domain = urlparse(request.url).netloc
        
        # 如果该域名已经有代理——使用它
        if domain in self.domain_proxy_map:
            proxy = self.domain_proxy_map[domain]
        else:
            # 否则选择一个新的并记住
            proxy = random.choice(self.proxy_list)
            self.domain_proxy_map[domain] = proxy
            spider.logger.info(f'分配 {proxy} 给 {domain}')
        
        request.meta['proxy'] = proxy

带有会话超时的粘性IP

更高级的选项：代理在特定时间内（例如10分钟）绑定到域名，然后更换：

import time
from urllib.parse import urlparse

class SessionProxyMiddleware:
    def __init__(self, proxy_list, session_timeout=600):
        self.proxy_list = proxy_list
        self.session_timeout = session_timeout  # 10分钟
        # 字典：域名 -> (代理，创建时间)
        self.sessions = {}
    
    @classmethod
    def from_crawler(cls, crawler):
        proxy_list = crawler.settings.getlist('PROXY_LIST')
        timeout = crawler.settings.getint('PROXY_SESSION_TIMEOUT', 600)
        return cls(proxy_list, timeout)
    
    def get_proxy_for_domain(self, domain):
        current_time = time.time()
        
        # 检查是否有活动会话
        if domain in self.sessions:
            proxy, created_at = self.sessions[domain]
            
            # 如果会话没有过期——使用同一代理
            if current_time - created_at < self.session_timeout:
                return proxy
        
        # 创建一个新的会话并使用新的代理
        new_proxy = random.choice(self.proxy_list)
        self.sessions[domain] = (new_proxy, current_time)
        
        return new_proxy
    
    def process_request(self, request, spider):
        domain = urlparse(request.url).netloc
        proxy = self.get_proxy_for_domain(domain)
        request.meta['proxy'] = proxy

与Cookie中间件集成

对于完整的会话，需要同步代理和cookies。Scrapy为每个域单独存储cookies，但在更换代理时需要清除cookies：

# settings.py

# 启用cookie中间件
COOKIES_ENABLED = True
COOKIES_DEBUG = False

# 用于同步代理和cookies的中间件
class ProxyCookieMiddleware:
    def process_request(self, request, spider):
        # 获取当前代理
        current_proxy = request.meta.get('proxy')
        
        # 如果代理发生变化——清除cookies
        previous_proxy = request.meta.get('previous_proxy')
        
        if previous_proxy and previous_proxy != current_proxy:
            # 清除该域的cookies
            jar = spider.crawler.engine.downloader.middleware.middlewares[0].jars
            domain = urlparse(request.url).netloc
            
            if domain in jar:
                jar[domain].clear()
                spider.logger.info(f'清除 {domain} 的cookies')
        
        request.meta['previous_proxy'] = current_proxy

结论

在Scrapy中正确设置代理是稳定抓取而不被封锁的基础。我们讨论了所有关键方面：从基本集成到高级轮换和会话管理技术。

主要结论：

• 对于生产环境，使用自定义中间件进行智能轮换和问题IP的黑名单处理
• 处理所有类型的错误：HTTP状态、网络异常、内容封锁
• 根据任务选择代理类型：数据中心用于简单网站，住宅代理用于受保护的网站
• 对于需要认证的网站，使用粘性会话将代理绑定到域名
• 从10-50个代理的池开始，随着负载增加进行扩展

如果您计划抓取受保护的网站（市场、社交网络、使用Cloudflare的网站），建议使用 住宅代理——它们提供最大的匿名性和最低的封锁风险。对于高速抓取，选择提供旋转端点和10000个IP地址池的提供商。

本文中的所有代码示例均在Scrapy 2.x上经过测试，并准备在生产环境中使用。根据您的需求调整它们，并随着项目的增长进行扩展。