分布式爬虫代理IP调度：Scrapy-Redis集群里代理该怎么管

单机Scrapy的代理方案，放到集群里哪里会出问题

单机Scrapy挂代理最常见的做法是写一个DownloaderMiddleware，在process_request里给每个请求分配一个代理：

```python

classProxyMiddleware:
def__init__(self):
self.proxies=load_proxies()#从文件或接口拿一批IP
self.index=0

defprocess_request(self,request,spider):
request.meta['proxy']=self.proxies[self.index%len(self.proxies)]

self.index+=1

```

单机跑没问题。但到了Scrapy-Redis分布式环境，你有5台、10台甚至更多Worker，每台都跑一个Scrapy进程，共享Redis里的请求队列。这时候上面这段代码的问题就来了：

每个Worker各自维护自己的代理列表。如果代理列表是从同一个接口拿的，那所有Worker拿到的是同一批IP——同一个IP可能同时被5个Worker用在同一个目标站上，目标站看到的并发量是你预期的5倍。

IP状态不共享。WorkerA发现某个IP已经失效了，把它从自己的列表里删掉。但WorkerB、C、D还在用，继续产生失败请求。

没有全局的使用频率控制。你以为每个Worker每分钟用某个IP发10个请求，合起来就是50个——但目标站的访问频率控制是按IP计的，不是按你的Worker计的。

这三个问题的根源是一样的：代理状态没有中心化。解决方案也很直接：把代理池从Worker本地搬到Redis里，所有Worker共享同一个池。

架构设计：Redis做代理池中枢，Worker只做"借"和"还"

整体架构分三层：

Redis代理池层。所有代理IP存在Redis里，包括IP地址、评分、上次使用时间、当前借出状态。用SortedSet做调度（Score是评分），用Hash存元数据，用Set存黑名单——和单机代理池的Redis数据结构设计思路一致，但要额外加一个字段：last_used_by，记录是哪个Worker在用。

代理调度服务层（可选但推荐）。在Redis和Worker之间加一个轻量的调度服务（可以是一个独立的Python进程或Go服务），负责：分配IP给Worker、接收Worker的使用反馈（成功/失败/延迟）、更新IP评分、触发健康检测。如果不想加独立服务，也可以让Worker直接操作Redis，但要注意并发安全。

Worker层。每个ScrapyWorker的Middleware不再自己维护代理列表，而是每次需要代理时向Redis（或调度服务）"借"一个IP，用完后"还"回去并附上使用结果。

这个"借-用-还"模型是整个方案的核心。下面详细讲怎么实现。

Redis数据结构：在单机代理池基础上加"借出锁"

在单机代理池的SortedSet+Hash+Set基础上，集群环境需要额外处理一个问题：同一个IP不能同时被借给两个Worker用在同一个目标站上。

方案是给每个IP加一个"借出锁"，用Redis的SETkeyvalueEXsecondsNX实现：

Key:proxy:lock:{ip:port}:{domain}
Value:worker_id
EX:30(秒，超时自动释放)
NX:只在key不存在时设置

Worker借IP时，先从SortedSet里取一个高分IP，再尝试对这个IP+目标域名加锁。加锁成功才算借到，加锁失败（说明别的Worker正在用这个IP访问同一个域名）就换一个IP重试。

为什么锁的粒度是ip+domain而不是只锁ip？因为同一个IP同时用在不同目标站上是可以的——目标站A的访问频率控制不会因为你同时在访问目标站B而触发。只锁IP会浪费代理资源。

锁的超时设30秒是个经验值。正常请求5-10秒就完成了，30秒兜底是防止Worker崩溃后锁不释放。如果你的采集任务有大文件下载（耗时可能超过30秒），需要调大这个值。

ScrapyMiddleware改造：从本地轮询到Redis借还

改造后的Middleware核心逻辑：

```python

importredis
importtime
importhashlib

classDistributedProxyMiddleware:
def__init__(self,redis_url,worker_id):
self.rdb=redis.from_url(redis_url)
self.worker_id=worker_id
self.pool_key="proxy:pool:default"
self.lock_ttl=30

@classmethod
deffrom_crawler(cls,crawler):
returncls(
redis_url=crawler.settings.get('REDIS_URL'),
worker_id=crawler.settings.get('WORKER_ID',f'worker-{id(crawler)}')
)

defprocess_request(self,request,spider):
domain=self._extract_domain(request.url)
proxy=self._borrow_proxy(domain,max_retries=10)
ifnotproxy:
raiseException("Noavailableproxy")
request.meta['proxy']=f"http://{proxy}"
request.meta['_proxy_addr']=proxy
request.meta['_proxy_domain']=domain

defprocess_response(self,request,response,spider):
proxy=request.meta.get('_proxy_addr')
domain=request.meta.get('_proxy_domain')
ifproxy:
ifresponse.statusin(200,301,302):
self._return_proxy(proxy,domain,success=True)
elifresponse.statusin(403,429):
#访问频率控制类状态码，降低评分但不直接淘汰
self._return_proxy(proxy,domain,success=False,soft=True)
else:
self._return_proxy(proxy,domain,success=False)
returnresponse

defprocess_exception(self,request,exception,spider):
proxy=request.meta.get('_proxy_addr')
domain=request.meta.get('_proxy_domain')
ifproxy:
self._return_proxy(proxy,domain,success=False)

def_borrow_proxy(self,domain,max_retries=10):
"""从Redis池里借一个IP，加锁成功才算借到"""
#取评分最高的一批候选
candidates=self.rdb.zrevrange(self.pool_key,0,49,withscores=True)
foraddr,scoreincandidates:
addr=addr.decode()ifisinstance(addr,bytes)elseaddr
lock_key=f"proxy:lock:{addr}:{domain}"
#尝试加锁
ifself.rdb.set(lock_key,self.worker_id,ex=self.lock_ttl,nx=True):
returnaddr
returnNone

def_return_proxy(self,addr,domain,success=True,soft=False):
"""归还IP，释放锁，更新评分"""
lock_key=f"proxy:lock:{addr}:{domain}"
self.rdb.delete(lock_key)

ifsuccess:
self.rdb.zincrby(self.pool_key,2,addr)
elifsoft:
self.rdb.zincrby(self.pool_key,-5,addr)
else:
self.rdb.zincrby(self.pool_key,-15,addr)
#评分过低则移除
score=self.rdb.zscore(self.pool_key,addr)
ifscoreisnotNoneandscore<10:
self.rdb.zrem(self.pool_key,addr)
self.rdb.sadd("proxy:blacklist",addr)

def_extract_domain(self,url):
fromurllib.parseimporturlparse
returnurlparse(url).netloc

```

几个要点：

_borrow_proxy从前50个高分IP里找第一个能加锁成功的。为什么不是从全量IP里找？因为低分IP大概率质量差，先试高分的命中率更高。50这个数字可以根据池子规模和Worker数量调——Worker多的时候需要加大候选范围，否则大家抢同一批高分IP导致锁竞争激烈。

_return_proxy区分了三种情况：成功加2分，访问频率控制类失败减5分（软惩罚），代理层失败减15分（硬惩罚）。这个区分很重要——如果不分，403/429会导致大量IP被误杀。

process_exception也要处理。连接超时、代理拒绝连接这类异常走process_exception而不是process_response，如果不在这里归还IP和释放锁，锁会一直占着直到超时。

WorkerID怎么分配

每个Worker需要一个唯一的WORKER_ID，用于锁的value和日志追踪。几种常见做法：

用主机名+进程PID：socket.gethostname()+"-"+str(os.getpid())，简单直接，但同一台机器上跑多个Scrapy进程时要注意PID可能重复（容器环境里PID通常是1）。

用环境变量：部署时给每个Worker注入一个唯一ID，比如WORKER_ID=worker-01。适合容器编排（DockerCompose、K8s）。

用UUID：str(uuid.uuid4())，保证唯一但不方便人工排查。

推荐用主机名+进程启动时间戳的组合，兼顾唯一性和可读性。

健康检测放在哪里跑

在分布式环境里，健康检测不应该放在每个Worker里跑。原因：

如果5个Worker各自跑一遍健康检测，同一个IP被检测5次，浪费资源。而且多个Worker同时更新同一个IP的评分，会产生竞态条件。

更好的做法是把健康检测抽成一个独立进程，和Worker分开部署。它只做一件事：每隔N分钟扫描Redis里的所有IP，逐个检测，更新评分，淘汰坏IP。这个进程全局只跑一个（或者加分布式锁保证同一时间只有一个实例在跑）。

```python

#health_checker.py——独立进程，不是Scrapy的一部分
importredis
importrequests
importtime

defrun_health_check(rdb,pool_key,check_url="zllp.myyzllp/=s_kwcy=p&nikl):
members=rdb.zrangebyscore(pool_key,'-inf','+inf',withscores=True)
foraddr,scoreinmembers:
addr=addr.decode()ifisinstance(addr,bytes)elseaddr
try:
resp=requests.get(
check_url,
proxies={"http":f"http://{addr}","https":f"http://{addr}"},
timeout=10
)
ifresp.status_code==200:
rdb.zincrby(pool_key,3,addr)
else:
rdb.zincrby(pool_key,-10,addr)
exceptException:
new_score=score*0.7
ifnew_score<10:
rdb.zrem(pool_key,addr)
rdb.sadd("proxy:blacklist",addr)
else:
rdb.zadd(pool_key,{addr:new_score})

if__name__=="__main__":
rdb=redis.from_url("redis://localhost:6379/0")
whileTrue:
run_health_check(rdb,"proxy:pool:default")