从脚本到系统：设计一个支持插件、限流、重试与监控的 Python 异步爬虫框架

从脚本到系统设计一个支持插件、限流、重试与监控的 Python 异步爬虫框架很多人第一次写 Python 爬虫都是从几十行脚本开始的requests.get()、BeautifulSoup、for循环、保存 CSV。它很快也很有成就感。但真实项目往往不是“抓一个页面”这么简单而是目标站点很多、页面结构变化频繁、网络偶发失败、接口有访问频率限制、任务需要长期运行、线上出错要能追踪。这时问题就从“写一个爬虫”变成了“设计一个爬虫系统”。本文面向既想打牢 Python 编程基础、又希望进入 Python 实战与工程化阶段的读者。我们将设计一个支持插件、限流、重试、监控的异步爬虫系统用它回答一个高级 Python 工程师经常面对的问题如何把混乱的抓取需求组织成可扩展、可维护、可观测的工程体系。在技术选型上我们使用asyncio作为并发基础。Python 官方文档说明asyncio用async/await编写并发代码并且是多个异步框架的基础这非常适合网络 I/O 密集型场景。(Python documentation) HTTP 客户端可以选择aiohttp它的客户端请求支持异步上下文管理器适合构建资源安全的异步请求流程。(AIOHTTP)一、先定义边界一个工程化爬虫系统应该解决什么一个可上线的异步爬虫系统至少要回答六个问题1. URL 从哪里来 2. 请求如何并发执行 3. 不同网站如何解析 4. 失败如何重试 5. 访问频率如何控制 6. 系统状态如何监控如果把它画成流程图大致是这样Seed URLsSchedulerRate LimiterAsync FetcherRetry PolicyPlugin ParserPipeline StorageMetrics Monitor这个结构的核心思想是抓取、解析、存储、监控分离。不要把所有逻辑写进一个crawl.py否则三天后你自己都会害怕打开它。二、项目目录设计先让系统有秩序推荐目录如下async_crawler/ ├── crawler/ │ ├── core.py # 爬虫主流程 │ ├── fetcher.py # 异步请求 │ ├── limiter.py # 限流器 │ ├── retry.py # 重试策略 │ ├── plugin.py # 插件协议与加载 │ ├── pipeline.py # 数据处理与存储 │ └── metrics.py # 监控指标 ├── plugins/ │ ├── news.py │ └── product.py ├── tests/ │ └── test_parser.py └── main.py目录不是形式主义。好的目录结构会告诉后来者哪里放规则哪里放变化哪里是稳定核心。三、基础能力异步请求器 Fetcher爬虫的第一层能力是请求页面。同步爬虫一次只能等待一个网络响应而异步爬虫可以在等待 A 网站返回时同时调度 B、C、D 请求。# crawler/fetcher.pyimportaiohttpimportasynciofromdataclassesimportdataclassdataclassclassFetchResult:url:strstatus:inttext:strclassAsyncFetcher:def__init__(self,timeout:int10):self.timeoutaiohttp.ClientTimeout(totaltimeout)asyncdeffetch(self,session:aiohttp.ClientSession,url:str)-FetchResult:asyncwithsession.get(url,timeoutself.timeout)asresp:textawaitresp.text(errorsignore)returnFetchResult(urlurl,statusresp.status,texttext)这里用了async with它不只是语法优雅更重要的是能确保连接资源被正确释放。高级 Python 编程的一个重要习惯是资源生命周期要显式管理。四、限流设计别让并发变成攻击很多新手误以为异步爬虫越快越好这是危险的。成熟爬虫一定要有礼貌设置合理User-Agent、尊重目标站规则、遵守robots.txt、限制并发、避免对目标服务造成压力。一个简单的限流器可以基于asyncio.Semaphore实现# crawler/limiter.pyimportasynciofromcollectionsimportdefaultdictclassDomainRateLimiter:def__init__(self,per_domain_limit:int3,delay:float0.5):self._locksdefaultdict(lambda:asyncio.Semaphore(per_domain_limit))self._delaydelayasyncdefacquire(self,domain:str):semself._locks[domain]awaitsem.acquire()awaitasyncio.sleep(self._delay)returnsem使用时fromurllib.parseimporturlparseasyncdeflimited_fetch(fetcher,session,limiter,url):domainurlparse(url).netloc semawaitlimiter.acquire(domain)try:returnawaitfetcher.fetch(session,url)finally:sem.release()这段代码解决的是“每个域名最多同时几个请求”。真实项目中还可以加令牌桶、滑动窗口、分布式限流等方案。五、重试机制失败不可怕不可控才可怕网络请求天然不稳定。DNS 抖动、连接超时、服务端 502、503、504 都可能发生。高级工程师不会简单地except Exception: pass而是设计明确的重试策略。如果项目允许依赖第三方库可以使用 Tenacity。Tenacity 是一个通用 Python 重试库目标就是简化重试行为的添加。(Tenacity)当然我们也可以先实现一个轻量版本# crawler/retry.pyimportasyncioimportrandom RETRY_STATUS{429,500,502,503,504}classRetryPolicy:def__init__(self,max_attempts:int3,base_delay:float0.5):self.max_attemptsmax_attempts self.base_delaybase_delayasyncdefrun(self,coro_factory):last_errorNoneforattemptinrange(1,self.max_attempts1):try:resultawaitcoro_factory()ifresult.statusnotinRETRY_STATUS:returnresult last_errorRuntimeError(fretryable status:{result.status})exceptExceptionasexc:last_errorexc delayself.base_delay*(2**(attempt-1))jitterrandom.uniform(0,0.2)awaitasyncio.sleep(delayjitter)raiselast_error这里用了指数退避和随机抖动。它的价值在于失败后不要所有任务一起立刻重试否则可能造成“雪崩式重试”。六、插件系统让变化关在笼子里不同网站结构不同解析规则也不同。如果把所有解析逻辑写进主流程系统很快就会变成一锅粥。我们可以定义插件协议# crawler/plugin.pyfromtypingimportProtocol,IterablefromdataclassesimportdataclassdataclassclassItem:source:strtitle:strurl:strclassSpiderPlugin(Protocol):name:strdefmatch(self,url:str)-bool:...defparse(self,url:str,html:str)-Iterable[Item]:...新闻插件示例# plugins/news.pyfrombs4importBeautifulSoupfromcrawler.pluginimportItemclassNewsPlugin:namenewsdefmatch(self,url:str)-bool:returnnewsinurldefparse(self,url:str,html:str):soupBeautifulSoup(html,html.parser)forainsoup.select(a):titlea.get_text(stripTrue)hrefa.get(href)iftitleandhref:yieldItem(sourceself.name,titletitle,urlhref)插件管理器# crawler/plugin.pyclassPluginManager:def__init__(self,plugins):self.pluginspluginsdefselect(self,url:str):forplugininself.plugins:ifplugin.match(url):returnpluginreturnNone插件化的意义不是“看起来架构很高级”而是把变化隔离起来。目标站点结构变了只改对应插件不动核心调度器。七、数据管道不要边爬边乱存抓到数据后不建议直接在解析函数里写数据库。更好的方式是走统一 Pipeline# crawler/pipeline.pyimportjsonclassJsonLinePipeline:def__init__(self,path:str):self.pathpathasyncdefsave_many(self,items):withopen(self.path,a,encodingutf-8)asf:foriteminitems:f.write(json.dumps(item.__dict__,ensure_asciiFalse)\n)真实项目可以替换成 MySQL、PostgreSQL、MongoDB、Kafka 或对象存储。核心原则是解析只负责解析存储只负责存储。八、监控没有指标的系统等于盲飞异步爬虫跑起来之后你最关心的不是“它现在还活着吗”而是每分钟抓取多少页面 成功率是多少 失败最多的是哪些状态码 平均响应时间是多少 队列是否积压 插件解析失败率是多少Prometheus 是常见的监控方案它提供多维数据模型、PromQL 查询语言和告警能力。(prometheus.io) 在 Python 中可以设计如下指标层# crawler/metrics.pyfromcollectionsimportCounterimporttimeclassMetrics:def__init__(self):self.counterCounter()self.latencies[]definc(self,name:str):self.counter[name]1defobserve_latency(self,seconds:float):self.latencies.append(seconds)defreport(self):avgsum(self.latencies)/len(self.latencies)ifself.latencieselse0return{counts:dict(self.counter),avg_latency:round(avg,4),}请求时打点importtimeasyncdefmonitored_fetch(fetcher,session,retry_policy,metrics,url):starttime.perf_counter()try:resultawaitretry_policy.run(lambda:fetcher.fetch(session,url))metrics.inc(fstatus_{result.status})returnresultexceptException:metrics.inc(failed)raisefinally:metrics.observe_latency(time.perf_counter()-start)很多工程事故不是因为没人写代码而是系统出问题时没人知道问题在哪里。监控就是给系统装上仪表盘。九、主流程把组件组装成系统现在我们把 Fetcher、Limiter、Retry、Plugin、Pipeline、Metrics 组合起来# crawler/core.pyimportaiohttpimportasyncioclassAsyncCrawler:def__init__(self,urls,fetcher,limiter,retry_policy,plugins,pipeline,metrics):self.urlsurls self.fetcherfetcher self.limiterlimiter self.retry_policyretry_policy self.pluginsplugins self.pipelinepipeline self.metricsmetricsasyncdefcrawl_one(self,session,url):resultawaitlimited_fetch(self.fetcher,session,self.limiter,url)pluginself.plugins.select(url)ifnotplugin:self.metrics.inc(no_plugin)returnitemslist(plugin.parse(url,result.text))awaitself.pipeline.save_many(items)self.metrics.inc(page_done)self.metrics.inc(fitems_{len(items)})asyncdefrun(self):asyncwithaiohttp.ClientSession(headers{User-Agent:AsyncCrawlerBot/1.0})assession:tasks[self.crawl_one(session,url)forurlinself.urls]awaitasyncio.gather(*tasks,return_exceptionsTrue)print(self.metrics.report())入口文件# main.pyimportasynciofromcrawler.fetcherimportAsyncFetcherfromcrawler.limiterimportDomainRateLimiterfromcrawler.retryimportRetryPolicyfromcrawler.pluginimportPluginManagerfromcrawler.pipelineimportJsonLinePipelinefromcrawler.metricsimportMetricsfromcrawler.coreimportAsyncCrawlerfromplugins.newsimportNewsPlugin urls[https://example.com/news/1,https://example.com/news/2,]crawlerAsyncCrawler(urlsurls,fetcherAsyncFetcher(timeout10),limiterDomainRateLimiter(per_domain_limit2,delay1),retry_policyRetryPolicy(max_attempts3),pluginsPluginManager([NewsPlugin()]),pipelineJsonLinePipeline(items.jsonl),metricsMetrics(),)asyncio.run(crawler.run())这已经不是一个“脚本”而是一个具备工程骨架的小型系统。十、最佳实践清单让爬虫长期可维护1. 尊重规则与边界爬虫不是绕过限制的工具。正式项目应遵守目标网站服务条款、robots.txt、版权规则和访问频率要求。2. 限流优先于重试很多失败来自访问太猛。先降低压力再谈重试。3. 解析逻辑插件化每个站点一个插件每个插件有单元测试。页面结构变化时故障范围可控。4. 日志与指标分层日志回答“发生了什么”指标回答“整体是否健康”。5. 数据落盘使用 JSONLJSONL 一行一条记录适合追加写入、断点恢复和后续批处理。6. 给任务加唯一 IDURL、插件名、时间戳、重试次数都应该进入上下文便于排查问题。7. 测试解析器而不是测试网络单元测试中保存 HTML 样本测试parse()输出是否稳定。deftest_news_plugin_parse():htmla href/a标题 A/apluginNewsPlugin()itemslist(plugin.parse(https://example.com/news,html))assertlen(items)1assertitems[0].title标题 A十一、从入门到高级这套系统背后的 Python 能力这个项目几乎串起了 Python 教程中的核心知识能力在项目中的体现列表、字典、集合URL 队列、状态码统计、插件注册函数与异常请求封装、失败处理类与面向对象Fetcher、Limiter、Plugin、Pipeline装饰器与上下文管理器监控、资源释放生成器解析结果流式产出异步编程高并发网络 I/O模块化设计各组件职责分离单元测试保障插件稳定性这也是 Python 的魅力所在它既能让初学者快速写出第一个程序也能支撑复杂系统的工程化演进。十二、未来扩展从单机爬虫到平台化系统当业务增长后可以继续演进单机队列 - Redis/Kafka 分布式队列 JSONL 文件 - PostgreSQL/Elasticsearch/对象存储 简单 Metrics - Prometheus Grafana 手动运行 - Airflow/Prefect 定时调度 静态插件 - 动态插件加载与版本管理 本地部署 - Docker Kubernetes也可以引入 FastAPI 做任务管理接口或用 Streamlit 做内部监控面板。FastAPI 官方定位是基于标准 Python 类型提示构建现代、高性能 APIStreamlit 文档则强调可以用纯 Python 构建和分享数据应用非常适合内部数据工具与监控看板。(FastAPI)结语高级不是更快而是更稳一个初级爬虫解决“我能不能抓到”一个工程化爬虫解决“我能不能长期、稳定、可控、可追踪地抓到”。Python 高级工程师的价值不在于能不能写出最炫的异步代码而在于能不能把真实世界的不确定性收进系统设计里网络会失败所以有重试目标站有压力所以有限流规则会变化所以有插件线上会出问题所以有监控团队会协作所以有模块边界和测试。这就是 Python 实战最迷人的地方你写的不只是代码而是一套让混乱变得有秩序的能力。