Scrapy代理中间件完整指南：住宅代理轮换与生产级配置

为什么Scrapy爬虫总是被封锁

你写好了Spider，数据跑得正欢，突然日志里全是403和CAPTCHA页面。单IP高频请求是反爬系统的头号信号。解决方案很明确：Scrapy proxy middleware——在下载器层注入代理，让每个请求看起来来自不同的客户端。

但Scrapy的代理配置远不止在settings.py里填一个HTTP_PROXY那么简单。生产级代理集成需要：轮换策略、失败重试、会话粘滞、封禁检测和监控。本文从Scrapy中间件架构出发，逐步构建一套完整的代理方案。

Scrapy下载器中间件架构

Scrapy的请求/响应处理由Downloader Middleware链驱动。每个中间件实现process_request、process_response或process_exception方法，按settings.py中定义的优先级顺序执行。

与代理相关的核心中间件及其默认优先级：

• HttpProxyMiddleware (750) — 从环境变量HTTP_PROXY读取代理并设置到request.meta['proxy']
• RetryMiddleware (500) — 请求失败时重试，但不切换代理
• RedirectMiddleware (800) — 处理3xx重定向
• RobotsTxtMiddleware (100) — robots.txt合规检查

关键洞察：process_request在请求发送前执行——这是注入代理的最佳位置。而process_response和process_exception则用于检测封禁和处理失败。

默认的HttpProxyMiddleware只读取一个静态代理地址，没有轮换、没有地理定位、没有故障转移。我们需要自己造轮子。

自定义代理轮换中间件

以下是一个完整可用的Scrapy residential proxies轮换中间件，基于ProxyHat的住宅代理池，通过用户名参数实现国家定位和会话粘滞。

import random
import string
import time
from scrapy import signals
from scrapy.exceptions import IgnoreRequest
from scrapy.utils.log import logger


class ProxyHatRotatingMiddleware:
    """
    Scrapy下载器中间件：ProxyHat住宅代理轮换
    
    支持三种模式：
    - per-request: 每个请求换IP（默认）
    - sticky: 粘滞会话，同一spider保持同一IP
    - geo-targeted: 按国家/城市定位
    """

    def __init__(self, username, password, country=None, mode='per-request'):
        self.username = username
        self.password = password
        self.country = country
        self.mode = mode
        self._sticky_session = None
        self._stats = {'total': 0, 'success': 0, 'banned': 0}

    @classmethod
    def from_crawler(cls, crawler):
        s = crawler.settings
        return cls(
            username=s.get('PROXYHAT_USERNAME', 'user'),
            password=s.get('PROXYHAT_PASSWORD', 'pass'),
            country=s.get('PROXYHAT_COUNTRY', None),
            mode=s.get('PROXYHAT_MODE', 'per-request'),
        )

    def _build_proxy_url(self, request):
        """根据模式构建ProxyHat代理URL"""
        user_parts = [self.username]

        # 地理定位
        if self.country:
            user_parts.append(f'country-{self.country}')

        # 城市级定位（从request.meta读取）
        city = request.meta.get('proxy_city')
        if city:
            user_parts.append(f'city-{city}')

        # 会话模式
        if self.mode == 'sticky':
            if self._sticky_session is None:
                self._sticky_session = ''.join(
                    random.choices(string.ascii_lowercase + string.digits, k=8)
                )
            user_parts.append(f'session-{self._sticky_session}')
        elif self.mode == 'per-request':
            # 每次请求生成新session，确保新IP
            session_id = f'{int(time.time())}-{random.randint(1000,9999)}'
            user_parts.append(f'session-{session_id}')

        username = '-'.join(user_parts)
        return f'http://{username}:{self.password}@gate.proxyhat.com:8080'

    def process_request(self, request, spider):
        proxy_url = self._build_proxy_url(request)
        request.meta['proxy'] = proxy_url
        request.meta['proxy_mode'] = self.mode
        self._stats['total'] += 1
        logger.debug(f"[ProxyHat] {self.mode} -> {request.url}")

    def process_response(self, request, response, spider):
        # 检测封禁信号
        if response.status in (403, 429, 503):
            self._stats['banned'] += 1
            # 粘滞会话被封时重置session
            if self.mode == 'sticky' and self._sticky_session:
                logger.warning(
                    f"[ProxyHat] Sticky session {self._sticky_session} banned, rotating"
                )
                self._sticky_session = None
            return request.dont_filter  # 允许重试中间件接管

        self._stats['success'] += 1
        return response

    def process_exception(self, request, exception, spider):
        logger.warning(f"[ProxyHat] Exception: {exception}, URL: {request.url}")
        return None  # 交给RetryMiddleware处理

在settings.py中注册中间件并配置参数：

# settings.py

# 禁用默认HttpProxyMiddleware，使用自定义中间件
DOWNLOADER_MIDDLEWARES = {
    'scrapy.downloadermiddlewares.httpproxy.HttpProxyMiddleware': None,
    'myproject.middlewares.ProxyHatRotatingMiddleware': 600,
}

# ProxyHat配置
PROXYHAT_USERNAME = 'your_username'
PROXYHAT_PASSWORD = 'your_password'
PROXYHAT_COUNTRY = 'US'        # 默认国家，None则全球随机
PROXYHAT_MODE = 'per-request'  # per-request | sticky

# 并发与延迟
CONCURRENT_REQUESTS = 32
DOWNLOAD_DELAY = 1
RANDOMIZE_DOWNLOAD_DELAY = True

在Spider中，你可以针对特定请求覆盖地理定位：

class PriceSpider(scrapy.Spider):
    name = 'price_spider'

    def start_requests(self):
        urls = [
            ('https://example.de/product/1', 'DE', 'berlin'),
            ('https://example.uk/product/2', 'GB', 'london'),
            ('https://example.com/product/3', 'US', None),
        ]
        for url, country, city in urls:
            yield scrapy.Request(
                url,
                callback=self.parse,
                meta={
                    'proxy_city': city,
                    'proxy_country': country,  # 可在中间件中读取
                },
            )

社区中间件对比：scrapy-rotating-proxies vs 自研

开源社区有几个Scrapy代理中间件，最知名的是scrapy-rotating-proxies。下表对比自研中间件与社区方案：

特性	scrapy-rotating-proxies	ProxyHat自研中间件
代理来源	自有代理列表文件	ProxyHat住宅池（自动管理）
IP轮换策略	随机/轮询	per-request / sticky / geo
地理定位	不支持	国家+城市级定位
会话粘滞	有限支持	内置session参数
封禁检测	基本（状态码）	状态码+响应内容模式
维护成本	需维护代理列表	零维护（池自动轮换）
代理质量	取决于列表质量	住宅IP，高匿名度

scrapy-rotating-proxies适合你有自建代理列表的场景。但如果你使用住宅代理服务（如ProxyHat），自研中间件更灵活——你不需要维护IP列表，轮换逻辑在服务端完成，中间件只需构建正确的认证URL。

核心原则：代理池管理交给服务端，中间件只负责请求注入和响应检测。这是住宅代理相比自建代理列表的根本优势。

代理失败重试：让RetryMiddleware与代理轮换协同

Scrapy内置的RetryMiddleware在请求失败时重试，但它不知道代理的存在——重试时可能使用同一个被封的IP。我们需要自定义重试中间件，在每次重试时轮换代理。

from scrapy.downloadermiddlewares.retry import RetryMiddleware
from scrapy.utils.response import response_status_message


class ProxyRetryMiddleware(RetryMiddleware):
    """
    重试中间件：每次重试时重置代理session，确保换IP
    """

    def __init__(self, settings):
        super().__init__(settings)
        self.max_retry_times = settings.getint('RETRY_TIMES', 3)

    def process_response(self, request, response, spider):
        # 封禁状态码直接触发重试
        ban_codes = spider.settings.getlist('BAN_STATUS_CODES', [403, 429, 503])
        
        if response.status in ban_codes:
            reason = response_status_message(response.status)
            # 重置sticky session，下次请求换IP
            self._rotate_proxy_session(request, spider)
            return self._retry(request, reason, spider) or response

        # 检测CAPTCHA页面
        if self._is_captcha_response(response):
            spider.logger.warning(f'[ProxyRetry] CAPTCHA detected: {request.url}')
            self._rotate_proxy_session(request, spider)
            return self._retry(request, 'captcha', spider) or response

        return response

    def process_exception(self, request, exception, spider):
        # 连接超时、代理错误等异常也触发重试+换IP
        if isinstance(exception, (
            self.exceptions.TimeoutError,
            self.exceptions.TCPTimedOutError,
            ConnectionRefusedError,
        )):
            self._rotate_proxy_session(request, spider)
            return self._retry(request, str(exception), spider)
        return None

    def _rotate_proxy_session(self, request, spider):
        """重置request.meta中的代理session标记"""
        # 标记需要新session，中间件下次process_request时会生成新session
        request.meta['force_new_session'] = True
        spider.logger.info(f'[ProxyRetry] Rotating proxy for: {request.url}')

    def _is_captcha_response(self, response):
        """简单CAPTCHA检测"""
        captcha_signals = ['captcha', 'recaptcha', 'cf-challenge', 'challenge-platform']
        body_lower = response.text.lower()[:5000]
        return any(sig in body_lower for sig in captcha_signals)

在settings.py中替换默认重试中间件：

DOWNLOADER_MIDDLEWARES = {
    'scrapy.downloadermiddlewares.retry.RetryMiddleware': None,
    'myproject.middlewares.ProxyRetryMiddleware': 500,
    'myproject.middlewares.ProxyHatRotatingMiddleware': 600,
}

RETRY_TIMES = 3
RETRY_HTTP_CODES = [403, 429, 500, 502, 503, 504]
BAN_STATUS_CODES = [403, 429, 503]

注意中间件优先级：ProxyRetryMiddleware (500)在ProxyHatRotatingMiddleware (600)之前执行。重试中间件设置force_new_session标记后，轮换中间件在process_request阶段读取该标记并生成新session。

JS渲染场景：scrapy-splash与scrapy-playwright的代理集成

许多目标网站依赖JavaScript渲染。Scrapy本身不执行JS，需要配合scrapy-splash或scrapy-playwright。两者都支持代理配置。

scrapy-splash代理配置

Splash是Docker化的无头浏览器服务。代理在Splash请求参数中传递：

class JSSpider(scrapy.Spider):
    name = 'js_spider'

    def start_requests(self):
        urls = ['https://spa-example.com/data']
        for url in urls:
            yield SplashRequest(
                url,
                callback=self.parse,
                endpoint='render.html',
                args={'wait': 3},
                proxy_url='http://user-country-US:pass@gate.proxyhat.com:8080',
                headers={'User-Agent': 'Mozilla/5.0'},
            )

    def parse(self, response):
        for item in response.css('.data-row'):
            yield {'value': item.css('::text').get()}

scrapy-playwright代理配置

scrapy-playwright直接在request.meta中设置代理，更贴近Scrapy原生风格：

# settings.py
DOWNLOAD_HANDLERS = {
    'http': 'scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler',
    'https': 'scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler',
}
TWISTED_REACTOR = 'twisted.internet.asyncioreactor.AsyncioSelectorReactor'

# Spider中
import scrapy


class PlaywrightSpider(scrapy.Spider):
    name = 'pw_spider'

    def start_requests(self):
        yield scrapy.Request(
            'https://spa-example.com/data',
            meta={
                'playwright': True,
                'playwright_include_page': False,
                'proxy_url': 'http://user-country-US:pass@gate.proxyhat.com:8080',
            },
        )

提示：scrapy-playwright的proxy_url是Playwright原生参数，支持SOCKS5。需要SOCKS5时使用socks5://user-country-US:pass@gate.proxyhat.com:1080。

部署方案：从Docker到Scrapyd

Docker + cron（最简方案）

对于定时运行的爬虫任务，Docker容器+cron是最直接的部署方式：

# Dockerfile
FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .

# cron任务定义
echo "0 */6 * * * cd /app && scrapy crawl price_spider >> /var/log/spider.log 2>&1" | crontab -
CMD ["cron", "-f"]

多Spider并发时，使用docker-compose定义多个服务，每个Spider一个容器，独立代理池：

# docker-compose.yml
services:
  spider-us:
    build: .
    environment:
      - PROXYHAT_COUNTRY=US
      - PROXYHAT_MODE=per-request
    command: scrapy crawl price_spider -a country=US

  spider-de:
    build: .
    environment:
      - PROXYHAT_COUNTRY=DE
      - PROXYHAT_MODE=sticky
    command: scrapy crawl price_spider -a country=DE

Scrapyd（集中管理）

Scrapyd是Scrapy官方的部署服务，支持通过API触发爬虫运行：

# 部署
curl http://scrapyd:6800/addversion.json -F project=myproject -F version=r1 -F egg=@dist/myproject-1.0.egg

# 运行
curl http://scrapyd:6800/schedule.json -d project=myproject -d spider=price_spider -d country=US

Scrapyd适合管理多个Spider的调度和版本管理，但不提供内置监控。你可以配合Scrapy监控面板方案使用。

ScrapeOps（托管方案）

ScrapeOps提供Scrapy的托管调度和监控，内置代理管理。适合不想自建基础设施的团队。但它会锁定你到特定平台——如果你的代理需求复杂（如城市级定位），自研中间件+ProxyHat的灵活度更高。

监控：每IP成功率与封禁检测

生产级爬虫必须有监控。以下是一个Scrapy扩展，追踪每个代理IP的成功率和封禁率：

from collections import defaultdict
from scrapy import signals
from scrapy.exceptions import NotConfigured


class ProxyStatsExtension:
    """
    追踪每个代理IP的成功/失败/封禁统计
    在spider关闭时输出报告
    """

    def __init__(self, stats):
        self.stats = stats
        self.ip_stats = defaultdict(lambda: {'success': 0, 'banned': 0, 'error': 0})

    @classmethod
    def from_crawler(cls, crawler):
        if not crawler.settings.getbool('PROXY_STATS_ENABLED'):
            raise NotConfigured
        ext = cls(crawler.stats)
        crawler.signals.connect(ext.spider_closed, signal=signals.spider_closed)
        crawler.signals.connect(ext.response_received, signal=signals.response_received)
        return ext

    def response_received(self, response, request, spider):
        ip = self._extract_ip(request)
        if not ip:
            return
        if response.status in (403, 429, 503):
            self.ip_stats[ip]['banned'] += 1
        elif response.status >= 400:
            self.ip_stats[ip]['error'] += 1
        else:
            self.ip_stats[ip]['success'] += 1

    def spider_closed(self, spider, reason):
        spider.logger.info('=' * 60)
        spider.logger.info('[ProxyStats] IP Performance Report')
        spider.logger.info('=' * 60)
        for ip, counts in sorted(self.ip_stats.items(),
                                  key=lambda x: x[1]['banned'], reverse=True):
            total = sum(counts.values())
            ban_rate = counts['banned'] / total * 100 if total else 0
            spider.logger.info(
                f'  {ip}: success={counts["success"]} '
                f'banned={counts["banned"]} '
                f'ban_rate={ban_rate:.1f}%'
            )
        # 推送到外部监控系统
        # requests.post('https://monitor.example.com/api/proxy_stats', json=dict(self.ip_stats))

    def _extract_ip(self, request):
        """从proxy URL提取标识（住宅代理通常不暴露真实IP，用session标识）"""
        proxy = request.meta.get('proxy', '')
        if 'session-' in proxy:
            # 提取session标识作为IP替代
            parts = proxy.split('session-')[-1].split(':')[0]
            return f'session:{parts}'
        return 'direct'

在settings.py中启用：

EXTENSIONS = {
    'myproject.extensions.ProxyStatsExtension': 500,
}
PROXY_STATS_ENABLED = True

封禁检测模式

除了状态码，还应该检测响应内容中的封禁信号：

• CAPTCHA页面：响应中包含recaptcha、cf-challenge等关键词
• 重定向陷阱：请求A页面却重定向到登录页
• 内容缩短：响应体大小明显小于正常页面
• 空数据页：页面正常渲染但数据为空

在ProxyRetryMiddleware中我们已经实现了CAPTCHA检测。你可以扩展_is_captcha_response方法来覆盖更多模式。

关键要点

• 禁用默认HttpProxyMiddleware，用自研中间件替代——这是Scrapy proxy rotation的基础
• 住宅代理轮换在用户名层完成，不需要维护IP列表；ProxyHat通过user-country-XX-session-YYYY格式实现地理定位和会话控制
• 重试必须换IP：自定义ProxyRetryMiddleware在每次重试时重置session，避免反复请求同一被封IP
• JS渲染场景：scrapy-splash通过proxy_url参数、scrapy-playwright通过meta['proxy_url']传递代理
• 部署首选Docker：每个Spider独立容器，环境变量控制代理参数，cron调度
• 监控不可少：追踪每IP成功率，封禁率超过阈值时自动告警

准备好在生产环境中使用Scrapy residential proxies了吗？访问ProxyHat定价页面选择适合你规模的代理套餐，或查看全球代理节点分布了解支持的地理定位范围。更多爬虫实战技巧，参考我们的Web Scraping用例指南。