企业官网建设流程全解析

Scrapling Spider 系统进阶实战：并发控制、AutoThrottle 自适应限速、断点续爬与流式输出

【免费下载链接】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 官方文档 Advanced usages 展开，系统讲解其爬虫框架（Spider System）的进阶能力：并发与限速控制、AutoThrottle 自适应延迟、uvloop 事件循环、断点续爬（checkpoint）、开发模式响应缓存、流式输出（streaming）、生命周期钩子、统计与日志。读完本文，你可以把 Scrapling 的 Spider 从"能跑起来"提升到"可控、可恢复、可观测、可嵌入应用"的生产级形态。前置知识请参考 Getting started。

并发控制：三个类属性 + robots.txt 合规

Spider 系统通过Spider基类的几个类属性控制爬取节奏。这些默认值可以在基类定义中直接看到（Spider 类属性）：

class PoliteSpider(Spider): name = "polite" start_urls = ["https://example.com"] # Be gentle with the server concurrent_requests = 4 concurrent_requests_per_domain = 2 download_delay = 1.0 # Wait 1 second between requests async def parse(self, response: Response): yield {"title": response.css("title::text").get("")}

当设置了concurrent_requests_per_domain时，引擎会为每个域名额外创建一个独立的并发限流器，叠加在全局限流之上。这在同时爬取多个域名时非常有用：你可以允许较高的全局并发，同时对每个具体域名保持克制。

download_delay则是在每个请求前无条件地加上一个固定等待时间，与域名无关，适合做简单的全局限速。

从源码看，这一机制落在 CrawlerEngine：当concurrent_requests_per_domain非零时，引擎用self._domain_limiters.setdefault(domain, CapacityLimiter(...))为每个域名惰性创建 an anyioCapacityLimiter；未设置时所有请求只受全局_global_limiter（容量为concurrent_requests）约束。另外在 crawl 主循环 中，只有_active_tasks < concurrent_requests时才会从调度器取请求并派生新任务，因此不会出现成千上万个任务排队等待的情况。

如果开启robots_txt_obey，延迟计算还会参考 robots.txt：_get_domain_delay 会取download_delay与 robots.txt 中Crawl-delay、Request-rate（换算为period / req_count）三者中的最大值，并按域名缓存。也就是说 robots.txt 的礼貌指令只会让爬虫更慢，不会更快。

AutoThrottle：按域名自适应调整延迟

固定的download_delay本质上是个猜测：设低了会被封，设高了爬一整晚。AutoThrottle 的思路是观察目标站点的实际响应速度，按域名独立调整延迟——快服务器就加速，慢或敌对的服务器就退让。

相关类属性（同样定义在 Spider 基类）：

class AdaptiveSpider(Spider): name = "adaptive" start_urls = ["https://example.com"] concurrent_requests = 8 concurrent_requests_per_domain = 1 autothrottle_enabled = True autothrottle_start_delay = 2.0 autothrottle_max_delay = 30.0 async def parse(self, response: Response): yield {"title": response.css("title::text").get("")}

延迟如何收敛

每个响应结束后，该域名的延迟都会向"服务器实际耗时"靠拢：若站点 0.5 秒就能响应，延迟最终会稳定在 0.5 秒左右，大致相当于同一时间只有一个请求在途。目标并发数的解析顺序是：设置了autothrottle_target_concurrency就用它；否则用concurrent_requests_per_domain；都没有则取 1。从源码看，CrawlerEngine 构造时的这段逻辑 正是如此实现：target_concurrency=(self.spider.autothrottle_target_concurrency or self.spider.concurrent_requests_per_domain or 1.0)，延迟会除以它。因此通常你完全不需要单独设置目标并发——你为域名限流配置的数值会被自动复用。

具体算法在 AutoThrottle.record()：目标延迟为latency / target_concurrency，新延迟为当前值与目标值的均值且不低于目标值本身——这意味着延迟尖峰会立即生效（爬虫立刻退让），而加速只会逐次渐进。最终结果被夹在floor（见下节"地板"说明）和autothrottle_max_delay之间。

被封锁时的退避（Backing off when blocked）

单看延迟是发现不了限流的：429、403或验证码页面往往比真实内容响应得更快，纯按延迟看反而像是"可以加速"的信号。因此任何不健康的响应——非 2xx 状态码，或你的is_blocked()返回True的响应——都会把该域名的延迟翻倍：

0.5s -> 1s -> 2s -> 4s -> 8s ... (up to autothrottle_max_delay)

只要网站持续拒绝，爬虫就会持续减速；等健康响应回归后，正常的均值化过程会把延迟重新拉回服务器的真实速度，整个过程无需人工干预。源码中对应 throttle.py：new_delay = max(new_delay, penalty, current_delay)，其中注释明确"A block can never speed the spider up"（被封锁永远不会让爬虫提速），翻倍系数即模块常量BLOCK_BACKOFF_FACTOR = 2.0。

如果网站通过Retry-After响应头明确告知等待时长（429或503），则直接采用该值替代翻倍。parse_retry_after() 同时支持数字形式（Retry-After: 120）和 HTTP-date 形式。

把autothrottle_block_backoff = False即可关闭此机制，退回纯延迟驱动：此时被封锁最多只能阻止爬虫加速，而不能触发翻倍退避。

每个域名的最终延迟会出现在统计信息里：

result = AdaptiveSpider().start() print(result.stats.autothrottle_delays) # {'example.com': 0.62}

几个值得注意的边界（文档原文说明 + 源码印证）：

• download_delay与 robots.txt 的Crawl-delay充当"地板"（floor）：AutoThrottle 只会在它们之上调整延迟，礼貌配置永远不会被 undercut。从 delay_for() 可以看到首值即min(max(floor, start_delay), max_delay)，而 engine 的 record 调用 每次都把该域名的 floor 传入。
• concurrent_requests_per_domain身兼二职：既限制在途请求数，又作为 AutoThrottle 的目标值，所以通常不必单独配置目标值。文档建议显式设置它——因为被节流休眠的域名会占用一个全局限流槽位，不设域名上限意味着这些休眠槽位会挤占其他域名的共享预算。
• 测量的延迟是完整的一次 fetch，包含内部重试，浏览器会话还包含页面渲染时间。
• autothrottle_max_delay是天花板，Retry-After同样受它约束；如果网站要求的等待时间超过你的上限，应调高autothrottle_max_delay才能真正服从。
• 学到的延迟不会被 checkpoint 保存。暂停后恢复，各域名会重新从autothrottle_start_delay起步（crawl() 中的 reset 调用 也会说明每次运行开始时延迟表都会被清空）。

使用 uvloop 事件循环

start()接受use_uvloop参数，在可用时使用更快的 uvloop（Linux/macOS）或 winloop（Windows）事件循环实现：

result = MySpider().start(use_uvloop=True)

这可以提升 I/O 密集型爬取的吞吐。需要自行安装uvloop或winloop包。从 start() 实现 看，该参数最终被转换为 anyio 的backend_options={"use_uvloop": True}，再传给anyio.run(..., backend="asyncio")；start()还支持透传其他backend_options。

暂停与恢复：Checkpoint 断点续爬

Spider 支持通过 checkpoint 实现优雅的暂停/恢复。启用方式是在构造时传入crawldir目录：

spider = MySpider(crawldir="crawl_data/my_spider") result = spider.start() if result.paused: print("Crawl was paused. Run again to resume.") else: print("Crawl completed!")

工作机制

1. 暂停：爬取过程中按Ctrl+C。Spider 会等待所有在途请求完成，保存一个 checkpoint（待处理请求队列 + 已见请求指纹集合），然后退出。
2. 强制停止：第二次按Ctrl+C立即停止，不再等待活动任务。
3. 恢复：用同一个crawldir再次运行 Spider。它会检测到 checkpoint，恢复队列与已见集合，从断点继续，并跳过start_requests()。
4. 清理：爬取正常完成（非暂停）时，checkpoint 文件会被自动删除。

爬取过程中也会周期性保存 checkpoint（默认每 5 分钟）。可以修改间隔：

# Save checkpoint every 2 minutes spider = MySpider(crawldir="crawl_data/my_spider", interval=120.0)

磁盘写入是原子的，完全安全。源码印证：CheckpointManager.save() 先序列化为checkpoint.tmp，再用temp_path.replace(...)原子改名；checkpoint 文件固定为crawldir/checkpoint.pkl（CHECKPOINT_FILE 常量）。暂停/强制停止的双级语义在 request_pause() 中实现：第一次调用置_pause_requested，主循环检测到后先在活动任务归零（或强制停止）时保存 checkpoint；第二次调用置_force_stop并立即取消任务组。周期性保存由 crawl 主循环 中的_is_checkpoint_time()驱动（默认interval=300.0秒，见 Spider.init与 CheckpointManager.init的校验逻辑）。

!!! 说明

即使没有启用 checkpoint，`Ctrl+C` 也总能触发优雅关闭；等待中再按一次会强制立即关闭。信号处理逻辑见 [_setup_signal_handler](https://link.gitcode.com/i/28d6cf21f11e9f5a4baffb25fc18bea8)。

感知自己是否在恢复

on_start()钩子会收到一个resuming标志：

async def on_start(self, resuming: bool = False): if resuming: self.logger.info("Resuming from checkpoint!") else: self.logger.info("Starting fresh crawl")

引擎在恢复成功时确实会传入该标志（crawl() 中的 resuming 变量），并且恢复时打印 "Resuming from checkpoint, skipping start_requests()" 后直接进入队列处理。

开发模式：响应本地缓存与回放

调试parse()逻辑时，每次运行都重新请求目标服务器既慢又吵。开发模式在首次运行把所有响应缓存到磁盘，之后每次运行都直接从磁盘回放——你可以随意改选择器、反复重跑，而不发出一个网络请求。

在你的 Spider 上设置development_mode = True开启：

class MySpider(Spider): name = "my_spider" start_urls = ["https://example.com"] development_mode = True async def parse(self, response: Response): yield {"title": response.css("title::text").get("")}

首次运行正常抓取并把每个响应存盘；此后每次运行都从缓存服务相同请求，完全跳过网络。

缓存位置

默认缓存在当前工作目录下（注意：是你运行 spider 的目录，而不是 spider 脚本所在的目录）的.scrapling_cache/{spider.name}/。可以用development_cache_dir覆盖：

class MySpider(Spider): name = "my_spider" start_urls = ["https://example.com"] development_mode = True development_cache_dir = "/tmp/my_spider_cache"

默认路径的拼装逻辑见 CrawlerEngine 构造：cache_dir = self.spider.development_cache_dir or f".scrapling_cache/{self.spider.name}"。

工作机制

1. 缓存键：每个响应以请求指纹（fingerprint）为键。任何影响指纹的属性（fp_include_kwargs、fp_include_headers、fp_keep_fragments）变化都会触发一次全新抓取。
2. 存储格式：每个响应一个 JSON 文件，命名{fingerprint_hex}.json；响应体 base64 编码以精确保存二进制内容；写入是原子的（临时文件 + rename）。实现见 ResponseCacheManager。
3. 回放：缓存命中时引擎完全跳过网络——包括download_delay、速率限制和is_blocked()重试路径——缓存的响应直接送进你的回调。对应 engine 中的缓存命中分支：命中后只累加统计并调用_run_callbacks，随后return。
4. 统计：缓存命中的请求同样计入requests_count、response_bytes和按状态码的计数，所以统计输出看起来与真实爬取一致；另有cache_hits与cache_misses两个计数器可以观察缓存表现。

清理缓存

缓存没有自动过期机制。要强制重新抓取，删除缓存目录，或调用缓存管理器的clear()方法（实现，会删除目录下所有.json文件）。

警告：开发模式只用于开发，不用于生产。缓存响应永不过期，回放还会绕过速率限制与被封锁重试。不要带着development_mode = True上线。

Streaming：用 stream() 实时获取条目

对于长时间运行的 Spider、或需要实时拿到抓取条目的应用，用stream()代替start()：

import anyio async def main(): spider = MySpider() async for item in spider.stream(): print(f"Got item: {item}") # Access real-time stats print(f"Items so far: {spider.stats.items_scraped}") print(f"Requests made: {spider.stats.requests_count}") anyio.run(main)

与start()的关键区别：

• stream()必须在异步上下文中调用；
• 条目在抓取的同时逐条 yield，而不是收集成列表；
• 迭代过程中可以随时读取spider.stats获得实时统计。

全部可用统计字段见下文"结果与统计"一节。

stream()也能与 checkpoint 体系配合，非常适合在 Spider 之上构建带实时数据、可暂停/恢复的 UI：

import anyio async def main(): spider = MySpider(crawldir="crawl_data/my_spider") async for item in spider.stream(): print(f"Got item: {item}") print(f"Items so far: {spider.stats.items_scraped}") print(f"Requests made: {spider.stats.requests_count}") anyio.run(main)

上面的代码里还可以调用spider.pause()来从代码内关停 Spider；若没有启用 checkpoint 系统，它就直接结束爬取。pause()的实现见 Spider.pause（本质是调用引擎的request_pause()）。需要注意，从 stream() 的 docstring 说明看，stream 模式下不提供 SIGINT（Ctrl+C）的暂停/恢复处理，程序化暂停是 stream 模式下的推荐方式。

生命周期钩子

Spider 提供若干可覆写的钩子，用于在爬取的不同阶段注入自定义行为。这些钩子的基类实现都在 Spider 类。

on_start

爬取开始前调用，适合做加载数据、初始化资源等准备：

async def on_start(self, resuming: bool = False): self.logger.info("Spider starting up") # Load seed URLs from a database, initialize counters, etc.

on_close

爬取结束后调用（无论完成还是暂停），适合做清理：

async def on_close(self): self.logger.info("Spider shutting down") # Close database connections, flush buffers, etc.

on_error

请求因异常失败时调用，适合做错误追踪或自定义恢复逻辑：

async def on_error(self, request: Request, error: Exception): self.logger.error(f"Failed: {request.url} - {error}") # Log to error tracker, save failed URL for later, etc.

on_scraped_item

每个条目在进入结果集之前都会经过它。返回条目（可修改）即保留，返回None即丢弃：

async def on_scraped_item(self, item: dict) -> dict | None: # Drop items without a title if not item.get("title"): return None # Modify items (e.g., add timestamps) item["scraped_at"] = "2026-01-01" return item

该钩子还可以用来把条目导向你自己的数据管道，并让条目不进入 Spider 的默认结果集——引擎在 _run_callbacks 中处理返回结果：保留则items_scraped += 1（stream 模式下通过内存流实时送出），丢弃则items_dropped += 1。

start_requests

覆写start_requests()可以完全自定义初始请求，替代start_urls，典型场景是先登录再爬：

async def start_requests(self): # POST request to log in first yield Request( "https://example.com/login", method="POST", data={"user": "admin", "pass": "secret"}, callback=self.after_login, ) async def after_login(self, response: Response): # Now crawl the authenticated pages yield response.follow("/dashboard", callback=self.parse)

结果与统计：CrawlResult 和 CrawlStats

start()返回的CrawlResult同时包含抓取条目和详细统计（定义见 result.py）：

result = MySpider().start() # Items print(f"Total items: {len(result.items)}") result.items.to_json("output.json", indent=True) # Did the crawl complete? print(f"Completed: {result.completed}") print(f"Paused: {result.paused}") # Statistics stats = result.stats print(f"Requests: {stats.requests_count}") print(f"Failed: {stats.failed_requests_count}") print(f"Blocked: {stats.blocked_requests_count}") print(f"Offsite filtered: {stats.offsite_requests_count}") print(f"Robots.txt disallowed: {stats.robots_disallowed_count}") print(f"Cache hits: {stats.cache_hits}") print(f"Cache misses: {stats.cache_misses}") print(f"Items scraped: {stats.items_scraped}") print(f"Items dropped: {stats.items_dropped}") print(f"Response bytes: {stats.response_bytes}") print(f"Duration: {stats.elapsed_seconds:.1f}s") print(f"Speed: {stats.requests_per_second:.1f} req/s")

其中completed是not paused的派生属性；elapsed_seconds/requests_per_second分别由起止时间戳和请求数计算（CrawlStats 属性）。result.items是ItemList（一个带导出能力的 list），除to_json()外还支持to_jsonl()、to_csv()、to_xml()（ItemList）。

详细统计字段

CrawlStats是一个 dataclass，字段全集见 定义：

stats = result.stats # Status code distribution print(stats.response_status_count) # {'status_200': 150, 'status_404': 3, 'status_403': 1} # Bytes downloaded per domain print(stats.domains_response_bytes) # {'example.com': 1234567, 'api.example.com': 45678} # Requests per session print(stats.sessions_requests_count) # {'http': 120, 'stealth': 34} # Proxies used during the crawl print(stats.proxies) # ['http://proxy1:8080', 'http://proxy2:8080'] # Log level counts print(stats.log_levels_counter) # {'debug': 200, 'info': 50, 'warning': 3, 'error': 1, 'critical': 0} # Timing information print(stats.start_time) # Unix timestamp when crawl started print(stats.end_time) # Unix timestamp when crawl finished print(stats.download_delay) # The download delay used (seconds) # Concurrency settings used print(stats.concurrent_requests) # Global concurrency limit print(stats.concurrent_requests_per_domain) # Per-domain concurrency limit # AutoThrottle print(stats.autothrottle_enabled) # Whether the adaptive delay was on print(stats.autothrottle_delays) # Final delay per domain, {'example.com': 0.62} # Custom stats (set by your spider code) print(stats.custom_stats) # {'login_attempts': 3, 'pages_with_errors': 5} # Export everything as a dict print(stats.to_dict())

这些数字分别在哪里累加，可以在引擎源码中一一对上：状态码与字节数在 fetch 成功后；被过滤的站外请求在 入队检查处；robots 拒绝在 can_fetch 检查处；代理列表在 请求携带 proxy 时 追加。to_dict()还会输出requests_per_second、四舍五入后的elapsed_seconds与autothrottle_delays，方便直接落盘或上报（to_dict 实现）。

日志：内置 logger 与四个配置属性

Spider 自带名为scrapling.spiders.{spider.name}的 logger（self.logger），预配置了 Spider 名称，并支持以下类属性：

import logging class MySpider(Spider): name = "my_spider" start_urls = ["https://example.com"] logging_level = logging.INFO log_file = "logs/my_spider.log" async def parse(self, response: Response): self.logger.info(f"Processing {response.url}") yield {"title": response.css("title::text").get("")}

日志文件所在目录不存在时会自动创建；控制台与文件使用同一格式。实现细节见 Spider.init的 logger 装配：logging_format中的{spider_name}通过.format(spider_name=self.name)注入；log_file非空时先mkdir(parents=True)再挂FileHandler。此外引擎里还有一个 LogCounterHandler 按级别计数所有日志，这就是统计中log_levels_counter的来源；爬取结束时文件 handler 会在finally中关闭以释放文件资源（__run 的清理逻辑）。

小结

Scrapling Spider 系统的进阶能力围绕四个目标组织：可控（全局/域名并发、固定延迟、AutoThrottle 自适应、robots.txt 合规）、可恢复（checkpoint 暂停/恢复、周期保存、优雅关停）、可调试（开发模式缓存回放、生命周期钩子、完整统计与日志）、可集成（stream()实时输出与spider.stats实时统计，方便在爬虫之上构建 UI）。所有类属性与默认值均以 Spider 基类 为准，核心调度逻辑集中在 CrawlerEngine，自适应限速算法在 AutoThrottle，测试用例可参考tests/spiders/目录下的test_throttle.py、test_cache.py、test_checkpoint.py、test_force_stop_checkpoint.py等文件以验证上述行为的实际表现。

【免费下载链接】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