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 类属性属性默认值说明concurrent_requests4同一时刻正在处理的请求数上限全局并发concurrent_requests_per_domain0单域名最大并发0 不做域名级限制download_delay0.0每个请求发出前等待的秒数robots_txt_obeyFalse是否遵守 robots.txt 规则Disallow、Crawl-delay、Request-rateclass 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 基类属性默认值说明autothrottle_enabledFalse开启自适应延迟autothrottle_start_delay5.0对某域名的首个请求使用的延迟autothrottle_max_delay60.0延迟允许达到的上限autothrottle_target_concurrencyNone每个域名希望保持在途的请求数见下文解析顺序autothrottle_block_backoffTrue每次被该域名封禁非 2xx 或被is_blocked()标记就把延迟翻倍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.pynew_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充当地板floorAutoThrottle 只会在它们之上调整延迟礼貌配置永远不会被 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参数在可用时使用更快的 uvloopLinux/macOS或 winloopWindows事件循环实现result MySpider().start(use_uvloopTrue)这可以提升 I/O 密集型爬取的吞吐。需要自行安装uvloop或winloop包。从 start() 实现 看该参数最终被转换为 anyio 的backend_options{use_uvloop: True}再传给anyio.run(..., backendasyncio)start()还支持透传其他backend_options。暂停与恢复Checkpoint 断点续爬Spider 支持通过 checkpoint 实现优雅的暂停/恢复。启用方式是在构造时传入crawldir目录spider MySpider(crawldircrawl_data/my_spider) result spider.start() if result.paused: print(Crawl was paused. Run again to resume.) else: print(Crawl completed!)工作机制暂停爬取过程中按CtrlC。Spider 会等待所有在途请求完成保存一个 checkpoint待处理请求队列 已见请求指纹集合然后退出。强制停止第二次按CtrlC立即停止不再等待活动任务。恢复用同一个crawldir再次运行 Spider。它会检测到 checkpoint恢复队列与已见集合从断点继续并跳过start_requests()。清理爬取正常完成非暂停时checkpoint 文件会被自动删除。爬取过程中也会周期性保存 checkpoint默认每 5 分钟。可以修改间隔# Save checkpoint every 2 minutes spider MySpider(crawldircrawl_data/my_spider, interval120.0)磁盘写入是原子的完全安全。源码印证CheckpointManager.save() 先序列化为checkpoint.tmp再用temp_path.replace(...)原子改名checkpoint 文件固定为crawldir/checkpoint.pklCHECKPOINT_FILE 常量。暂停/强制停止的双级语义在 request_pause() 中实现第一次调用置_pause_requested主循环检测到后先在活动任务归零或强制停止时保存 checkpoint第二次调用置_force_stop并立即取消任务组。周期性保存由 crawl 主循环 中的_is_checkpoint_time()驱动默认interval300.0秒见 Spider.init与 CheckpointManager.init的校验逻辑。!!! 说明即使没有启用 checkpointCtrlC 也总能触发优雅关闭等待中再按一次会强制立即关闭。信号处理逻辑见 [_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}。工作机制缓存键每个响应以请求指纹fingerprint为键。任何影响指纹的属性fp_include_kwargs、fp_include_headers、fp_keep_fragments变化都会触发一次全新抓取。存储格式每个响应一个 JSON 文件命名{fingerprint_hex}.json响应体 base64 编码以精确保存二进制内容写入是原子的临时文件 rename。实现见 ResponseCacheManager。回放缓存命中时引擎完全跳过网络——包括download_delay、速率限制和is_blocked()重试路径——缓存的响应直接送进你的回调。对应 engine 中的缓存命中分支命中后只累加统计并调用_run_callbacks随后return。统计缓存命中的请求同样计入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(fGot item: {item}) # Access real-time stats print(fItems so far: {spider.stats.items_scraped}) print(fRequests made: {spider.stats.requests_count}) anyio.run(main)与start()的关键区别stream()必须在异步上下文中调用条目在抓取的同时逐条 yield而不是收集成列表迭代过程中可以随时读取spider.stats获得实时统计。全部可用统计字段见下文结果与统计一节。stream()也能与 checkpoint 体系配合非常适合在 Spider 之上构建带实时数据、可暂停/恢复的 UIimport anyio async def main(): spider MySpider(crawldircrawl_data/my_spider) async for item in spider.stream(): print(fGot item: {item}) print(fItems so far: {spider.stats.items_scraped}) print(fRequests made: {spider.stats.requests_count}) anyio.run(main)上面的代码里还可以调用spider.pause()来从代码内关停 Spider若没有启用 checkpoint 系统它就直接结束爬取。pause()的实现见 Spider.pause本质是调用引擎的request_pause()。需要注意从 stream() 的 docstring 说明看stream 模式下不提供 SIGINTCtrlC的暂停/恢复处理程序化暂停是 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(fFailed: {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 1stream 模式下通过内存流实时送出丢弃则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, methodPOST, data{user: admin, pass: secret}, callbackself.after_login, ) async def after_login(self, response: Response): # Now crawl the authenticated pages yield response.follow(/dashboard, callbackself.parse)结果与统计CrawlResult 和 CrawlStatsstart()返回的CrawlResult同时包含抓取条目和详细统计定义见 result.pyresult MySpider().start() # Items print(fTotal items: {len(result.items)}) result.items.to_json(output.json, indentTrue) # Did the crawl complete? print(fCompleted: {result.completed}) print(fPaused: {result.paused}) # Statistics stats result.stats print(fRequests: {stats.requests_count}) print(fFailed: {stats.failed_requests_count}) print(fBlocked: {stats.blocked_requests_count}) print(fOffsite filtered: {stats.offsite_requests_count}) print(fRobots.txt disallowed: {stats.robots_disallowed_count}) print(fCache hits: {stats.cache_hits}) print(fCache misses: {stats.cache_misses}) print(fItems scraped: {stats.items_scraped}) print(fItems dropped: {stats.items_dropped}) print(fResponse bytes: {stats.response_bytes}) print(fDuration: {stats.elapsed_seconds:.1f}s) print(fSpeed: {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}的 loggerself.logger预配置了 Spider 名称并支持以下类属性属性默认值说明logging_levellogging.DEBUG最低日志级别logging_format[%(asctime)s]:({spider_name}) %(levelname)s: %(message)s日志格式{spider_name}会被替换logging_date_format%Y-%m-%d %H:%M:%S日志中的日期格式log_fileNone日志文件路径在控制台输出之外追加写文件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(fProcessing {response.url}) yield {title: response.css(title::text).get()}日志文件所在目录不存在时会自动创建控制台与文件使用同一格式。实现细节见 Spider.init的 logger 装配logging_format中的{spider_name}通过.format(spider_nameself.name)注入log_file非空时先mkdir(parentsTrue)再挂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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Scrapling Spider 系统进阶实战：并发控制、AutoThrottle 自适应限速、断点续爬与流式输出
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 类属性属性默认值说明concurrent_requests4同一时刻正在处理的请求数上限全局并发concurrent_requests_per_domain0单域名最大并发0 不做域名级限制download_delay0.0每个请求发出前等待的秒数robots_txt_obeyFalse是否遵守 robots.txt 规则Disallow、Crawl-delay、Request-rateclass 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 基类属性默认值说明autothrottle_enabledFalse开启自适应延迟autothrottle_start_delay5.0对某域名的首个请求使用的延迟autothrottle_max_delay60.0延迟允许达到的上限autothrottle_target_concurrencyNone每个域名希望保持在途的请求数见下文解析顺序autothrottle_block_backoffTrue每次被该域名封禁非 2xx 或被is_blocked()标记就把延迟翻倍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.pynew_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充当地板floorAutoThrottle 只会在它们之上调整延迟礼貌配置永远不会被 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参数在可用时使用更快的 uvloopLinux/macOS或 winloopWindows事件循环实现result MySpider().start(use_uvloopTrue)这可以提升 I/O 密集型爬取的吞吐。需要自行安装uvloop或winloop包。从 start() 实现 看该参数最终被转换为 anyio 的backend_options{use_uvloop: True}再传给anyio.run(..., backendasyncio)start()还支持透传其他backend_options。暂停与恢复Checkpoint 断点续爬Spider 支持通过 checkpoint 实现优雅的暂停/恢复。启用方式是在构造时传入crawldir目录spider MySpider(crawldircrawl_data/my_spider) result spider.start() if result.paused: print(Crawl was paused. Run again to resume.) else: print(Crawl completed!)工作机制暂停爬取过程中按CtrlC。Spider 会等待所有在途请求完成保存一个 checkpoint待处理请求队列 已见请求指纹集合然后退出。强制停止第二次按CtrlC立即停止不再等待活动任务。恢复用同一个crawldir再次运行 Spider。它会检测到 checkpoint恢复队列与已见集合从断点继续并跳过start_requests()。清理爬取正常完成非暂停时checkpoint 文件会被自动删除。爬取过程中也会周期性保存 checkpoint默认每 5 分钟。可以修改间隔# Save checkpoint every 2 minutes spider MySpider(crawldircrawl_data/my_spider, interval120.0)磁盘写入是原子的完全安全。源码印证CheckpointManager.save() 先序列化为checkpoint.tmp再用temp_path.replace(...)原子改名checkpoint 文件固定为crawldir/checkpoint.pklCHECKPOINT_FILE 常量。暂停/强制停止的双级语义在 request_pause() 中实现第一次调用置_pause_requested主循环检测到后先在活动任务归零或强制停止时保存 checkpoint第二次调用置_force_stop并立即取消任务组。周期性保存由 crawl 主循环 中的_is_checkpoint_time()驱动默认interval300.0秒见 Spider.init与 CheckpointManager.init的校验逻辑。!!! 说明即使没有启用 checkpointCtrlC 也总能触发优雅关闭等待中再按一次会强制立即关闭。信号处理逻辑见 [_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}。工作机制缓存键每个响应以请求指纹fingerprint为键。任何影响指纹的属性fp_include_kwargs、fp_include_headers、fp_keep_fragments变化都会触发一次全新抓取。存储格式每个响应一个 JSON 文件命名{fingerprint_hex}.json响应体 base64 编码以精确保存二进制内容写入是原子的临时文件 rename。实现见 ResponseCacheManager。回放缓存命中时引擎完全跳过网络——包括download_delay、速率限制和is_blocked()重试路径——缓存的响应直接送进你的回调。对应 engine 中的缓存命中分支命中后只累加统计并调用_run_callbacks随后return。统计缓存命中的请求同样计入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(fGot item: {item}) # Access real-time stats print(fItems so far: {spider.stats.items_scraped}) print(fRequests made: {spider.stats.requests_count}) anyio.run(main)与start()的关键区别stream()必须在异步上下文中调用条目在抓取的同时逐条 yield而不是收集成列表迭代过程中可以随时读取spider.stats获得实时统计。全部可用统计字段见下文结果与统计一节。stream()也能与 checkpoint 体系配合非常适合在 Spider 之上构建带实时数据、可暂停/恢复的 UIimport anyio async def main(): spider MySpider(crawldircrawl_data/my_spider) async for item in spider.stream(): print(fGot item: {item}) print(fItems so far: {spider.stats.items_scraped}) print(fRequests made: {spider.stats.requests_count}) anyio.run(main)上面的代码里还可以调用spider.pause()来从代码内关停 Spider若没有启用 checkpoint 系统它就直接结束爬取。pause()的实现见 Spider.pause本质是调用引擎的request_pause()。需要注意从 stream() 的 docstring 说明看stream 模式下不提供 SIGINTCtrlC的暂停/恢复处理程序化暂停是 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(fFailed: {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 1stream 模式下通过内存流实时送出丢弃则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, methodPOST, data{user: admin, pass: secret}, callbackself.after_login, ) async def after_login(self, response: Response): # Now crawl the authenticated pages yield response.follow(/dashboard, callbackself.parse)结果与统计CrawlResult 和 CrawlStatsstart()返回的CrawlResult同时包含抓取条目和详细统计定义见 result.pyresult MySpider().start() # Items print(fTotal items: {len(result.items)}) result.items.to_json(output.json, indentTrue) # Did the crawl complete? print(fCompleted: {result.completed}) print(fPaused: {result.paused}) # Statistics stats result.stats print(fRequests: {stats.requests_count}) print(fFailed: {stats.failed_requests_count}) print(fBlocked: {stats.blocked_requests_count}) print(fOffsite filtered: {stats.offsite_requests_count}) print(fRobots.txt disallowed: {stats.robots_disallowed_count}) print(fCache hits: {stats.cache_hits}) print(fCache misses: {stats.cache_misses}) print(fItems scraped: {stats.items_scraped}) print(fItems dropped: {stats.items_dropped}) print(fResponse bytes: {stats.response_bytes}) print(fDuration: {stats.elapsed_seconds:.1f}s) print(fSpeed: {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}的 loggerself.logger预配置了 Spider 名称并支持以下类属性属性默认值说明logging_levellogging.DEBUG最低日志级别logging_format[%(asctime)s]:({spider_name}) %(levelname)s: %(message)s日志格式{spider_name}会被替换logging_date_format%Y-%m-%d %H:%M:%S日志中的日期格式log_fileNone日志文件路径在控制台输出之外追加写文件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(fProcessing {response.url}) yield {title: response.css(title::text).get()}日志文件所在目录不存在时会自动创建控制台与文件使用同一格式。实现细节见 Spider.init的 logger 装配logging_format中的{spider_name}通过.format(spider_nameself.name)注入log_file非空时先mkdir(parentsTrue)再挂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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考