Scrapy动态设置代理IP全流程！中间件接入代码一键复制

写 Scrapy 爬虫最怕什么？不是代码写不出来，而是跑了两百个请求，IP 就被封了。手动换 IP 不现实，写死一个代理又等于没换——这才是大多数爬虫开发者真正的痛点。

好消息是，Scrapy 的中间件机制天然适合做代理管理。你只需要写一个 ProxyMiddleware，就能实现：每个请求自动分配不同 IP、失败自动切换、优质 IP 优先使用。下面这篇文章，从原理到完整代码，一步步带你把动态代理 IP 接入 Scrapy，所有代码段都可以直接复制粘贴。

一、为什么 Scrapy 必须用动态代理 IP

先搞清楚”动态”两个字到底在动态什么。

如果你把代理 IP 写死在 settings.py 的 HTTP_PROXY 里，那所有请求都走同一个出口。目标站点的风控系统只要识别到短时间内大量请求来自同一 IP，立刻就会触发反爬——轻则返回 403，重则直接 ban 掉这个 IP 段。

动态代理 IP 的核心价值在于”出口地址持续变化”。每一次 HTTP 请求，都可以走一个不同的代理节点。对目标站点来说，你的爬虫看起来就是几百个不同的用户在正常浏览，风控规则根本无从下手。

具体来说，动态代理能帮你解决三个问题：

① 突破频率限制：很多站点限制单 IP 每分钟最多请求 N 次。轮换 IP 后，每个 IP 的请求量被摊薄，轻松绕过限制。

② 规避地域封锁：某些内容只允许特定地区 IP 访问。动态代理池里如果覆盖了多个地区节点，你就能灵活切换。

③ 提高采集稳定性：某个代理节点挂了或延迟飙升，中间件可以立刻剔除并换下一个，爬虫不会卡死。

光络云提供的代理 IP 服务覆盖国内和海外多个节点，支持按请求计费，配合 Scrapy 中间件使用，基本可以实现”无感换 IP”。需要注意的是，光络云的代理 IP 需要你的服务器本身具备海外网络环境才能正常调用（TikTok 专线产品除外），部署前请确认网络条件。

二、中间件核心原理：请求是怎么被”劫持”换 IP 的

Scrapy 的请求生命周期是这样的：Spider 产生一个 Request 对象 → 经过 Downloader Middleware 链 → 发出 HTTP 请求 → 拿到 Response → 再经过中间件链 → 回到 Spider。

代理 IP 的注入就发生在 Downloader Middleware 的 process_request 钩子里。这个钩子在请求真正发出去之前执行，你可以在这里往 request.meta 里塞代理地址，Scrapy 的下载器就会自动走这个代理。

整个链路可以简化为五步：

① 请求发出：Spider 的 parse() 方法里 yield Request(url)，此时还没有代理信息。

② 代理注入：你的 ProxyMiddleware.process_request() 被调用，从代理池里取出一个 IP，写入 request.meta['proxy']。

③ 目标站点：Scrapy 下载器带着代理地址发出 HTTP 请求，目标站点看到的是代理 IP 而非你的真实 IP。

④ 响应返回：正常拿到 200 响应，数据回到 Spider 解析。

⑤ 失败重试：如果返回 403、429 或连接超时，process_response 或 process_exception 钩子捕获异常，把当前 IP 标记为不可用，重新从池里取一个新 IP 发起重试。

理解了这条链路，写代码就只是”把每一步的逻辑填进去”。

三、完整代码：从代理池到中间件，一键复制

下面给出一个可以直接跑的完整实现，包含代理池管理、中间件、重试逻辑和 Scrapy 配置。代码按文件拆分，你只需要对应放到你的项目里。

3.1 项目结构

my_spider/
├── my_spider/
│   ├── __init__.py
│   ├── settings.py
│   ├── proxy_pool.py      # 代理池
│   ├── middlewares.py     # 代理中间件
│   └── spiders/
│       └── example.py
└── scrapy.cfg

3.2 代理池（proxy_pool.py）

这一层负责管理所有可用代理。实际生产中，代理列表通常从光络云的管理后台 API 拉取，这里用一个本地列表做演示，结构完全一致。

# proxy_pool.py
import random
import time
from collections import defaultdict

class ProxyPool:
    """
    代理池：管理可用 IP，支持随机取、失败剔除、定时刷新。
    生产环境可替换为从光络云 API 动态拉取代理列表。
    """

    def __init__(self, proxies: list[str] | None = None):
        # 格式: "http://user:pass@host:port"
        # 从光络云后台获取的代理地址直接填这里即可
        self._all: list[str] = proxies or []
        self._available: list[str] = list(self._all)
        self._fail_count: dict[str, int] = defaultdict(int)
        self._last_refresh: float = 0.0
        self.refresh_interval = 300  # 每 5 分钟重新拉取一次代理列表
        self.max_failures = 3        # 连续失败 3 次则暂时剔除

    def get_proxy(self) -> str | None:
        """随机取一个可用代理"""
        if not self._available:
            self._reset()
        return random.choice(self._available) if self._available else None

    def mark_failure(self, proxy: str):
        """标记一次失败，达到阈值则剔除"""
        self._fail_count[proxy] += 1
        if self._fail_count[proxy] >= self.max_failures:
            self._available.remove(proxy)
            print(f"[ProxyPool] 剔除失败代理: {proxy}")

    def mark_success(self, proxy: str):
        """成功后重置失败计数"""
        self._fail_count[proxy] = 0

    def _reset(self):
        """重置代理池（定时或池空时触发）"""
        self._available = list(self._all)
        self._fail_count.clear()
        self._last_refresh = time.time()
        print(f"[ProxyPool] 代理池已重置，共 {len(self._all)} 个")

    def refresh(self):
        """
        生产环境：调用光络云 API 拉取最新代理列表。
        示例（替换为实际 API 地址和鉴权）：
        
        import requests
        resp = requests.get(
            "https://api.ipipgo.com/proxies/list",
            headers={"Authorization": "Bearer YOUR_TOKEN"},
            params={"type": "residential", "country": "US"}
        )
        self._all = resp.json()["proxies"]
        self._reset()
        """
        self._reset()

3.3 代理中间件（middlewares.py）

这是核心文件，包含请求注入、响应处理、异常重试三个钩子。

# middlewares.py
import logging
from scrapy import signals
from scrapy.http import Request, Response
from scrapy.exceptions import IgnoreRequest
from .proxy_pool import ProxyPool

logger = logging.getLogger(__name__)

class DynamicProxyMiddleware:
    """
    Scrapy 动态代理 IP 中间件
    - 每个请求自动分配代理
    - 403/429/超时 自动换 IP 重试
    - 连续失败的 IP 自动剔除
    """

    MAX_RETRIES = 3  # 单个 URL 最多换 IP 重试 3 次

    def __init__(self, proxies: list[str]):
        self.pool = ProxyPool(proxies)

    @classmethod
    def from_crawler(cls, crawler):
        """Scrapy 标准中间件工厂方法，从 settings 读取配置"""
        proxies = crawler.settings.getlist("PROXY_LIST", [])
        middleware = cls(proxies)
        # 可选：注册定时刷新
        crawler.signals.connect(middleware._schedule_refresh, signal=signals.spider_opened)
        return middleware

    def process_request(self, request: Request, spider):
        """请求发出前：注入代理"""
        # 如果已经重试过，不再重复注入（避免死循环）
        if request.meta.get("proxy_retry_count", 0) >= self.MAX_RETRIES:
            logger.warning(f"URL 重试次数已达上限: {request.url}")
            return None  # 放行，不再换 IP

        proxy = self.pool.get_proxy()
        if proxy:
            request.meta["proxy"] = proxy
            request.meta["current_proxy"] = proxy
            # 带上重试计数
            request.meta["proxy_retry_count"] = request.meta.get("proxy_retry_count", 0) + 1
            logger.debug(f"使用代理 {proxy} 请求 {request.url}")
        else:
            logger.warning("代理池为空，走直连")

    def process_response(self, request: Request, response: Response, spider):
        """响应返回后：判断是否需要换 IP 重试"""
        proxy = request.meta.get("current_proxy")

        # 正常响应，标记成功
        if response.status == 200:
            if proxy:
                self.pool.mark_success(proxy)
            return response

        # 403 / 429 / 503 → 大概率是代理被识别，换 IP 重试
        if response.status in (403, 429, 503):
            if proxy:
                self.pool.mark_failure(proxy)
                logger.info(f"状态码 {response.status}，换 IP 重试: {request.url}")
            return self._retry(request, spider)

        # 其他状态码正常返回给 Spider 处理
        return response

    def process_exception(self, request: Request, exception, spider):
        """连接超时 / DNS 失败等异常：换 IP 重试"""
        proxy = request.meta.get("current_proxy")
        if proxy:
            self.pool.mark_failure(proxy)
            logger.info(f"异常 {type(exception).__name__}，换 IP 重试: {request.url}")
        return self._retry(request, spider)

    def _retry(self, request: Request, spider):
        """构造重试请求，重置代理"""
        if request.meta.get("proxy_retry_count", 0) >= self.MAX_RETRIES:
            logger.error(f"重试耗尽，放弃: {request.url}")
            return None  # 交给 Scrapy 的 RetryMiddleware 或 Spider 自行处理

        # 清除旧代理，让 process_request 重新分配
        del request.meta["proxy"]
        del request.meta["current_proxy"]
        return request  # 返回 Request 对象 → Scrapy 会重新走中间件链

    def _schedule_refresh(self, spider):
        """Spider 启动时刷新一次代理池"""
        self.pool.refresh()

3.4 Scrapy 配置（settings.py）

# settings.py

# ===== 代理 IP 配置 =====
# 从光络云后台获取的代理地址列表
# 格式: "http://user:pass@host:port"
PROXY_LIST = [
    "http://user:pass@1.2.3.4:8080",
    "http://user:pass@5.6.7.8:8080",
    "http://user:pass@9.10.11.12:8080",
    # ... 光络云代理列表
]

# 注册中间件（数字越小优先级越高）
DOWNLOADER_MIDDLEWARES = {
    "my_spider.middlewares.DynamicProxyMiddleware": 610,
    # 关闭 Scrapy 自带的重试，避免和我们的逻辑冲突
    "scrapy.downloadermiddlewares.retry.RetryMiddleware": None,
}

# ===== 请求行为 =====
CONCURRENT_REQUESTS = 10          # 并发数，配合代理池大小调整
CONCURRENT_REQUESTS_PER_DOMAIN = 5
DOWNLOAD_DELAY = 0.5              # 同域名请求间隔（秒）
DOWNLOAD_TIMEOUT = 15             # 超时时间（秒）

# 关闭自动代理（我们自己在中间件里管）
HTTPPROXY_ENABLED = False

# ===== 日志 =====
LOG_LEVEL = "INFO"

3.5 Spider 示例（spiders/example.py）

# spiders/example.py
import scrapy

class ExampleSpider(scrapy.Spider):
    name = "example"
    start_urls = ["https://example.com/page/1"]

    def parse(self, response):
        # 正常解析逻辑
        for item in response.css("div.item"):
            yield {
                "title": item.css("h2::text").get(),
                "link":  item.css("a::attr(href)").get(),
            }

        # 翻页
        next_page = response.css("a.next::attr(href)").get()
        if next_page:
            yield scrapy.Request(next_page, callback=self.parse)
            # 注意：不需要手动设 proxy，中间件会自动处理

把以上四个文件放进你的 Scrapy 项目，scrapy crawl example 就能跑起来。每个请求都会自动走代理池里的不同 IP，遇到 403 或超时会自动换 IP 重试，连续失败的 IP 会被暂时剔除。

四、动态切换策略：不止”随机”一种玩法

上面的代码用的是随机轮换——每次请求从池子里随机挑一个。这是最基础也最通用的策略，但根据实际场景，你还可以叠加更多规则：

以失败熔断 + 冷却恢复为例，在 ProxyPool 里加一段逻辑：

import time

class ProxyPool:
    # ... 前面代码不变 ...

    def __init__(self, proxies=None):
        # ...
        self._cooldown: dict[str, float] = {}  # proxy → 冷却截止时间
        self.cooldown_seconds = 300  # 冷却 5 分钟

    def mark_failure(self, proxy: str):
        self._fail_count[proxy] += 1
        if self._fail_count[proxy] >= self.max_failures:
            self._available.remove(proxy)
            self._cooldown[proxy] = time.time() + self.cooldown_seconds
            print(f"[Pool] 熔断 {proxy}，冷却 {self.cooldown_seconds}s")

    def get_proxy(self) -> str | None:
        # 先检查冷却期是否结束，恢复可用 IP
        now = time.time()
        recovered = [p for p, t in self._cooldown.items() if now >= t]
        for p in recovered:
            if p not in self._available:
                self._available.append(p)
            del self._cooldown[p]
            self._fail_count[p] = 0
            print(f"[Pool] 恢复 {p}")

        if not self._available:
            self._reset()
        return random.choice(self._available) if self._available else None

这样，一个”坏掉”的 IP 不会永远被丢弃，而是冷却一段时间后自动回到池子里重新参与轮换，代理利用率更高。

五、接入光络云代理的实操要点

上面代码里的 PROXY_LIST 是硬编码的，实际生产环境建议通过 API 动态拉取。光络云管理后台提供了代理列表接口，你只需要把 ProxyPool.refresh() 方法里的示例代码替换成实际调用即可。

几个实操建议：

① 代理池大小要匹配并发数。如果你 CONCURRENT_REQUESTS = 50，但池子里只有 5 个 IP，那每个 IP 同时扛 10 个请求，很容易被风控。建议代理数量 ≥ 并发数的 2-3 倍。

② 注意网络环境。光络云的代理 IP 服务需要你的服务器具备海外网络环境才能正常调用（TikTok 专线除外）。如果你的爬虫跑在国内服务器上，需要先确认出口网络条件，否则代理握手会超时。

③ 日志一定要开。调试阶段把 LOG_LEVEL 设为 "DEBUG"，中间件里每条 logger.debug 都会打印当前用了哪个代理、目标 URL 是什么。排查”到底哪个 IP 被 ban 了”全靠它。

④ 别把重试逻辑和 Scrapy 自带的 RetryMiddleware 混用。上面 settings 里已经把 RetryMiddleware 设为 None，如果你同时启用两个重试机制，会出现”重试套重试”的情况，请求量翻倍但成功率不升反降。

常见问题

Q: 我的代理 IP 是光络云提供的，但 Scrapy 连不上，一直超时，怎么回事？
大概率是网络环境问题。光络云的代理 IP 需要你的服务器具备海外网络出口才能正常握手。请检查你的服务器是否能直接访问海外节点（curl -I https://api.ipipgo.com 测试）。TikTok 专线产品不受此限制，可直接使用。

Q: 代理池里只有 10 个 IP，能支撑多大的并发？
建议并发数不超过代理数量的 1/2 到 1/3。10 个 IP 的话，CONCURRENT_REQUESTS 设 3-5 比较安全。如果目标站点风控严格，甚至建议并发 1-2 + 每次请求换 IP 的方式，用时间换稳定。

Q: 中间件里 process_request 返回 None 和返回 Request 有什么区别？
返回 None 表示”我不处理，交给下一个中间件或下载器”，请求原样放行。返回一个新的 Request 对象表示”我替换了请求”，Scrapy 会用你返回的新请求继续走后续中间件链。在重试场景里，我们返回的是修改过的原 request 对象（清除了旧代理），让 process_request 重新分配。

Q: 怎么判断当前代理是不是被目标站点 ban 了，而不是网络本身的问题？
在 process_response 里看状态码：403、429 基本可以确定是代理被识别；连接超时（TimeoutError）可能是代理节点本身挂了，也可能是网络抖动。可以在中间件里加一个计数器，如果同一个代理连续 3 次超时，大概率是节点故障而非目标站点封禁。

Q: 光络云的代理 IP 支持按地区筛选吗？
支持。光络云管理后台和 API 都提供了按国家/地区筛选代理节点的参数。如果你的目标站点只允许特定地区 IP 访问，在 PROXY_LIST 拉取时加上地区参数即可，中间件代码不需要任何改动。

Q: 这套中间件能同时用于 Scrapy 和 requests 吗？
中间件代码是 Scrapy 专属的（依赖 process_request 等钩子）。如果你用 requests 库，逻辑类似但写法不同——直接在 session.get(url, proxies={"http": proxy}) 里传代理即可，不需要中间件。核心思路（代理池 + 失败剔除 + 重试）是通用的。