做爬虫这件事最怕的不是写不出来而是跑着跑着就“失联”了。我自己早年做数据采集时踩过一个坑写了一个Python爬虫脚本扔在服务器上每天手动执行一次。一开始还挺顺利后来有一天发现某个字段连续三天都是空值数据表里安静得可怕。因为目标页面默默改了结构爬虫一直在跑、一直在报错但没有任何人知道。也是从那时候开始我意识到一个关键问题爬虫不能只是“能跑”它必须是一个可持续运行、可追溯状态、可生成报告的系统。于是就有了这套“可复现的爬虫运行报告监控系统”的实践。这个系统解决的核心问题有三类第一爬虫“死没死”要有感知不能等业务方来找你才发现第二每次运行的结果要有完整记录出了问题能回看、能复盘第三运行过程必须可复现换机器、改参数、过一个月重启行为模式是可预期的。无论你是刚入门Python爬虫还是已经在用Scrapy跑分布式任务这套思路都可以直接套用。下面我把整个系统的设计思路和核心实现完整拆开讲。1. 为什么爬虫需要一套“可复现监控”的体系很多人的爬虫生涯都是从一次性的临时脚本开始的requests请求、解析、存CSV完事。这种模式在小范围实验时没有问题但一旦涉及定时采集、批量任务、长期运维它就会暴露出大量问题。这一节先把“为什么”讲明白后面再给方案。1.1 临时脚本看着能跑真上线全是坑典型的临时脚本有三个致命伤。第一是“裸奔式请求”。没有统一的请求头管理、没有超时控制、没有重试机制目标网站一个网络抖动整个脚本直接抛异常退出。第二是“结果不可验证”。数据抓下来之后不校验总量、不检查字段完整性跑完就算完事至于有没有漏数据根本不知道。第三是“状态不可观测”。脚本运行到哪一步了、还有多少任务没抓、失败了多少次这些问题在脚本里完全没有答案只能靠print输出碰运气。我自己见过最夸张的一次是同事的爬虫任务因为页面改版连续七天都在“成功返回200”但解析出来的结果是空列表。数据表里新增了七天的空记录业务方看到报表才发现数据不对。这种问题不是靠改代码能解决的而是整个体系缺了“运行报告”和“监控”这两层。所以爬虫一旦进入常态化运行阶段就必须把可观测性纳入设计而不是事后补救。这也是本文整套方案的核心出发点。1.2 可复现意味着什么不是跑通而是可追溯、可重放、可验证一说“可复现”有人会理解成“脚本能重复执行”。这个理解太浅了。真正的可复现包含三个层次。第一个层次是行为可追溯。每次运行任务名、启动时间、结束时间、执行时长、成功数量、失败数量、错误类型分布这些信息必须被记录。出问题时能回答“那次运行到底发生了什么”。第二个层次是过程可重放。给定同一份配置、同一批输入数据、同一个时间窗口爬虫的执行行为应该是一致的。这不是要求抓到的数据永远相同目标网站本来就在变而是要求爬虫自身的请求行为、调度顺序、容错策略是可预期、可重复的。第三个层次是环境可重建。换一台机器按固定步骤装依赖、配环境、放配置就能跑起来。这要求依赖锁定、配置分离、路径无关。三个层次都做到了才叫可复现。下面整个系统的设计我都是围绕这三个层次展开的。1.3 监控到底监控什么六个核心指标监控不是“挂了发个告警”那么简单一套实用的爬虫监控至少要盯住六个指标。任务存活爬虫进程是否在跑心跳是否正常。执行成功率请求成功数占总请求数的比例。数据产出量每个任务实际入库条数是涨是跌。耗时表现单任务总耗时、单请求平均耗时趋势是否劣化。错误分布超时、连接错误、解析错误、HTTP状态码异常各占多少。反爬事件是否触发验证码、封IP、请求被拒绝等异常信号。这六个指标不是均匀用力而是分层处理。任务存活和执行成功率属于“第一优先级”出了问题必须即时告警数据产出量和耗时属于“第二优先级”适合放进日报里观察趋势错误分布和反爬事件属于“诊断信息”只在排查问题时深挖。明确了监控的维度下面就可以开始设计系统了。2. 系统整体架构与模块设计一套可复现的监控系统架构上不需要复杂但分层必须清晰。我的做法是五层拆分调度层、抓取层、存储层、报告层、监控层。每一层职责单一层与层之间通过约定好的接口交互这样后期替换组件或者扩展任务类型都不会牵一发动全身。2.1 模块分层与数据流设计先看整体数据流调度器触发任务 → 抓取层按计划采集 → 数据写入存储层 → 运行结束后报告层汇总执行信息 → 监控层对比历史数据判断是否需要告警。五个模块各有各的关注点。调度层负责“什么时候跑”。我用的是APScheduler支持cron表达式和interval两种触发方式。抓取层负责“怎么拿数据”。这里我封装了一个带重试、超时、UA轮换、代理切换的请求会话对象。存储层负责“数据放哪”我用了SQLite做本地存储轻量、零配置、适合单机任务。报告层负责“跑完怎么说”每次任务结束生成一份JSON格式的运行摘要按日归档。监控层负责“出事了怎么办”它消费报告层的数据结合心跳检查做告警。这五个模块之间的关系是单向依赖的调度层不关心抓取层怎么实现报告层不知道存储层是SQLite还是MySQL。这种松耦合带来的直接好处是未来你要把抓取层从requests换成httpx或者把存储层从SQLite换成PostgreSQL都不需要动其他模块。2.2 技术选型requests封装还是Scrapy关于抓取层用什么我经常被问到。直接说结论如果是定时小规模采集、业务逻辑个性化强、需要深度定制监控报告的场景用requests加自己封装更顺手如果是大规模采集、已具备标准数据管道、不介意框架约束的场景用Scrapy更合理。我这套系统选择了requests自己封装核心原因有三点。第一任务规模不大单机日请求量在几万以内requests完全够用犯不着为Scrapy引入Twisted异步框架的调试成本。第二可复现性要求行为高度可控Scrapy的中间件、调度器、去重逻辑配置项太多出问题排查链路长自己封装每个环节都是自己的代码出问题一眼能定位。第三报告系统和监控系统需要大量自定义埋点自己封装可以随时插入统计逻辑而Scrapy的信号系统虽然也能支持但写法更绕。这不是说Scrapy不好。如果你日均请求量在十万以上有多站点并发需求请直接上Scrapy不必在这套方案里纠结。2.3 数据模型与目录结构数据模型设计是“可复现”的基石。我先给出一份目录结构方便你对照后面的代码。crawler_project/ ├── config/ │ ├── settings.yaml # 全局配置并发、间隔、代理、UA │ └── tasks.yaml # 任务配置每个抓取任务的参数 ├── crawler/ │ ├── __init__.py │ ├── http_client.py # 请求会话封装 │ ├── scheduler.py # 调度器 │ ├── tasks/ # 具体任务逻辑 │ │ ├── base.py │ │ └── demo_task.py │ ├── monitor/ │ │ ├── heartbeat.py # 心跳管理 │ │ ├── report.py # 报告生成 │ │ └── alert.py # 告警通知 │ └── storage/ │ ├── db.py # SQLite操作 │ └── models.py # ORM模型 ├── logs/ # 日志目录 ├── reports/ # 运行报告归档 ├── requirements.txt └── main.py # 程序入口任务表核心字段包括任务ID、任务名、目标URL模板、抓取参数翻页方式、解析规则、请求间隔、超时时间、重试次数。运行状态表记录每次执行运行ID、任务ID、启动时间、结束时间、状态、成功数、失败数、错误摘要。心跳表记录最近一次心跳时间、存活状态。数据表则根据业务来定义字段但必须包含唯一键用于去重。这套模型设计的关键点在于任务配置和运行状态是分离的。配置是“怎么跑”状态是“跑得怎么样”两者解耦后复查问题和调整策略都很方便。3. 核心实现抓取、调度、报告、监控一条链前面是设计这一节是真正能抄作业的代码。我会把一个最小可用的监控系统核心模块完整拆解每个模块给出关键代码和设计理由。3.1 抓取核心一个带“保命”功能的请求会话封装抓取层的核心是对requests.Session做一层封装。裸用requests最大的问题是每次请求都重新建立连接效率低而且异常处理写起来很啰嗦很容易在某些边角抛异常导致整个任务挂掉。我的封装思路是Session负责连接池复用外层包装超时、重试、UA轮换、代理切换和状态上报。import random import time import logging import requests from requests.adapters import HTTPAdapter logger logging.getLogger(crawler.http) class SmartHttpClient: def __init__(self, settings: dict): self.session requests.Session() # 连接池配置池大小和最大重试次数 adapter HTTPAdapter( pool_connectionssettings.get(pool_size, 10), pool_maxsizesettings.get(pool_max, 20), max_retries0, ) self.session.mount(http://, adapter) self.session.mount(https://, adapter) self.user_agents settings.get(user_agents, []) self.proxies settings.get(proxies, []) self.timeout settings.get(timeout, 15) def _random_headers(self): ua random.choice(self.user_agents) if self.user_agents else Mozilla/5.0 return { User-Agent: ua, Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, } def get(self, url, paramsNone, retries3, backoff_factor2, **kwargs): headers kwargs.pop(headers, {}) headers.update(self._random_headers()) proxy random.choice(self.proxies) if self.proxies else None for attempt in range(1, retries 1): try: resp self.session.get( url, paramsparams, headersheaders, timeoutself.timeout, proxies{http: proxy, https: proxy} if proxy else None, **kwargs, ) resp.raise_for_status() return resp except requests.RequestException as exc: logger.warning( 请求失败 url%s attempt%d/%d error%s, url, attempt, retries, exc, ) if attempt retries: sleep_time backoff_factor * (2 ** (attempt - 1)) time.sleep(sleep_time) return None这段代码里有几个细节值得强调。第一个是max_retries0。虽然HTTPAdapter本身支持重试但我在外层自己实现重试逻辑因为外层重试可以配合指数退避、切换代理和状态上报控制力更强。第二个是指数退避的重试间隔。backoff_factor * (2 ** (attempt - 1))表示第一次重试等2秒第二次等4秒第三次等8秒。这种退避策略对缓解目标服务器压力、降低封禁概率非常有效。第三个是每次请求都返回None表示失败调用方需要处理None的情况这是刻意设计的——用None代替抛异常让业务逻辑更简单。3.2 调度层定时任务、互斥锁与异常退出调度层我选APScheduler。它支持cron表达式、支持持久化任务存储、还支持监听任务事件对于爬虫这种定时任务场景非常合适。from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR from crawler.tasks.demo_task import DemoTask def build_scheduler(): scheduler BlockingScheduler() scheduler.add_listener(job_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR) scheduler.add_job( DemoTask().run, CronTrigger(hour2, minute0), # 每天凌晨两点 iddemo_task, name示例采集任务, replace_existingTrue, coalesceTrue, max_instances1, ) return scheduler def job_listener(event): if event.exception: # 任务执行抛异常必须记录并告警 logger.error(任务 %s 执行异常: %s, event.job_id, event.exception) else: logger.info(任务 %s 正常执行完成返回值长度%s, event.job_id, event.retval)调度器最容易踩的坑是max_instances1没设置。如果任务本身跑得久而cron触发间隔又短下一个任务可能会在上一个任务还没结束时就启动导致两个实例同时操作数据库、同时请求目标网站行为完全不可控。max_instances1配合coalesceTrue的意思是上一次没跑完这次触发就跳过如果因为关机错过多次触发恢复后只补跑最近一次。另一个关键点是任务执行必须捕获顶层异常。APScheduler虽然会捕获异常并触发EVENT_JOB_ERROR但如果你的任务内部有自己的循环异常被内部吞掉事件监听是感知不到的。所以我的每个任务类都会在最外层run方法里做兜底def run(self): start_ts datetime.now() try: self._execute() self._write_report(statussuccess) except Exception as exc: logger.exception(任务执行异常) self._write_report(statusfailed, errorstr(exc)) raise finally: write_heartbeat(task_idself.task_id)3.3 运行报告用数据说话给业务方一个交代运行报告是整个系统里最有“说服力”的部分。它回答的关键问题是这批数据到底抓得怎么样能不能信。我的做法是每次任务结束后生成一个JSON摘要文件包含以下信息任务ID、任务名、开始/结束时间、总耗时、请求总数、成功数、失败数、成功率、采集数据量、错误TOP5、反爬事件记录。同时维护一个日汇总报告把当天所有任务的运行数据聚合在一起。import json from datetime import datetime def write_run_report(run_id, task_id, task_name, metrics, output_dirreports): report { run_id: run_id, task_id: task_id, task_name: task_name, timestamp: datetime.now().isoformat(), metrics: metrics, } path f{output_dir}/{task_id}_{run_id}.json with open(path, w, encodingutf-8) as f: json.dump(report, f, ensure_asciiFalse, indent2) return path这里的metrics是任务执行过程中由埋点收集的统计结果。埋点怎么埋我建议用装饰器或者上下文管理器统一处理不在业务代码里到处插统计逻辑。一个简单但有效的做法是用一个StatsCollector类抓取层每次请求成功或失败都调用它记录任务结束时导出class StatsCollector: def __init__(self): self.total 0 self.success 0 self.failed 0 self.errors {} self.start_ts time.time() def record_success(self): self.total 1 self.success 1 def record_failure(self, error_type): self.total 1 self.failed 1 self.errors[error_type] self.errors.get(error_type, 0) 1 def summary(self): return { total: self.total, success: self.success, failed: self.failed, success_rate: round(self.success / max(self.total, 1), 4), errors: self.errors, duration: round(time.time() - self.start_ts, 2), }报告除了给自己看更重要的是给没有技术背景的业务方看。所以我还会用Jinja2模板渲染一个简单的HTML报表把成功率、数据量、耗时做成图表放在服务器的report目录下业务方能直接打开浏览器看。3.4 监控告警心跳机制与多级通知任务存活监控方面我用了最经典也最可靠的方式心跳文件加时间戳。任务进程每30秒往心跳文件里写一次当前时间。监控进程定期检查心跳文件如果当前时间与心跳时间的差值超过阈值就判定任务失联。这种方式的好处是简单、零依赖、不引入额外的消息队列。缺点是实时性一般但在爬虫监控这个场景完全够用。import os import time class Heartbeat: def __init__(self, task_id, pathlogs/heartbeat): self.task_id task_id self.path path os.makedirs(self.path, exist_okTrue) def beat(self): ts time.time() with open(f{self.path}/{self.task_id}.hb, w) as f: f.write(str(ts)) staticmethod def check(task_id, pathlogs/heartbeat, max_gap180): hb_file f{path}/{task_id}.hb if not os.path.exists(hb_file): return False last_ts float(open(hb_file).read().strip()) return time.time() - last_ts max_gap告警我接了企业微信和钉钉的Webhook。这两种方式都是最简单的HTTP POST不需要申请复杂的鉴权SDK配置好一个Webhook地址就能用。告警级别分为三级心跳丢失和任务异常属于紧急成功率低于阈值属于警告数据量异常波动属于提醒。不同级别推送的群组、文案都会做区分。import requests def send_alert(title, content, webhook_url, levelwarning): data { msgtype: text, text: {content: f[{level.upper()}] {title}\n{content}}, } try: requests.post(webhook_url, jsondata, timeout5) except requests.RequestException as exc: logger.error(告警发送失败: %s, exc)4. 可复现性的工程化实践架构和核心模块讲完了接下来要解决的是“换台机器能不能跑起来”“过了一个月还能不能重放当时的结果”这些工程化问题。这一部分没有炫技都是实打实的规范。4.1 配置文件与运行参数分离我见过太多人把UA、数据库路径、请求间隔、代理地址写死在代码里。这带来的问题是换环境必须改代码改动了代码就破坏了运行过程的可复现性。规范做法是把所有可变参数放进YAML配置文件代码里只保留逻辑。以settings.yaml为例http: timeout: 15 pool_size: 10 pool_max: 20 retries: 3 backoff_factor: 2 user_agents: - Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 - Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 proxies: [] storage: db_path: data/crawler.db schedule: timezone: Asia/Shanghai jobs: - id: demo_task cron: 0 2 * * *代码里用yaml.safe_load加载再传给各个模块。任务参数文件tasks.yaml单独维护新增任务不需要改代码只需要加一段配置。这里特别提醒一个坑时区问题。APScheduler默认是本地时区如果你把服务器部署在别的时区环境cron触发时间可能出现偏移。务必在配置里显式指定timezone: Asia/Shanghai。4.2 幂等设计与数据去重可复现的另一个关键是幂等同一个任务跑两次结果不会出现重复数据。这个靠数据表唯一键加插入前的查询判断实现。def upsert_article(conn, article: dict) - bool: cur conn.execute( SELECT 1 FROM articles WHERE url_hash ? LIMIT 1, (article[url_hash],), ) if cur.fetchone(): return False conn.execute( INSERT INTO articles (url_hash, title, content, published_at, crawled_at) VALUES (?, ?, ?, ?, ?), (article[url_hash], article[title], article[content], article[published_at], datetime.now().isoformat()), ) return True这里用url_hash作为业务唯一键而不是自增ID。因为同一篇文章重复抓取时自增ID永远不同查不到重复而URL哈希在内容来源固定时是稳定的。幂等设计的另一个好处是可以安全地重跑失败任务。一个任务执行到一半挂了下一轮调度重新跑已处理过的URL会跳过只补充那些没抓成功的进度自然衔接。这个对监控系统的告警恢复路径非常有价值——修复问题后直接重放任务即可不需要手动清理数据。4.3 日志体系与运行痕迹归档日志体系是运行痕迹的核心。我不建议只用print更不建议长期不开日志轮转让磁盘被写爆。我是这样做的全局logging配置输出到文件和控制台日志文件按天轮转保留最近30天业务日志用结构化前缀行便于grep。import logging from logging.handlers import TimedRotatingFileHandler def setup_logging(log_dirlogs, levellogging.INFO): fmt logging.Formatter( %(asctime)s %(levelname)s [%(name)s] %(message)s ) handler TimedRotatingFileHandler( filenamef{log_dir}/crawler.log, whenmidnight, backupCount30, encodingutf-8, ) handler.setFormatter(fmt) console logging.StreamHandler() console.setFormatter(fmt) root logging.getLogger() root.setLevel(level) root.addHandler(handler) root.addHandler(console)日志和运行报告是两个东西。日志是过程记录运行报告是结果总结。排查问题时我一般先看运行报告里的错误TOP5锁定方向后再去日志里搜具体异常堆栈。为了让这个链路更顺畅我在每次请求失败时都会输出一条包含URL、错误类型、重试次数的结构化日志格式统一为error_type|url|attempt|detail方便grep和后续统计。4.4 定时部署crontab还是systemd项目部署和定时任务注册我有两个推荐方案。小规模场景直接用crontab优点是简单直观。入口是main.py通过参数选择运行单次任务还是启动调度器。# 每天凌晨2点跑一次爬虫任务单次模式 0 2 * * * cd /opt/crawler_project /usr/bin/python3 main.py --task demo_task --once logs/cron.log 21 # 启动常驻调度器监控模式 reboot cd /opt/crawler_project /usr/bin/python3 main.py --scheduler logs/cron.log 21规模大一点、需要考虑进程crash后自动重启的场景我推荐systemd service加timer。systemd最大的优势是进程管理完善开机自启、异常退出自动拉起、崩溃前的标准输出会被journald完整保存。在部署层面还有一个核心动作锁定依赖版本。requirements.txt里不要写不带版本号的包名否则下周装出来的环境和你今天的可能就不一致了可复现性直接被破坏。更稳妥的做法是用pip freeze requirements.lock锁定全量版本。5. 常见问题与排查技巧实录所有系统上线后都会遇到问题。这个监控系统运行至今我积累了不少排查经验挑最有价值的整理在下面。5.1 高频问题速查表问题现象可能原因排查方向心跳正常但数据没更新页面结构变化解析规则失效查看运行报告中的采集量对比历史走势请求大量超时目标网站限流或网络问题检查错误类型的比例超时是否集中在特定URL段成功率骤降IP被限制返回403/503/验证码查看HTTP状态码分布确认是否触发反爬调度任务不触发时区配置错误或进程未常驻检查运行日志中的调度记录核对系统时区日志文件无限增长未配置日志轮转检查log目录大小确认TimedRotatingFileHandler生效重复数据激增唯一键失效或去重逻辑被绕过检查URL hash字段是否稳定是否存在URL跳转导致hash变化告警频繁误报心跳阈值设置过短结合任务时长调整心跳gap阈值一般设180秒以上5.2 排查案例一成功率正常但数据量下降一半有一次我发现某个任务的采集量连续三天从每天2000条降到了1000条但成功率显示99%以上心跳也正常。起初以为目标网站下架了部分内容后来排查才发现原因页面改版后列表页的翻页参数从?page2变成了?page2sortlatest我的分页逻辑没有携带新增参数导致爬虫始终在抓前1000条数据。这个问题的教训是不能只看请求成功率一定要对比数据量的历史趋势。为此我在监控指标里专门加了“采集量环比变化”这一项超过20%的波动就触发提醒。5.3 排查案例二定时任务静默消失另一个典型问题是APScheduler的任务突然不再执行但进程还在跑。排查下来发现是任务执行过程中抛了一个未捕获的自定义异常而事件监听器的代码本身也有问题导致异常被吞掉。APScheduler的任务队列因为多次异常被阻塞后面的触发全部被跳过。修复方案是双重保险第一所有任务的最外层run方法加try...except兜底异常统一进报告第二增加“调度器存活检查”的独立监控任务每5分钟检查任务的下次运行时间是否合理如果调度器队列卡死这个监控会发现下次运行时间不更新并触发告警。6. 监控系统的告警恢复与人工处理机制告警发出去了问题解决了但这套系统还有一个容易忽略的环节告警恢复。很多人只做了告警触发没做恢复通知导致值班人员处理完问题后还要手动在群里说明“已修复”。我的做法是每个监控检查逻辑都附带恢复检测。具体来说心跳文件在任务恢复正常后会重新更新监控进程检测到心跳恢复后自动发送一条“任务已恢复”的确认消息。成功率指标则在任务下一次运行结束后对比报告里的成功率是否回到阈值之上达标后自动发送恢复消息。这看起来是小功能但在真实运维场景里非常省心减少了很多无效沟通。最后再分享一个实用小技巧运行报告里的JSON摘要在归档时记得用日期做目录层级形如reports/2025/06/17/demo_task_20250617_020001.json。这样一个月后要复盘某一天的运行情况直接按目录翻就行不需要写脚本去全量扫文件名。这套监控系统运行了一年多这种目录规整带来的检索效率提升是当初设计时完全没有预料到的额外红利。 做爬虫这件事最怕的不是写不出来而是跑着跑着就“失联”了。我自己早年做数据采集时踩过一个坑写了一个Python爬虫脚本扔在服务器上每天手动执行一次。一开始还挺顺利后来有一天发现某个字段连续三天都是空值数据表里安静得可怕。因为目标页面默默改了结构爬虫一直在跑、一直在报错但没有任何人知道。也是从那时候开始我意识到一个关键问题爬虫不能只是“能跑”它必须是一个可持续运行、可追溯状态、可生成报告的系统。于是就有了这套“可复现的爬虫运行报告监控系统”的实践。这个系统解决的核心问题有三类第一爬虫“死没死”要有感知不能等业务方来找你才发现第二每次运行的结果要有完整记录出了问题能回看、能复盘第三运行过程必须可复现换机器、改参数、过一个月重启行为模式是可预期的。无论你是刚入门Python爬虫还是已经在用Scrapy跑分布式任务这套思路都可以直接套用。下面我把整个系统的设计思路和核心实现完整拆开讲。1. 为什么爬虫需要一套“可复现监控”的体系很多人的爬虫生涯都是从一次性的临时脚本开始的requests请求、解析、存CSV完事。这种模式在小范围实验时没有问题但一旦涉及定时采集、批量任务、长期运维它就会暴露出大量问题。这一节先把“为什么”讲明白后面再给方案。1.1 临时脚本看着能跑真上线全是坑典型的临时脚本有三个致命伤。第一是“裸奔式请求”。没有统一的请求头管理、没有超时控制、没有重试机制目标网站一个网络抖动整个脚本直接抛异常退出。第二是“结果不可验证”。数据抓下来之后不校验总量、不检查字段完整性跑完就算完事至于有没有漏数据根本不知道。第三是“状态不可观测”。脚本运行到哪一步了、还有多少任务没抓、失败了多少次这些问题在脚本里完全没有答案只能靠print输出碰运气。我自己见过最夸张的一次是同事的爬虫任务因为页面改版连续七天都在“成功返回200”但解析出来的结果是空列表。数据表里新增了七天的空记录业务方看到报表才发现数据不对。这种问题不是靠改代码能解决的而是整个体系缺了“运行报告”和“监控”这两层。所以爬虫一旦进入常态化运行阶段就必须把可观测性纳入设计而不是事后补救。这也是本文整套方案的核心出发点。1.2 可复现意味着什么不是跑通而是可追溯、可重放、可验证一说“可复现”有人会理解成“脚本能重复执行”。这个理解太浅了。真正的可复现包含三个层次。第一个层次是行为可追溯。每次运行任务名、启动时间、结束时间、执行时长、成功数量、失败数量、错误类型分布这些信息必须被记录。出问题时能回答“那次运行到底发生了什么”。第二个层次是过程可重放。给定同一份配置、同一批输入数据、同一个时间窗口爬虫的执行行为应该是一致的。这不是要求抓到的数据永远相同目标网站本来就在变而是要求爬虫自身的请求行为、调度顺序、容错策略是可预期、可重复的。第三个层次是环境可重建。换一台机器按固定步骤装依赖、配环境、放配置就能跑起来。这要求依赖锁定、配置分离、路径无关。三个层次都做到了才叫可复现。下面整个系统的设计我都是围绕这三个层次展开的。1.3 监控到底监控什么六个核心指标监控不是“挂了发个告警”那么简单一套实用的爬虫监控至少要盯住六个指标。任务存活爬虫进程是否在跑心跳是否正常。执行成功率请求成功数占总请求数的比例。数据产出量每个任务实际入库条数是涨是跌。耗时表现单任务总耗时、单请求平均耗时趋势是否劣化。错误分布超时、连接错误、解析错误、HTTP状态码异常各占多少。反爬事件是否触发验证码、封IP、请求被拒绝等异常信号。这六个指标不是均匀用力而是分层处理。任务存活和执行成功率属于“第一优先级”出了问题必须即时告警数据产出量和耗时属于“第二优先级”适合放进日报里观察趋势错误分布和反爬事件属于“诊断信息”只在排查问题时深挖。明确了监控的维度下面就可以开始设计系统了。2. 系统整体架构与模块设计一套可复现的监控系统架构上不需要复杂但分层必须清晰。我的做法是五层拆分调度层、抓取层、存储层、报告层、监控层。每一层职责单一层与层之间通过约定好的接口交互这样后期替换组件或者扩展任务类型都不会牵一发动全身。2.1 模块分层与数据流设计先看整体数据流调度器触发任务 → 抓取层按计划采集 → 数据写入存储层 → 运行结束后报告层汇总执行信息 → 监控层对比历史数据判断是否需要告警。五个模块各有各的关注点。调度层负责“什么时候跑”。我用的是APScheduler支持cron表达式和interval两种触发方式。抓取层负责“怎么拿数据”。这里我封装了一个带重试、超时、UA轮换、代理切换的请求会话对象。存储层负责“数据放哪”我用了SQLite做本地存储轻量、零配置、适合单机任务。报告层负责“跑完怎么说”每次任务结束生成一份JSON格式的运行摘要按日归档。监控层负责“出事了怎么办”它消费报告层的数据结合心跳检查做告警。这五个模块之间的关系是单向依赖的调度层不关心抓取层怎么实现报告层不知道存储层是SQLite还是MySQL。这种松耦合带来的直接好处是未来你要把抓取层从requests换成httpx或者把存储层从SQLite换成PostgreSQL都不需要动其他模块。2.2 技术选型requests封装还是Scrapy关于抓取层用什么我经常被问到。直接说结论如果是定时小规模采集、业务逻辑个性化强、需要深度定制监控报告的场景用requests加自己封装更顺手如果是大规模采集、已具备标准数据管道、不介意框架约束的场景用Scrapy更合理。我这套系统选择了requests自己封装核心原因有三点。第一任务规模不大单机日请求量在几万以内requests完全够用犯不着为Scrapy引入Twisted异步框架的调试成本。第二可复现性要求行为高度可控Scrapy的中间件、调度器、去重逻辑配置项太多出问题排查链路长自己封装每个环节都是自己的代码出问题一眼能定位。第三报告系统和监控系统需要大量自定义埋点自己封装可以随时插入统计逻辑而Scrapy的信号系统虽然也能支持但写法更绕。这不是说Scrapy不好。如果你日均请求量在十万以上有多站点并发需求请直接上Scrapy不必在这套方案里纠结。2.3 数据模型与目录结构数据模型设计是“可复现”的基石。我先给出一份目录结构方便你对照后面的代码。crawler_project/ ├── config/ │ ├── settings.yaml # 全局配置并发、间隔、代理、UA │ └── tasks.yaml # 任务配置每个抓取任务的参数 ├── crawler/ │ ├── __init__.py │ ├── http_client.py # 请求会话封装 │ ├── scheduler.py # 调度器 │ ├── tasks/ # 具体任务逻辑 │ │ ├── base.py │ │ └── demo_task.py │ ├── monitor/ │ │ ├── heartbeat.py # 心跳管理 │ │ ├── report.py # 报告生成 │ │ └── alert.py # 告警通知 │ └── storage/ │ ├── db.py # SQLite操作 │ └── models.py # ORM模型 ├── logs/ # 日志目录 ├── reports/ # 运行报告归档 ├── requirements.txt └── main.py # 程序入口任务表核心字段包括任务ID、任务名、目标URL模板、抓取参数翻页方式、解析规则、请求间隔、超时时间、重试次数。运行状态表记录每次执行运行ID、任务ID、启动时间、结束时间、状态、成功数、失败数、错误摘要。心跳表记录最近一次心跳时间、存活状态。数据表则根据业务来定义字段但必须包含唯一键用于去重。这套模型设计的关键点在于任务配置和运行状态是分离的。配置是“怎么跑”状态是“跑得怎么样”两者解耦后复查问题和调整策略都很方便。3. 核心实现抓取、调度、报告、监控一条链前面是设计这一节是真正能抄作业的代码。我会把一个最小可用的监控系统核心模块完整拆解每个模块给出关键代码和设计理由。3.1 抓取核心一个带“保命”功能的请求会话封装抓取层的核心是对requests.Session做一层封装。裸用requests最大的问题是每次请求都重新建立连接效率低而且异常处理写起来很啰嗦很容易在某些边角抛异常导致整个任务挂掉。我的封装思路是Session负责连接池复用外层包装超时、重试、UA轮换、代理切换和状态上报。import random import time import logging import requests from requests.adapters import HTTPAdapter logger logging.getLogger(crawler.http) class SmartHttpClient: def __init__(self, settings: dict): self.session requests.Session() # 连接池配置池大小和最大重试次数 adapter HTTPAdapter( pool_connectionssettings.get(pool_size, 10), pool_maxsizesettings.get(pool_max, 20), max_retries0, ) self.session.mount(http://, adapter) self.session.mount(https://, adapter) self.user_agents settings.get(user_agents, []) self.proxies settings.get(proxies, []) self.timeout settings.get(timeout, 15) def _random_headers(self): ua random.choice(self.user_agents) if self.user_agents else Mozilla/5.0 return { User-Agent: ua, Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9,en;q0.8, } def get(self, url, paramsNone, retries3, backoff_factor2, **kwargs): headers kwargs.pop(headers, {}) headers.update(self._random_headers()) proxy random.choice(self.proxies) if self.proxies else None for attempt in range(1, retries 1): try: resp self.session.get( url, paramsparams, headersheaders, timeoutself.timeout, proxies{http: proxy, https: proxy} if proxy else None, **kwargs, ) resp.raise_for_status() return resp except requests.RequestException as exc: logger.warning( 请求失败 url%s attempt%d/%d error%s, url, attempt, retries, exc, ) if attempt retries: sleep_time backoff_factor * (2 ** (attempt - 1)) time.sleep(sleep_time) return None这段代码里有几个细节值得强调。第一个是max_retries0。虽然HTTPAdapter本身支持重试但我在外层自己实现重试逻辑因为外层重试可以配合指数退避、切换代理和状态上报控制力更强。第二个是指数退避的重试间隔。backoff_factor * (2 ** (attempt - 1))表示第一次重试等2秒第二次等4秒第三次等8秒。这种退避策略对缓解目标服务器压力、降低封禁概率非常有效。第三个是每次请求都返回None表示失败调用方需要处理None的情况这是刻意设计的——用None代替抛异常让业务逻辑更简单。3.2 调度层定时任务、互斥锁与异常退出调度层我选APScheduler。它支持cron表达式、支持持久化任务存储、还支持监听任务事件对于爬虫这种定时任务场景非常合适。from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR from crawler.tasks.demo_task import DemoTask def build_scheduler(): scheduler BlockingScheduler() scheduler.add_listener(job_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR) scheduler.add_job( DemoTask().run, CronTrigger(hour2, minute0), # 每天凌晨两点 iddemo_task, name示例采集任务, replace_existingTrue, coalesceTrue, max_instances1, ) return scheduler def job_listener(event): if event.exception: # 任务执行抛异常必须记录并告警 logger.error(任务 %s 执行异常: %s, event.job_id, event.exception) else: logger.info(任务 %s 正常执行完成返回值长度%s, event.job_id, event.retval)调度器最容易踩的坑是max_instances1没设置。如果任务本身跑得久而cron触发间隔又短下一个任务可能会在上一个任务还没结束时就启动导致两个实例同时操作数据库、同时请求目标网站行为完全不可控。max_instances1配合coalesceTrue的意思是上一次没跑完这次触发就跳过如果因为关机错过多次触发恢复后只补跑最近一次。另一个关键点是任务执行必须捕获顶层异常。APScheduler虽然会捕获异常并触发EVENT_JOB_ERROR但如果你的任务内部有自己的循环异常被内部吞掉事件监听是感知不到的。所以我的每个任务类都会在最外层run方法里做兜底def run(self): start_ts datetime.now() try: self._execute() self._write_report(statussuccess) except Exception as exc: logger.exception(任务执行异常) self._write_report(statusfailed, errorstr(exc)) raise finally: write_heartbeat(task_idself.task_id)3.3 运行报告用数据说话给业务方一个交代运行报告是整个系统里最有“说服力”的部分。它回答的关键问题是这批数据到底抓得怎么样能不能信。我的做法是每次任务结束后生成一个JSON摘要文件包含以下信息任务ID、任务名、开始/结束时间、总耗时、请求总数、成功数、失败数、成功率、采集数据量、错误TOP5、反爬事件记录。同时维护一个日汇总报告把当天所有任务的运行数据聚合在一起。import json from datetime import datetime def write_run_report(run_id, task_id, task_name, metrics, output_dirreports): report { run_id: run_id, task_id: task_id, task_name: task_name, timestamp: datetime.now().isoformat(), metrics: metrics, } path f{output_dir}/{task_id}_{run_id}.json with open(path, w, encodingutf-8) as f: json.dump(report, f, ensure_asciiFalse, indent2) return path这里的metrics是任务执行过程中由埋点收集的统计结果。埋点怎么埋我建议用装饰器或者上下文管理器统一处理不在业务代码里到处插统计逻辑。一个简单但有效的做法是用一个StatsCollector类抓取层每次请求成功或失败都调用它记录任务结束时导出class StatsCollector: def __init__(self): self.total 0 self.success 0 self.failed 0 self.errors {} self.start_ts time.time() def record_success(self): self.total 1 self.success 1 def record_failure(self, error_type): self.total 1 self.failed 1 self.errors[error_type] self.errors.get(error_type, 0) 1 def summary(self): return { total: self.total, success: self.success, failed: self.failed, success_rate: round(self.success / max(self.total, 1), 4), errors: self.errors, duration: round(time.time() - self.start_ts, 2), }报告除了给自己看更重要的是给没有技术背景的业务方看。所以我还会用Jinja2模板渲染一个简单的HTML报表把成功率、数据量、耗时做成图表放在服务器的report目录下业务方能直接打开浏览器看。3.4 监控告警心跳机制与多级通知任务存活监控方面我用了最经典也最可靠的方式心跳文件加时间戳。任务进程每30秒往心跳文件里写一次当前时间。监控进程定期检查心跳文件如果当前时间与心跳时间的差值超过阈值就判定任务失联。这种方式的好处是简单、零依赖、不引入额外的消息队列。缺点是实时性一般但在爬虫监控这个场景完全够用。import os import time class Heartbeat: def __init__(self, task_id, pathlogs/heartbeat): self.task_id task_id self.path path os.makedirs(self.path, exist_okTrue) def beat(self): ts time.time() with open(f{self.path}/{self.task_id}.hb, w) as f: f.write(str(ts)) staticmethod def check(task_id, pathlogs/heartbeat, max_gap180): hb_file f{path}/{task_id}.hb if not os.path.exists(hb_file): return False last_ts float(open(hb_file).read().strip()) return time.time() - last_ts max_gap告警我接了企业微信和钉钉的Webhook。这两种方式都是最简单的HTTP POST不需要申请复杂的鉴权SDK配置好一个Webhook地址就能用。告警级别分为三级心跳丢失和任务异常属于紧急成功率低于阈值属于警告数据量异常波动属于提醒。不同级别推送的群组、文案都会做区分。import requests def send_alert(title, content, webhook_url, levelwarning): data { msgtype: text, text: {content: f[{level.upper()}] {title}\n{content}}, } try: requests.post(webhook_url, jsondata, timeout5) except requests.RequestException as exc: logger.error(告警发送失败: %s, exc)4. 可复现性的工程化实践架构和核心模块讲完了接下来要解决的是“换台机器能不能跑起来”“过了一个月还能不能重放当时的结果”这些工程化问题。这一部分没有炫技都是实打实的规范。4.1 配置文件与运行参数分离我见过太多人把UA、数据库路径、请求间隔、代理地址写死在代码里。这带来的问题是换环境必须改代码改动了代码就破坏了运行过程的可复现性。规范做法是把所有可变参数放进YAML配置文件代码里只保留逻辑。以settings.yaml为例http: timeout: 15 pool_size: 10 pool_max: 20 retries: 3 backoff_factor: 2 user_agents: - Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 - Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 proxies: [] storage: db_path: data/crawler.db schedule: timezone: Asia/Shanghai jobs: - id: demo_task cron: 0 2 * * *代码里用yaml.safe_load加载再传给各个模块。任务参数文件tasks.yaml单独维护新增任务不需要改代码只需要加一段配置。这里特别提醒一个坑时区问题。APScheduler默认是本地时区如果你把服务器部署在别的时区环境cron触发时间可能出现偏移。务必在配置里显式指定timezone: Asia/Shanghai。4.2 幂等设计与数据去重可复现的另一个关键是幂等同一个任务跑两次结果不会出现重复数据。这个靠数据表唯一键加插入前的查询判断实现。def upsert_article(conn, article: dict) - bool: cur conn.execute( SELECT 1 FROM articles WHERE url_hash ? LIMIT 1, (article[url_hash],), ) if cur.fetchone(): return False conn.execute( INSERT INTO articles (url_hash, title, content, published_at, crawled_at) VALUES (?, ?, ?, ?, ?), (article[url_hash], article[title], article[content], article[published_at], datetime.now().isoformat()), ) return True这里用url_hash作为业务唯一键而不是自增ID。因为同一篇文章重复抓取时自增ID永远不同查不到重复而URL哈希在内容来源固定时是稳定的。幂等设计的另一个好处是可以安全地重跑失败任务。一个任务执行到一半挂了下一轮调度重新跑已处理过的URL会跳过只补充那些没抓成功的进度自然衔接。这个对监控系统的告警恢复路径非常有价值——修复问题后直接重放任务即可不需要手动清理数据。4.3 日志体系与运行痕迹归档日志体系是运行痕迹的核心。我不建议只用print更不建议长期不开日志轮转让磁盘被写爆。我是这样做的全局logging配置输出到文件和控制台日志文件按天轮转保留最近30天业务日志用结构化前缀行便于grep。import logging from logging.handlers import TimedRotatingFileHandler def setup_logging(log_dirlogs, levellogging.INFO): fmt logging.Formatter( %(asctime)s %(levelname)s [%(name)s] %(message)s ) handler TimedRotatingFileHandler( filenamef{log_dir}/crawler.log, whenmidnight, backupCount30, encodingutf-8, ) handler.setFormatter(fmt) console logging.StreamHandler() console.setFormatter(fmt) root logging.getLogger() root.setLevel(level) root.addHandler(handler) root.addHandler(console)日志和运行报告是两个东西。日志是过程记录运行报告是结果总结。排查问题时我一般先看运行报告里的错误TOP5锁定方向后再去日志里搜具体异常堆栈。为了让这个链路更顺畅我在每次请求失败时都会输出一条包含URL、错误类型、重试次数的结构化日志格式统一为error_type|url|attempt|detail方便grep和后续统计。4.4 定时部署crontab还是systemd项目部署和定时任务注册我有两个推荐方案。小规模场景直接用crontab优点是简单直观。入口是main.py通过参数选择运行单次任务还是启动调度器。# 每天凌晨2点跑一次爬虫任务单次模式 0 2 * * * cd /opt/crawler_project /usr/bin/python3 main.py --task demo_task --once logs/cron.log 21 # 启动常驻调度器监控模式 reboot cd /opt/crawler_project /usr/bin/python3 main.py --scheduler logs/cron.log 21规模大一点、需要考虑进程crash后自动重启的场景我推荐systemd service加timer。systemd最大的优势是进程管理完善开机自启、异常退出自动拉起、崩溃前的标准输出会被journald完整保存。在部署层面还有一个核心动作锁定依赖版本。requirements.txt里不要写不带版本号的包名否则下周装出来的环境和你今天的可能就不一致了可复现性直接被破坏。更稳妥的做法是用pip freeze requirements.lock锁定全量版本。5. 常见问题与排查技巧实录所有系统上线后都会遇到问题。这个监控系统运行至今我积累了不少排查经验挑最有价值的整理在下面。5.1 高频问题速查表问题现象可能原因排查方向心跳正常但数据没更新页面结构变化解析规则失效查看运行报告中的采集量对比历史走势请求大量超时目标网站限流或网络问题检查错误类型的比例超时是否集中在特定URL段成功率骤降IP被限制返回403/503/验证码查看HTTP状态码分布确认是否触发反爬调度任务不触发时区配置错误或进程未常驻检查运行日志中的调度记录核对系统时区日志文件无限增长未配置日志轮转检查log目录大小确认TimedRotatingFileHandler生效重复数据激增唯一键失效或去重逻辑被绕过检查URL hash字段是否稳定是否存在URL跳转导致hash变化告警频繁误报心跳阈值设置过短结合任务时长调整心跳gap阈值一般设180秒以上5.2 排查案例一成功率正常但数据量下降一半有一次我发现某个任务的采集量连续三天从每天2000条降到了1000条但成功率显示99%以上心跳也正常。起初以为目标网站下架了部分内容后来排查才发现原因页面改版后列表页的翻页参数从?page2变成了?page2sortlatest我的分页逻辑没有携带新增参数导致爬虫始终在抓前1000条数据。这个问题的教训是不能只看请求成功率一定要对比数据量的历史趋势。为此我在监控指标里专门加了“采集量环比变化”这一项超过20%的波动就触发提醒。5.3 排查案例二定时任务静默消失另一个典型问题是APScheduler的任务突然不再执行但进程还在跑。排查下来发现是任务执行过程中抛了一个未捕获的自定义异常而事件监听器的代码本身也有问题导致异常被吞掉。APScheduler的任务队列因为多次异常被阻塞后面的触发全部被跳过。修复方案是双重保险第一所有任务的最外层run方法加try...except兜底异常统一进报告第二增加“调度器存活检查”的独立监控任务每5分钟检查任务的下次运行时间是否合理如果调度器队列卡死这个监控会发现下次运行时间不更新并触发告警。6. 监控系统的告警恢复与人工处理机制告警发出去了问题解决了但这套系统还有一个容易忽略的环节告警恢复。很多人只做了告警触发没做恢复通知导致值班人员处理完问题后还要手动在群里说明“已修复”。我的做法是每个监控检查逻辑都附带恢复检测。具体来说心跳文件在任务恢复正常后会重新更新监控进程检测到心跳恢复后自动发送一条“任务已恢复”的确认消息。成功率指标则在任务下一次运行结束后对比报告里的成功率是否回到阈值之上达标后自动发送恢复消息。这看起来是小功能但在真实运维场景里非常省心减少了很多无效沟通。最后再分享一个实用小技巧运行报告里的JSON摘要在归档时记得用日期做目录层级形如reports/2025/06/17/demo_task_20250617_020001.json。这样一个月后要复盘某一天的运行情况直接按目录翻就行不需要写脚本去全量扫文件名。这套监控系统运行了一年多这种目录规整带来的检索效率提升是当初设计时完全没有预料到的额外红利。