Scrapling 爬虫代理管理与封锁处理：ProxyRotator 轮转与 Blocked 重试机制实战-迪斯科星球

Scrapling 爬虫代理管理与封锁处理：ProxyRotator 轮转与 Blocked 重试机制实战

【免费下载链接】Scrapling🕷️ An adaptive Web Scraping framework that handles everything from a single request to a full-scale crawl!项目地址: https://gitcode.com/GitHub_Trending/sc/Scrapling

本文基于 Scrapling 官方技能文档 proxy-blocking.md 展开，系统讲解ProxyRotator代理轮转器（字符串/字典双格式、自定义轮转策略、按请求覆盖代理）以及 Spider 引擎内建的被封锁请求检测与重试流程（is_blocked/retry_blocked_request/max_blocked_retries）。读完后你能够：为任意类型会话配置代理池、追踪每次请求实际使用的代理、按“廉价代理先用、被封锁后自动换昂贵代理”的模式组织多会话降级策略，并理解引擎重试时的队列细节（dont_filter、优先级下调、代理参数自动清空）。

一、整体架构：两层机制如何协同

Scrapling 的 Spider 体系（scrapling/spiders/ 目录）将“代理问题”拆成两个正交层面：

会话层（Session 层）：ProxyRotator负责在每次请求前分配代理。它工作在 FetcherSession、AsyncDynamicSession / AsyncStealthySession 等所有会话类型内部，并在代理连接失败（如net::err_proxy、connection refused）时触发会话自身的重试，且重试时自动换一个新代理。
蜘蛛层（Spider 层）：引擎在收到响应后调用is_blocked(response)判断是否被目标站点封锁（403/429 等状态码或自定义特征），被封锁的请求会被复制、调整并重新入队，最多重试max_blocked_retries次（默认 3 次）。

两层共用同一个ProxyRotator实例，但计数器互相独立——这正是文档中特别提示“max_blocked_retries与会话 retries 不共享计数器”的原因。下文逐层拆解。

二、ProxyRotator 基础用法

ProxyRotator管理一个代理列表并自动轮转。将它通过proxy_rotator参数传给任意会话类型即可（完整实现见 proxy_rotation.py）：

from scrapling.spiders import Spider, Response from scrapling.fetchers import FetcherSession, ProxyRotator class MySpider(Spider): name = "my_spider" start_urls = ["https://example.com"] def configure_sessions(self, manager): rotator = ProxyRotator([ "http://proxy1:8080", "http://proxy2:8080", "http://user:pass@proxy3:8080", ]) manager.add("default", FetcherSession(proxy_rotator=rotator)) async def parse(self, response: Response): # Check which proxy was used print(f"Proxy used: {response.meta.get('proxy')}") yield {"title": response.css("title::text").get("")}

每个请求自动拿到轮转列表中的下一个代理。实际使用的代理会写入response.meta["proxy"]，方便追踪“哪个代理抓到了哪个页面”——这一点在源码中可以印证：静态请求的响应工厂在创建Response时显式传入meta={"proxy": proxy}（见 static.py），浏览器会话同理（见 _controllers.py）。

代理的两种格式

字符串代理对所有会话类型都可用；浏览器会话额外支持 Playwright 风格的字典代理：

from scrapling.fetchers import AsyncDynamicSession, AsyncStealthySession, ProxyRotator # String proxies work for all session types rotator = ProxyRotator([ "http://proxy1:8080", "http://proxy2:8080", ]) # Dict proxies (Playwright format) work for browser sessions rotator = ProxyRotator([ {"server": "http://proxy1:8080", "username": "user", "password": "pass"}, {"server": "http://proxy2:8080"}, ]) # Then inside the spider def configure_sessions(self, manager): rotator = ProxyRotator(["http://proxy1:8080", "http://proxy2:8080"]) manager.add("browser", AsyncStealthySession(proxy_rotator=rotator))

从源码看，ProxyRotator.__init__对每个元素做严格校验：

空列表抛出ValueError("At least one proxy must be provided")；
字典代理缺少server键抛出ValueError("Proxy dict must have a 'server' key")；
既不是str也不是dict的元素抛出TypeError；
strategy参数必须可调用，否则抛出TypeError（见 proxy_rotation.py）。

这些行为均有测试覆盖，见 tests/fetchers/test_proxy_rotation.py 中的TestProxyRotatorCreation用例。

默认轮转与线程安全

默认策略是cyclic_rotation（循环轮转）：按顺序取代理，到末尾后回绕：

def cyclic_rotation(proxies: List[ProxyType], current_index: int) -> Tuple[ProxyType, int]: idx = current_index % len(proxies) return proxies[idx], (idx + 1) % len(proxies)

（见 proxy_rotation.py，idx % len还保证了索引越界时自动回绕。）get_proxy()内部用threading.Lock保护索引更新，因此并发取代理是安全的——test_proxy_rotation.py 中用 10 个线程 × 100 次取代理、以及ThreadPoolExecutor并发的用例验证了这一点。类使用__slots__定义，并对外提供proxies属性（返回副本，防止外部篡改）、__len__与__repr__（输出形如ProxyRotator(proxies=3)）。

三、自定义轮转策略

默认是顺序循环轮转，你也可以传入自定义策略函数，但签名必须匹配：

from scrapling.core._types import ProxyType def my_strategy(proxies: list, current_index: int) -> tuple[ProxyType, int]: ...

函数接收代理列表和当前索引，返回“选中的代理 + 下一个索引”。

随机轮转

import random from scrapling.fetchers import ProxyRotator def random_strategy(proxies, current_index): idx = random.randint(0, len(proxies) - 1) return proxies[idx], idx rotator = ProxyRotator( ["http://proxy1:8080", "http://proxy2:8080", "http://proxy3:8080"], strategy=random_strategy, )

加权轮转

import random def weighted_strategy(proxies, current_index): # First proxy gets 60% of traffic, others split the rest weights = [60] + [40 // (len(proxies) - 1)] * (len(proxies) - 1) proxy = random.choices(proxies, weights=weights, k=1)[0] return proxy, current_index # Index doesn't matter for weighted

对于纯随机的策略，返回的“下一个索引”可以是任意值（如上例直接回传current_index）。策略函数甚至可以写成 lambda，例如strategy=lambda proxies, idx: (proxies[-1], idx)；带调用计数的“两比一交替”策略也在 tests/fetchers/test_proxy_rotation.py 的TestCustomStrategies中得到验证。

四、按请求覆盖代理（Per-Request Proxy Override）

可以在单个请求上通过proxy=关键字参数绕过轮转器：

async def parse(self, response: Response): # This request uses the rotator's next proxy yield response.follow("/page1", callback=self.parse_page) # This request uses a specific proxy, bypassing the rotator yield response.follow( "/special-page", callback=self.parse_page, proxy="http://special-proxy:8080", )

这在对区域内容有特殊要求时很有用（例如地理定位代理）。源码中的优先级规则是：只要请求携带了静态proxy，就不会调用self._proxy_rotator.get_proxy()，静态代理优先（见 static.py 与 _controllers.py）。对浏览器会话，静态代理同样会触发“新建独立 context”的行为（浏览器 fetch 的参数文档即注明“A new browser context will be created and used with it”）。

两条重要约束

proxy_rotator不能与静态proxy/proxies参数同时配置在同一个会话上。配置阶段就会直接抛错，两条实现路径的错误信息分别为：静态会话抛出Cannot use 'proxy_rotator' together with 'proxy' or 'proxies'. Use either a static proxy or proxy rotation, not both.（static.py）；浏览器会话的配置校验器在 _validators.py 中检查proxy与proxy_rotator的互斥。按需在单请求级别再用proxy=覆盖是允许的。
浏览器会话使用ProxyRotator时的上下文模型变化：默认情况下，浏览器会话使用持久化 context（带标签页池）；但浏览器无法为单个标签页设置代理，因此一旦启用ProxyRotator，fetcher 会为每个代理单独开一个 context、每个 context 开一个标签页，任务完成后标签页和 context 一起关闭。源码印证：start()中当self._config.proxy_rotator为真时，走的是chromium.launch(...)（非持久化）分支而非launch_persistent_context(...)（见 _controllers.py），代理参数则由每次 fetch 时的_page_generator按当前分配的代理创建对应 context。

五、被封锁请求的检测与重试（Blocked Request Handling）

Spider 内建了被封锁检测与重试。默认视为“被封锁”的 HTTP 状态码为：401、403、407、429、444、500、502、503、504。该集合定义在 spider.py 的BLOCKED_CODES常量中，默认的is_blocked()仅检查状态码是否落在该集合内（spider.py，行为在 tests/spiders/test_spider.py 中有参数化验证）。

引擎侧的重试流程（核心实现在 engine.py）：

响应返回后，引擎调用await self.spider.is_blocked(response)；
若判定为封锁且尚未超重试上限，引擎复制原请求（request.copy()），把_retry_count加 1，调用retry_blocked_request()让你在重发前修改它；
重试请求以dont_filter=True（绕过去重过滤）且priority -= 1（降低优先级，避免立刻重发）重新入队；
循环直到达到max_blocked_retries次（默认 3），超出后引擎记录Max retries exceeded for blocked request: {url}并放弃。

对应日志形如Scheduled blocked request for retry (1/3): https://...，排查重试行为时可直接在运行日志中观察。

两条关键提示：

重试时自动清空代理参数：源码中retry_request._session_kwargs.pop("proxy", None)与.pop("proxies", None)（engine.py）保证重发请求不再沿用上一次失败所用的静态代理，从而由轮转器分配一个全新的代理。
max_blocked_retries与 sessions retries 是两套独立计数器。会话层的 retries（FetcherSession默认retries=3、retry_delay=1，见 static.py）处理的是“请求抛异常”的场景，典型如代理连接失败——is_proxy_error()会识别net::err_proxy、connection refused、connection reset、connection timed out、failed to connect、could not resolve proxy等特征串（proxy_rotation.py），此时会话按retry_delay间隔重试并换一个新代理；而max_blocked_retries处理的是“响应正常返回但被站点封锁”的场景。两者不共享计数，也不互相触发。

六、自定义封锁检测：重写 is_blocked()

默认只看状态码，你可以通过重写is_blocked()叠加基于响应内容的检测（如反爬提示页）：

class MySpider(Spider): name = "my_spider" start_urls = ["https://example.com"] async def is_blocked(self, response: Response) -> bool: # Check status codes (default behavior) if response.status in {403, 429, 503}: return True # Check response content body = response.body.decode("utf-8", errors="ignore") if "access denied" in body.lower() or "rate limit" in body.lower(): return True return False async def parse(self, response: Response): yield {"title": response.css("title::text").get("")}

七、自定义重试行为：重写 retry_blocked_request() 并切换会话

retry_blocked_request(request, response) -> Request在每次重试前被调用（默认实现原样返回请求，见 spider.py）。典型用法是：日常请求走轻量 HTTP 会话，被封锁后自动切换到隐身浏览器会话。max_blocked_retries控制最多重试次数（默认 3）：

from scrapling.spiders import Spider, SessionManager, Request, Response from scrapling.fetchers import FetcherSession, AsyncStealthySession class MySpider(Spider): name = "my_spider" start_urls = ["https://example.com"] max_blocked_retries = 5 def configure_sessions(self, manager: SessionManager) -> None: manager.add('requests', FetcherSession(impersonate=['chrome', 'firefox', 'safari'])) manager.add('stealth', AsyncStealthySession(block_webrtc=True), lazy=True) async def retry_blocked_request(self, request: Request, response: Response) -> Request: request.sid = "stealth" self.logger.info(f"Retrying blocked request: {request.url}") return request async def parse(self, response: Response): yield {"title": response.css("title::text").get("")}

上面的效果是：封锁检测逻辑保持默认，蜘蛛主要使用 HTTP 请求（impersonate传列表时每次请求会在 chrome/firefox/safari 指纹间随机挑选），一旦被封，重试请求的sid被改写到stealth会话，由隐身浏览器完成抓取。注意lazy=True的注册方式——SessionManager中懒加载会话只会在第一次被请求引用时才启动（session.py），因此隐身浏览器只有在真正触发降级时才会被拉起，避免不必要的启动开销。

八、完整实战：廉价代理打头，封锁后升级到昂贵代理

把代理轮转和会话降级组合起来，就是常见的成本控制方案：数据中心（廉价）代理池先行，被封锁后切到住宅/移动端（昂贵）代理池：

from scrapling.spiders import Spider, SessionManager, Request, Response from scrapling.fetchers import FetcherSession, AsyncStealthySession, ProxyRotator cheap_proxies = ProxyRotator([ "http://proxy1:8080", "http://proxy2:8080"]) # A format acceptable by the browser expensive_proxies = ProxyRotator([ {"server": "http://residential_proxy1:8080", "username": "user", "password": "pass"}, {"server": "http://residential_proxy2:8080", "username": "user", "password": "pass"}, {"server": "http://mobile_proxy1:8080", "username": "user", "password": "pass"}, {"server": "http://mobile_proxy2:8080", "username": "user", "password": "pass"}, ]) class MySpider(Spider): name = "my_spider" start_urls = ["https://example.com"] max_blocked_retries = 5 def configure_sessions(self, manager: SessionManager) -> None: manager.add('requests', FetcherSession(impersonate=['chrome', 'firefox', 'safari'], proxy_rotator=cheap_proxies)) manager.add('stealth', AsyncStealthySession(block_webrtc=True, proxy_rotator=expensive_proxies), lazy=True) async def retry_blocked_request(self, request: Request, response: Response) -> Request: request.sid = "stealth" self.logger.info(f"Retrying blocked request: {request.url}") return request async def parse(self, response: Response): yield {"title": response.css("title::text").get("")}

这条链路的完整运行逻辑：

请求默认落在requests会话（SessionManager.add中先注册的会话即默认会话，见 session.py），每次请求从cheap_proxies循环取一个廉价代理；
响应状态码落入BLOCKED_CODES时，引擎清空请求上的代理参数、降低优先级并重新入队；
retry_blocked_request()把sid改写为stealth，首次触发时懒加载启动隐身浏览器，且其请求走expensive_proxies（浏览器会话接受字典格式代理，每个代理独立 context）；
若昂贵代理仍被封锁，继续按max_blocked_retries=5上限重试，超限则放弃并记警告日志。

九、验证与延伸阅读

代理轮转的完整行为（循环回绕、单代理、字典代理、自定义策略、lambda 策略、线程安全、is_proxy_error的识别边界）均可在 tests/fetchers/test_proxy_rotation.py 中对照验证；
is_blocked()/retry_blocked_request()的默认行为（状态码判定、原样返回请求）在 tests/spiders/test_spider.py 中有单元测试；
引擎层的入队、去重与优先级机制见 scrapling/spiders/scheduler.py；
代理轮转的 API 参考文档见 docs/api-reference/proxy-rotation.md，抓取器通用参数说明见 docs/api-reference/fetchers.md。

适用前提与限制：ProxyRotator的互斥校验发生在会话构造/配置阶段，因此“同会话静态代理 + 轮转”的写法会直接失败；浏览器会话启用轮转后不再使用持久化 context，长会话 cookie/本地存储不会跨代理保留；is_blocked的默认判定只覆盖状态码，对返回 200 的软性反爬页（验证码、空白页）需要自行重写检测逻辑。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

企业官网建设流程全解析