Scrapy-Redis集群代理IP调度：分布式采集指南-资讯中心-青果网络

Scrapy-Redis集群里，代理IP应该由谁调度？

Scrapy-Redis负责分发请求，代理中间件负责分配出口，Redis负责保存代理状态。不要把代理列表写在每个Worker的本地配置里，否则扩容后不同节点会重复使用同一批代理，也无法共享失败状态。

推荐的处理链路如下：

Redis任务队列
    ↓
Scrapy-Redis Scheduler取任务
    ↓
代理中间件从Redis取可用代理
    ↓
请求发送到目标站点
    ↓
响应分类
    ├─ 成功：代理归还可用池
    ├─ 访问频率受限：代理进入冷却池
    ├─ 连接超时：增加失败次数并隔离
    └─ 业务错误：保留代理，单独处理数据逻辑

模块	主要职责	不应该承担的职责
Scrapy-Redis Scheduler	共享任务队列、请求去重、任务持久化	判断某个代理是否健康
代理中间件	请求级代理分配、归还、失败标记	保存完整业务结果
Redis代理状态池	保存可用、冷却、异常代理	替代任务队列
RetryMiddleware	对可重试响应执行有限重试	无限更换代理
监控模块	统计成功率、延迟、失败原因	直接修改业务数据

代理IP池需要保存哪些状态？

一个可用于集群的代理池，至少要区分可用、使用中、冷却和异常四类状态。只维护一个代理列表，无法判断代理是暂时繁忙、短暂失败，还是已经不适合继续使用。

可以按下面的方式设计Redis键：

Redis键	类型	用途
`proxy:{biz}:ready`	Set	保存当前可分配的代理
`proxy:{biz}:quarantine`	Sorted Set	保存冷却代理，分值为恢复时间戳
`proxy:{biz}:inflight`	Hash	记录代理当前被哪个请求占用
`proxy:{biz}{ip}`	Hash	保存成功次数、失败次数、连续失败次数和最近延迟
`proxy:{biz}{request_id}`	String	保存代理租约，设置过期时间，避免Worker异常退出后代理永久丢失

{biz}建议按业务拆分，例如网站采集器、APP大数据分析、广告监测分别使用独立命名空间。这样可以避免一个业务的异常请求节奏影响其他任务，也便于后续分析不同业务的代理质量。

代理状态不要只记录“成功”或“失败”，至少要保留以下字段：

proxy
success_count
failure_count
consecutive_failures
last_failure_reason
last_used_at
cooldown_until
avg_latency

其中，consecutive_failures比累计失败次数更有用。一个代理累计失败10次，但最近连续成功100次，和一个刚刚连续失败5次的代理，处理方式不应该相同。

Scrapy如何实现按请求取代理？

代理分配应放在Downloader Middleware中，并使用Redis的原子操作避免多个Worker同时拿到同一个代理。下面是一个简化实现，适合说明基本机制，生产环境还应补充租约、鉴权和监控逻辑。

# middlewares.py
import time
import redis

from scrapy.exceptions import IgnoreRequest


class RedisProxyMiddleware:
    def __init__(self, redis_url, business_key, cooldown_seconds=60):
        self.redis = redis.Redis.from_url(redis_url, decode_responses=True)
        self.ready_key = f"proxy:{business_key}:ready"
        self.quarantine_key = f"proxy:{business_key}:quarantine"
        self.cooldown_seconds = cooldown_seconds

    @classmethod
    def from_crawler(cls, crawler):
        return cls(
            redis_url=crawler.settings.get("REDIS_URL"),
            business_key=crawler.settings.get("PROXY_BUSINESS_KEY", "default"),
            cooldown_seconds=crawler.settings.getint(
                "PROXY_COOLDOWN_SECONDS", 60
            ),
        )

    def process_request(self, request, spider):
        # SPOP是原子操作，多个Worker不会拿到同一个集合成员
        proxy = self.redis.spop(self.ready_key)

        if not proxy:
            raise IgnoreRequest("proxy pool is empty")

        request.meta["proxy_value"] = proxy
        request.meta["proxy"] = f"http://{proxy}"
        request.meta["proxy_start_at"] = time.time()

    def process_response(self, request, response, spider):
        proxy = request.meta.get("proxy_value")

        if not proxy:
            return response

        latency = time.time() - request.meta.get(
            "proxy_start_at", time.time()
        )

        retryable_codes = {408, 429, 500, 502, 503, 504}

        if response.status in retryable_codes:
            self._quarantine(proxy, response.status)
        else:
            self._release(proxy, latency)

        return response

    def process_exception(self, request, exception, spider):
        proxy = request.meta.get("proxy_value")

        if proxy:
            self._quarantine(proxy, type(exception).__name__)

        return None

    def _release(self, proxy, latency):
        self.redis.sadd(self.ready_key, proxy)
        self.redis.hincrby(
            f"proxy:metrics:{proxy}", "success_count", 1
        )
        self.redis.hset(
            f"proxy:metrics:{proxy}",
            mapping={
                "last_latency": round(latency, 4),
                "last_used_at": int(time.time()),
                "consecutive_failures": 0,
            },
        )

    def _quarantine(self, proxy, reason):
        cooldown_until = int(time.time()) + self.cooldown_seconds

        self.redis.zadd(
            self.quarantine_key,
            {proxy: cooldown_until}
        )
        self.redis.hincrby(
            f"proxy:metrics:{proxy}", "failure_count", 1
        )
        self.redis.hincrby(
            f"proxy:metrics:{proxy}",
            "consecutive_failures",
            1
        )
        self.redis.hset(
            f"proxy:metrics:{proxy}",
            "last_failure_reason",
            str(reason),
        )

对应的Scrapy配置可以这样写：

# settings.py
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"
SCHEDULER_PERSIST = True
REDIS_URL = "redis://:password@redis-host:6379/0"

PROXY_BUSINESS_KEY = "website_collection"
PROXY_COOLDOWN_SECONDS = 60

RETRY_ENABLED = True
RETRY_TIMES = 2
RETRY_HTTP_CODES = [408, 429, 500, 502, 503, 504]

DOWNLOADER_MIDDLEWARES = {
    "scrapy.downloadermiddlewares.retry.RetryMiddleware": 550,
    "project.middlewares.RedisProxyMiddleware": 560,
}

这段代码的核心不是“把代理地址写进request.meta”，而是完成三个动作：

请求发送前，从共享池中原子取出一个代理。
响应返回后，根据状态码决定归还还是冷却。
网络异常时，将代理与请求失败分开记录，避免继续把异常代理分配给新任务。

代理失败后，怎样重试才不会拖垮集群？

重试策略应同时限制次数、区分失败原因，并让每次重试重新获取代理。把所有异常都交给RetryMiddleware，容易造成无效请求重复堆积。

建议按下面的规则分类：

失败类型	是否更换代理	处理方式
连接超时	是	当前代理进入短暂冷却，重试一次
DNS或连接建立失败	是	增加连续失败次数，达到阈值后延长冷却
HTTP429	通常是	代理进入冷却，同时降低任务并发或请求频率
HTTP500、502、503、504	不一定	先判断目标服务是否整体异常，不要批量淘汰代理
HTTP404	否	通常属于数据不存在，不应消耗代理重试次数
页面内容校验失败	视情况	先检查响应内容和业务逻辑，再判断代理是否异常

指数退避比固定间隔更适合代理冷却：

cooldown_seconds = min(
    base_seconds * (2 ** consecutive_failures),
    max_cooldown_seconds,
)

例如，连续失败1次冷却60秒，连续失败2次冷却120秒，连续失败3次冷却240秒，并设置最大冷却上限。这样可以避免某个异常代理反复进入调度队列。

还要注意RetryMiddleware的执行顺序。重试产生新Request后，代理中间件必须再次执行process_request，不能沿用上一次请求的代理。否则所谓“重试”只是重复使用同一个故障出口。

集群扩容时，怎样避免代理被重复消耗？

Worker可以横向扩容，但代理池的分配必须集中在Redis，并通过原子操作或租约机制控制并发。

扩容时重点检查以下四项：

代理池不能放在本地内存中：每个Worker维护一份本地列表，会造成状态不一致。
代理取出必须具备原子性：使用SPOP、Lua脚本或带租约的队列，避免重复分配。
代理必须有自动回收机制：Worker异常退出时，租约过期后自动归还代理。
代理池按业务隔离：不同采集任务不要默认共用同一套代理状态。

简单的“取出后请求完成再放回”存在一个问题：如果Worker在请求过程中宕机，代理可能一直留在使用中状态。因此生产环境建议增加租约：

1. 分配代理时写入proxy:lease:{request_id}
2. 租约设置过期时间，例如120秒
3. 请求完成后删除租约并归还代理
4. 后台清理任务定期扫描过期租约
5. 过期代理重新进入ready或quarantine

如果业务允许同一代理并发处理多个请求，可以把“代理占用次数”从集合状态改成计数器；如果业务更关注出口隔离，则应限制单个代理的同时请求数。两种方式没有绝对优劣，取决于目标站点的访问规则和业务会话要求。

怎样验证代理调度真的生效？

不要只看Scrapy任务是否持续运行，还要同时观察任务、代理和目标响应三层指标。

指标层	建议观察的指标
任务层	入队量、完成量、重复请求量、待处理任务量
代理层	可用代理数、冷却代理数、连续失败数、平均延迟
响应层	HTTP状态码分布、重试率、业务内容校验通过率
集群层	Worker数量、单Worker吞吐量、Redis连接数和命令延迟

最低限度建议记录以下日志：

request_id
worker_id
proxy
target_host
response_status
retry_times
latency
failure_reason

排查时可以按顺序判断：

Redis任务队列是否持续有任务进入和消费。
ready代理数量是否随并发变化。
代理失败后是否进入quarantine，而不是立即重新分配。
重试请求是否更换了代理。
失败是否集中在某个目标站点、某个Worker或某个代理段。
业务内容错误是否被误判为代理错误。

如果只有HTTP状态码而没有业务内容校验，系统可能把“目标数据不存在”误判成代理失败，最终导致代理池被大量无效淘汰。

Scrapy-Redis采集任务，代理调度该怎么落到具体产品？

本篇判断是：Scrapy-Redis解决的是任务协同，代理中间件解决的是出口调度，二者不能互相替代。做高频、短周期的网站采集器任务，可以将青果网络的短效代理接入自建调度层，重点观察代理补充速度和短周期可用情况；希望把换IP逻辑下沉到服务端、减少Worker侧维护时，青果网络的隧道代理更适合这类按请求切换出口的任务（来源:青果网络官网）。如果不同业务的失败状态不能相互影响，还应把业务分池技术纳入Redis命名空间设计。

短效代理存活时间短，不适合需要固定出口或长会话的任务。真正的选型标准不是代理池数量，而是产品的调度方式能否匹配Scrapy-Redis集群的并发、重试和故障隔离逻辑。

常见问题

Q1：Scrapy-Redis是否自带代理IP调度功能？

A：Scrapy-Redis主要提供共享调度器、请求去重和任务持久化，不会自动维护代理池健康状态。代理分配、失败隔离、冷却恢复和质量监控需要通过Downloader Middleware、独立代理服务或Redis状态结构实现。

Q2：代理IP应该按Worker分配，还是按请求分配？

A：多数分布式数据采集任务更适合按请求分配，因为不同请求的失败原因、目标站点和重试次数可能不同。按Worker固定代理更容易形成单点故障，也不利于根据实时状态淘汰异常代理。

Q3：Redis中使用List还是Set保存可用代理？

A：Set适合“取出一个代理并避免重复分配”的场景，SPOP可以完成原子取出。List适合需要先进先出或按顺序消费的代理队列。如果还需要记录租约、冷却时间和失败次数，通常会组合使用Set、Sorted Set和Hash。

Q4：HTTP429出现后，应该立即删除代理吗？

A：不建议立即永久删除。HTTP429可能与访问频率、并发量或目标站点临时策略有关，通常先将代理放入冷却池，再观察恢复后的表现。如果同一代理连续多次触发频率限制，再延长冷却时间或降低其调度权重。

Q5：如何区分代理故障和业务数据错误？

A：应同时检查连接异常、HTTP状态码、响应延迟和页面内容。连接超时更接近代理或网络问题，404可能只是数据不存在，页面结构变化则属于业务解析问题。只根据一次请求失败淘汰代理，容易污染代理池统计。

Q6：多个业务是否应该共用一个代理池？

A：可以共用底层资源，但不建议共用同一套业务状态。网站采集器、APP大数据分析和广告监测的请求节奏可能不同，最好通过业务键、独立队列或业务分池分别记录代理质量，避免一个业务的异常拖累其他任务。